From cbda65611573a95a3866cd035451f369631aa42d Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Feb 19 2024 12:59:48 +0000 Subject: [PATCH 1/4] better ordering support in QueryView --- diff --git a/kojihub/db.py b/kojihub/db.py index a7cd1cf..40c121d 100644 --- a/kojihub/db.py +++ b/kojihub/db.py @@ -617,13 +617,15 @@ class QueryProcessor(object): group: a column or alias name to use in the 'GROUP BY' clause (controlled by enable_group) - enable_group: if True, opts.group will be enabled + - order_map: (optional) a name:expression map of allowed orders. Otherwise any column or alias + is allowed """ iterchunksize = 1000 def __init__(self, columns=None, aliases=None, tables=None, joins=None, clauses=None, values=None, transform=None, - opts=None, enable_group=False): + opts=None, enable_group=False, order_map=None): self.columns = columns self.aliases = aliases if columns and aliases: @@ -656,6 +658,7 @@ class QueryProcessor(object): self.opts = opts else: self.opts = {} + self.order_map = order_map self.enable_group = enable_group self.logger = logging.getLogger('koji.db') @@ -754,14 +757,19 @@ SELECT %(col_str)s else: direction = '' # Check if we're ordering by alias first - orderCol = self.colsByAlias.get(order) - if orderCol: - pass - elif order in self.columns: - orderCol = order + if self.order_map is not None: + # order should only be a key in the map + expr = self.order_map.get(order) + if not expr: + raise koji.ParameterError(f'Invalid order term: {order}') else: - raise Exception('Invalid order: ' + order) - order_exprs.append(orderCol + direction) + expr = self.colsByAlias.get(order) + if not expr: + if order in self.columns: + expr = order + else: + raise Exception('Invalid order: ' + order) + order_exprs.append(expr + direction) return 'ORDER BY ' + ', '.join(order_exprs) else: return '' @@ -881,18 +889,37 @@ class QueryView: default_fields = () def __init__(self, clauses=None, fields=None, opts=None): + self.clauses = clauses + self.fields = fields + self.opts = opts + self._query = None + + @property + def query(self): + if self._query is not None: + return self._query + else: + return self.get_query() + + def get_query(self): self.extra_joins = [] self.values = {} + self.order_map = {} + + self.check_opts() + tables = list(self.tables) # copy - fields = self.get_fields(fields) - fields, aliases = zip(*fields.items()) - clauses = self.get_clauses(clauses) + fields = self.get_fields(self.fields) + columns, aliases = zip(*fields.items()) + clauses = self.get_clauses() joins = self.get_joins() - self.query = QueryProcessor( - columns=fields, aliases=aliases, + self._query = QueryProcessor( + columns=columns, aliases=aliases, tables=tables, joins=joins, clauses=clauses, values=self.values, - opts=opts) + opts=self.opts, order_map=self.order_map) + + return self._query def get_fields(self, fields): fields = fields or self.default_fields @@ -901,6 +928,19 @@ class QueryView: return {self.map_field(f): f for f in fields} + def check_opts(self): + # some options may trigger joins + if self.opts is None: + return + if 'order' in self.opts: + for key in self.opts['order'].split(','): + if key.startswith('-'): + key = key[1:] + self.order_map[key] = self.map_field(key) + if 'group' in self.opts: + for key in self.opts['group'].split(','): + self.map_field(key) + def map_field(self, field): f_info = self.fieldmap.get(field) if f_info is None: @@ -912,10 +952,10 @@ class QueryView: # duplicates removed later return fullname - def get_clauses(self, clauses): + def get_clauses(self): # for now, just a very simple implementation result = [] - clauses = clauses or [] + clauses = self.clauses or [] for n, clause in enumerate(clauses): # TODO checks check checks if len(clause) == 2: From e1f13ad9f885163989994958c61fa61b8767031e Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Feb 19 2024 13:00:34 +0000 Subject: [PATCH 2/4] order log messages --- diff --git a/kojihub/scheduler.py b/kojihub/scheduler.py index 961ef39..fce8ff6 100644 --- a/kojihub/scheduler.py +++ b/kojihub/scheduler.py @@ -53,7 +53,7 @@ class LogMessagesQuery(QueryView): def get_log_messages(clauses=None, fields=None): - return LogMessagesQuery(clauses, fields).execute() + return LogMessagesQuery(clauses, fields, opts={'order': 'id'}).execute() def get_tasks_for_host(hostID, retry=True): From b803e76997dd1927cef80ce25e7f41b09315d67f Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Feb 19 2024 13:01:41 +0000 Subject: [PATCH 3/4] fakehub: better kw arg handling --- diff --git a/devtools/fakehub b/devtools/fakehub index 4e7920d..cc475bc 100755 --- a/devtools/fakehub +++ b/devtools/fakehub @@ -113,10 +113,12 @@ def get_options(): for s in args[1:]: if '=' in s: k, v = s.split('=', 1) - v = nice_literal(v) - kwargs[k] = v - else: - callargs.append(nice_literal(s)) + if k.isidentifier(): + v = nice_literal(v) + kwargs[k] = v + continue + # else + callargs.append(nice_literal(s)) callargs = koji.encode_args(*callargs, **kwargs) opts.request = koji.xmlrpcplus.dumps(callargs, method, allow_none=1) From b9959999c4d70797369e435b01f04ce3829e2273 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Feb 19 2024 13:02:33 +0000 Subject: [PATCH 4/4] more unit tests --- diff --git a/tests/test_hub/test_query_view.py b/tests/test_hub/test_query_view.py new file mode 100644 index 0000000..1ed3117 --- /dev/null +++ b/tests/test_hub/test_query_view.py @@ -0,0 +1,66 @@ +import mock +import unittest + +import koji +import kojihub.db +import kojihub.scheduler + + +class TestQueryView(unittest.TestCase): + def setUp(self): + # using a convenient view from scheduler + self.viewclass = kojihub.scheduler.TaskRefusalsQuery + + def tearDown(self): + mock.patch.stopall() + + def test_no_joins_needed(self): + view = self.viewclass(fields=['id', 'task_id']) + self.assertEqual(set(view.query.aliases), set(['id', 'task_id'])) + self.assertEqual(view.query.joins, []) + self.assertEqual(view.query.clauses, []) + + def test_one_join_needed(self): + # the additional fields require joining task table + view = self.viewclass(fields=['id', 'task_id', 'method', 'state']) + self.assertEqual(set(view.query.aliases), set(['id', 'task_id', 'method', 'state'])) + self.assertEqual(view.query.joins, ['task ON scheduler_task_refusals.task_id = task.id']) + self.assertEqual(view.query.clauses, []) + + def test_implicit_equal(self): + view = self.viewclass(fields=['id', 'task_id'], clauses=[['id', 23]]) + self.assertEqual(view.query.values, {'v_id_0': 23}) + self.assertEqual(view.query.clauses, ['scheduler_task_refusals.id = %(v_id_0)s']) + + def test_implicit_in(self): + view = self.viewclass(fields=['id', 'task_id'], clauses=[['id', [42, 137]]]) + self.assertEqual(view.query.values, {'v_id_0': [42, 137]}) + self.assertEqual(view.query.clauses, ['scheduler_task_refusals.id IN %(v_id_0)s']) + + def test_explicit_op(self): + view = self.viewclass(fields=['id', 'task_id'], clauses=[['id', '<', 5]]) + self.assertEqual(view.query.values, {'v_id_0': 5}) + self.assertEqual(view.query.clauses, ['scheduler_task_refusals.id < %(v_id_0)s']) + + def test_invalid_op(self): + with self.assertRaises(koji.ParameterError) as e: + view = self.viewclass(fields=['id', 'task_id'], clauses=[['id', '==', 5]]) + view.get_query() + + def test_invalid_clause(self): + with self.assertRaises(koji.ParameterError) as e: + view = self.viewclass(fields=['id', 'task_id'], clauses=[['id', 'NOT', 'EQUAL', 5]]) + view.get_query() + + def test_invalid_field(self): + with self.assertRaises(koji.ParameterError) as e: + view = self.viewclass(fields=['id', 'task_id', 'nosuchfield']) + view.get_query() + + def test_default_fields(self): + view = self.viewclass() + self.assertEqual(set(view.query.aliases), set(self.viewclass.default_fields)) + + def test_all_fields(self): + view = self.viewclass(fields='*') + self.assertEqual(set(view.query.aliases), set(self.viewclass.fieldmap.keys())) diff --git a/tests/test_hub/test_scheduler.py b/tests/test_hub/test_scheduler.py new file mode 100644 index 0000000..3ad6464 --- /dev/null +++ b/tests/test_hub/test_scheduler.py @@ -0,0 +1,220 @@ +import datetime +import mock +import unittest + +import koji +import kojihub +import kojihub.db +from kojihub import scheduler + + +QP = scheduler.QueryProcessor +IP = scheduler.InsertProcessor +UP = scheduler.UpdateProcessor +TASK = kojihub.Task + + +class MyError(Exception): + pass + + +class BaseTest(unittest.TestCase): + + def setUp(self): + self.context = mock.patch('kojihub.scheduler.context').start() + self.context.opts = { + # duplicating hub defaults + 'MaxJobs': 15, + 'CapacityOvercommit':5, + 'ReadyTimeout': 180, + 'AssignTimeout': 300, + 'SoftRefusalTimeout': 900, + 'HostTimeout': 900, + 'RunInterval': 60, + } + + self.db_lock = mock.patch('kojihub.scheduler.db_lock').start() + self.db_lock.return_value = True + + self.QueryProcessor = mock.patch('kojihub.scheduler.QueryProcessor', + side_effect=self.getQuery).start() + self.queries = [] + self.InsertProcessor = mock.patch('kojihub.scheduler.InsertProcessor', + side_effect=self.getInsert).start() + self.inserts = [] + self.UpdateProcessor = mock.patch('kojihub.scheduler.UpdateProcessor', + side_effect=self.getUpdate).start() + self.updates = [] + self._dml = mock.patch('kojihub.db._dml').start() + self.exports = kojihub.RootExports() + self.get_tag = mock.patch('kojihub.kojihub.get_tag').start() + self.query_executeOne = mock.MagicMock() + + self.get_task_refusals = mock.patch('kojihub.scheduler.get_task_refusals').start() + self.get_task_runs = mock.patch('kojihub.scheduler.get_task_runs').start() + + def tearDown(self): + mock.patch.stopall() + + def getQuery(self, *args, **kwargs): + query = QP(*args, **kwargs) + query.execute = mock.MagicMock() + query.executeOne = self.query_executeOne + self.queries.append(query) + return query + + def getInsert(self, *args, **kwargs): + insert = IP(*args, **kwargs) + insert.execute = mock.MagicMock() + self.inserts.append(insert) + return insert + + def getUpdate(self, *args, **kwargs): + update = UP(*args, **kwargs) + update.execute = mock.MagicMock() + self.updates.append(update) + return update + + +class TestLogging(BaseTest): + + def test_log_both(self): + msg = 'Does logging work?' + scheduler.log_both(msg, host_id=1, task_id=2) + self.assertEqual(len(self.inserts), 1) + expected = {'msg': msg, 'host_id': 1, 'task_id': 2} + self.assertEqual(self.inserts[0].data, expected) + + +class TestScheduler(BaseTest): + + def setUp(self): + super(TestScheduler, self).setUp() + + def test_ran_recently(self): + s = scheduler.TaskScheduler() + # scheduler should not run if check_ts says not to + s.check_ts = mock.MagicMock(return_value=False) + s.get_tasks = mock.MagicMock(side_effect=MyError('should not reach unless forced')) + s.run() + self.assertEqual(len(self.queries), 0) + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.updates), 0) + # ... unless we use force + with self.assertRaises(MyError): + s.run(force=True) + + def test_no_lock(self): + self.db_lock.return_value = False + s = scheduler.TaskScheduler() + s.get_tasks = mock.MagicMock(side_effect=MyError('should not reach')) + s.run() + self.assertEqual(len(self.queries), 0) + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.updates), 0) + + def test_run(self): + s = scheduler.TaskScheduler() + s.check_ts = mock.MagicMock(return_value=True) + s.run() + # TODO + + +class TestCheckActiveRuns(BaseTest): + + def setUp(self): + super(TestCheckActiveRuns, self).setUp() + self.sched = scheduler.TaskScheduler() + + self.get_active_runs = mock.MagicMock() + self.sched.get_active_runs = self.get_active_runs + + self.frees = [] + self.assigns = [] + def my_free(task): + self.frees.append(task.id) + def my_assign(task, host_id, force=False): + self.assigns.append((task.id, host_id, force)) + mock.patch('kojihub.Task.free', new=my_free).start() + mock.patch('kojihub.Task.assign', new=my_assign).start() + self.log_db = mock.MagicMock() + mock.patch('kojihub.scheduler.log_db', new=self.log_db).start() + + def test_check_no_active(self): + self.assertEqual(self.sched.active_tasks, []) # set by init + self.sched.check_active_tasks() + # with no active tasks, we shouldn't have done much + self.get_active_runs.assert_called_once() + self.assertEqual(self.frees, []) + self.assertEqual(self.assigns, []) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'scheduler_task_runs') + + def test_check_no_host(self): + # 'Active task with no host' case + self.sched.active_tasks = [{'task_id': 99, 'host_id': None}] + self.sched.check_active_tasks() + self.log_db.assert_called_once_with('Active task with no host', 99, None) + self.get_active_runs.assert_called_once() + self.assertEqual(self.frees, [99]) + self.assertEqual(self.assigns, []) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'scheduler_task_runs') + + def test_check_override(self): + # 'Override task assignment' case + self.sched.active_tasks = [{'task_id': 99, 'host_id': 23, 'state': koji.TASK_STATES['ASSIGNED']}] + self.sched.hosts = {23: {'id': 23, 'name': 'test host 23'}} + self.sched.get_active_runs.return_value = {} + with mock.patch('kojihub.scheduler.logger') as _logger: + self.sched.check_active_tasks() + _logger.debug.assert_called_once_with('Override task assignment: task %i, host %s', + 99, 'test host 23') + self.get_active_runs.assert_called_once() + # this case is not logged to the db + self.log_db.assert_not_called() + # we should not free such tasks + self.assertEqual(self.frees, []) + self.assertEqual(self.assigns, []) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'scheduler_task_runs') + + def test_check_assign_timeout(self): + # 'Task assignment timeout' case + create_ts = 0 + now = 1000000 + self.sched.active_tasks = [{'task_id': 99, 'host_id': 23, 'state': koji.TASK_STATES['ASSIGNED']}] + self.sched.hosts = {23: {'id': 23, 'name': 'test host 23'}} + self.sched.get_active_runs.return_value = {99: [{'create_ts': create_ts}]} + with mock.patch('time.time', return_value=now): + self.sched.check_active_tasks() + self.get_active_runs.assert_called_once() + self.log_db.assert_called_once_with('Task assignment timeout', 99, 23) + # we should free such tasks + self.assertEqual(self.frees, [99]) + self.assertEqual(self.assigns, []) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'scheduler_task_runs') + + def test_check_unresponsive(self): + # 'Freeing task from unresponsive host' case + now = 1000000 + create_ts = 0 + update_ts = 0 + self.sched.active_tasks = [{'task_id': 99, 'host_id': 23, 'state': koji.TASK_STATES['OPEN']}] + self.sched.hosts = {23: {'id': 23, 'name': 'test host 23', 'update_ts': update_ts}} + self.sched.get_active_runs.return_value = {99: [{'create_ts': create_ts}]} + with mock.patch('time.time', return_value=now): + self.sched.check_active_tasks() + self.get_active_runs.assert_called_once() + # we should free such tasks + self.log_db.assert_called_once_with('Freeing task from unresponsive host', 99, 23) + self.assertEqual(self.frees, [99]) + self.assertEqual(self.assigns, []) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'scheduler_task_runs') diff --git a/tests/test_hub/test_upsert_processor.py b/tests/test_hub/test_upsert_processor.py new file mode 100644 index 0000000..cf7aee8 --- /dev/null +++ b/tests/test_hub/test_upsert_processor.py @@ -0,0 +1,38 @@ +import mock +import unittest + +import koji +import kojihub + + +class TestUpsertProcessor(unittest.TestCase): + def setUp(self): + self.context_db = mock.patch('kojihub.db.context').start() + + def tearDown(self): + mock.patch.stopall() + + def test_required_args(self): + with self.assertRaises(ValueError) as e: + proc = kojihub.UpsertProcessor('sometable') + self.assertEqual(e.msg, 'either keys or skip_dup must be set') + + def test_skip_dup(self): + proc = kojihub.UpsertProcessor('sometable', data={'foo': 'bar'}, skip_dup=True) + actual = str(proc) + expected = 'INSERT INTO sometable (foo) VALUES (%(foo)s) ON CONFLICT DO NOTHING' + self.assertEqual(actual, expected) + + def test_key(self): + proc = kojihub.UpsertProcessor('sometable', data={'id': 1, 'foo': 'bar'}, keys=['id']) + actual = str(proc) + expected = 'INSERT INTO sometable (foo, id) VALUES (%(foo)s, %(id)s) ON CONFLICT (id) DO UPDATE SET foo = %(foo)s' + self.assertEqual(actual, expected) + + def test_keys(self): + proc = kojihub.UpsertProcessor('sometable', data={'id': 1, 'package': 'koji', 'foo': 'bar'}, keys=['id', 'package']) + actual = str(proc) + expected = 'INSERT INTO sometable (foo, id, package) VALUES (%(foo)s, %(id)s, %(package)s) ' \ + 'ON CONFLICT (id,package) DO UPDATE SET foo = %(foo)s' + self.assertEqual(actual, expected) +