527 lines
35 KiB
Python
527 lines
35 KiB
Python
from unittest import mock
|
|
import os
|
|
import time
|
|
|
|
import koji
|
|
import kojihub
|
|
import kojihub.kojihub
|
|
from .utils import DBQueryTestCase
|
|
|
|
|
|
class TestQueryHistory(DBQueryTestCase):
|
|
def setUp(self):
|
|
super(TestQueryHistory, self).setUp()
|
|
self.maxDiff = None
|
|
self.context = mock.patch('kojihub.kojihub.context').start()
|
|
self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start()
|
|
self.get_build = mock.patch('kojihub.kojihub.get_build').start()
|
|
self.get_id = mock.patch('kojihub.kojihub.get_id').start()
|
|
self.get_package_id = mock.patch('kojihub.kojihub.get_package_id').start()
|
|
self.get_user = mock.patch('kojihub.kojihub.get_user').start()
|
|
self.get_perm_id = mock.patch('kojihub.kojihub.get_perm_id').start()
|
|
self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start()
|
|
self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start()
|
|
self.get_build_target_id = mock.patch('kojihub.kojihub.get_build_target_id').start()
|
|
self.get_group_id = mock.patch('kojihub.kojihub.get_group_id').start()
|
|
self.original_timezone = os.environ.get('TZ')
|
|
os.environ['TZ'] = 'UTC'
|
|
time.tzset()
|
|
|
|
def tearDown(self):
|
|
if self.original_timezone is None:
|
|
del os.environ['TZ']
|
|
else:
|
|
os.environ['TZ'] = self.original_timezone
|
|
time.tzset()
|
|
mock.patch.stopall()
|
|
|
|
def test_not_exist_table(self):
|
|
table_name = 'test-table'
|
|
with self.assertRaises(koji.GenericError) as cm:
|
|
kojihub.query_history(tables=[table_name])
|
|
self.assertEqual(f"No such history table: {table_name}", str(cm.exception))
|
|
self.assertEqual(len(self.queries), 0)
|
|
|
|
def test_valid(self):
|
|
self.get_tag_id.return_value = 12
|
|
self.get_build.return_value = {'id': 13}
|
|
self.get_package_id.return_value = 15
|
|
self.get_user.return_value = {'id': 357, 'name': 'test-editor'}
|
|
kojihub.query_history(tables=['user_perms', 'user_groups', 'cg_users', 'tag_inheritance',
|
|
'tag_config', 'tag_extra', 'build_target_config',
|
|
'external_repo_config', 'host_config', 'host_channels',
|
|
'tag_external_repos', 'tag_listing', 'tag_packages',
|
|
'tag_package_owners', 'group_config', 'group_req_listing',
|
|
'group_package_listing'],
|
|
tag='test-tag', build='test-build-1.23-1', package='test-pkg',
|
|
active=True, editor='test-editor', afterEvent=555)
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['tag_listing'])
|
|
self.assertEqual(query.columns, ['tag_listing.create_event > %(afterEvent)i',
|
|
'creator.id = %(editor)i',
|
|
'tag_listing.revoke_event > %(afterEvent)i',
|
|
'revoker.id = %(editor)i', 'tag_listing.active',
|
|
'build.state', 'tag_listing.build_id',
|
|
'tag_listing.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'tag_listing.creator_id', 'creator.name', 'build.epoch',
|
|
'package.name', 'build.release',
|
|
'tag_listing.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'tag_listing.revoker_id', 'revoker.name', 'tag.name',
|
|
'tag_listing.tag_id', 'build.version'])
|
|
self.assertEqual(query.clauses, ['active = TRUE', 'build.id = %(build_id)i',
|
|
'creator.id = %(editor)i OR revoker.id = %(editor)i',
|
|
'package.id = %(pkg_id)i', 'tag.id = %(tag_id)i',
|
|
'tag_listing.create_event > %(afterEvent)i OR '
|
|
'tag_listing.revoke_event > %(afterEvent)i'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'build ON build_id = build.id',
|
|
'package ON build.pkg_id = package.id',
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id'])
|
|
self.assertEqual(query.values, {'afterEvent': 555, 'build_id': 13, 'editor': 357,
|
|
'pkg_id': 15, 'tag_id': 12})
|
|
|
|
def test_group_key(self):
|
|
self.get_group_id.return_value = 24
|
|
kojihub.query_history(group='test-grp')
|
|
self.assertEqual(len(self.queries), 3)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['group_config'])
|
|
self.assertEqual(query.columns, ['group_config.active', 'group_config.biarchonly',
|
|
'group_config.blocked', 'group_config.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'group_config.creator_id', 'creator.name',
|
|
'group_config.description', 'group_config.display_name',
|
|
'group_config.exported', 'groups.name',
|
|
'group_config.group_id', 'group_config.is_default',
|
|
'group_config.langonly', 'group_config.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'group_config.revoker_id', 'revoker.name', 'tag.name',
|
|
'group_config.tag_id', 'group_config.uservisible'])
|
|
self.assertEqual(query.clauses, ['groups.id = %(group_id)i'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'groups ON group_id = groups.id',
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id'])
|
|
self.assertEqual(query.values, {'group_id': 24})
|
|
query = self.queries[1]
|
|
self.assertEqual(query.tables, ['group_package_listing'])
|
|
self.assertEqual(query.columns, ['group_package_listing.active',
|
|
'group_package_listing.basearchonly',
|
|
'group_package_listing.blocked',
|
|
'group_package_listing.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'group_package_listing.creator_id', 'creator.name',
|
|
'groups.name', 'group_package_listing.group_id',
|
|
'group_package_listing.package',
|
|
'group_package_listing.requires',
|
|
'group_package_listing.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'group_package_listing.revoker_id',
|
|
'revoker.name', 'tag.name',
|
|
'group_package_listing.tag_id',
|
|
'group_package_listing.type'])
|
|
self.assertEqual(query.clauses, ['groups.id = %(group_id)i'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'groups ON group_id = groups.id',
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id'])
|
|
self.assertEqual(query.values, {'group_id': 24})
|
|
|
|
query = self.queries[2]
|
|
self.assertEqual(query.tables, ['group_req_listing'])
|
|
self.assertEqual(query.columns, ['group_req_listing.active', 'group_req_listing.blocked',
|
|
'group_req_listing.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'group_req_listing.creator_id', 'creator.name',
|
|
'groups.name', 'group_req_listing.group_id',
|
|
'group_req_listing.is_metapkg', 'req.name',
|
|
'group_req_listing.req_id',
|
|
'group_req_listing.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'group_req_listing.revoker_id', 'revoker.name',
|
|
'tag.name', 'group_req_listing.tag_id',
|
|
'group_req_listing.type'])
|
|
self.assertEqual(query.clauses, ['req.id = %(group_id)i'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'groups ON group_id = groups.id',
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id',
|
|
'LEFT OUTER JOIN groups AS req ON req_id = req.id'])
|
|
self.assertEqual(query.values, {'group_id': 24})
|
|
|
|
def test_host_key_and_active_false(self):
|
|
self.get_id.return_value = 22
|
|
kojihub.query_history(host='test-host', active=False)
|
|
self.assertEqual(len(self.queries), 2)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['host_channels'])
|
|
self.assertEqual(query.clauses, ['active IS NULL', 'host.id = %(host_id)i'])
|
|
self.assertEqual(query.columns, ['host_channels.active', 'host_channels.channel_id',
|
|
'channels.name', 'host_channels.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'host_channels.creator_id', 'creator.name', 'host.name',
|
|
'host_channels.host_id', 'host_channels.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'host_channels.revoker_id', 'revoker.name'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN host ON host_id = host.id',
|
|
'LEFT OUTER JOIN channels ON channel_id = channels.id'])
|
|
self.assertEqual(query.values, {'host_id': 22})
|
|
query = self.queries[1]
|
|
self.assertEqual(query.tables, ['host_config'])
|
|
self.assertEqual(query.columns, ['host_config.active', 'host_config.arches',
|
|
'host_config.capacity', 'host_config.comment',
|
|
'host_config.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'host_config.creator_id', 'creator.name',
|
|
'host_config.description', 'host_config.enabled',
|
|
'host.name', 'host_config.host_id',
|
|
'host_config.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'host_config.revoker_id', 'revoker.name'])
|
|
self.assertEqual(query.clauses, ['active IS NULL', 'host.id = %(host_id)i'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN host ON host_id = host.id'])
|
|
self.assertEqual(query.values, {'host_id': 22})
|
|
|
|
def test_channel_and_before_event_key(self):
|
|
self.get_id.return_value = 22
|
|
kojihub.query_history(channel='test-channel', beforeEvent=99)
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['host_channels'])
|
|
self.assertEqual(query.clauses, ['channels.id = %(channel_id)i',
|
|
'host_channels.create_event < %(beforeEvent)i OR '
|
|
'host_channels.revoke_event < %(beforeEvent)i'])
|
|
self.assertEqual(query.columns, ['host_channels.create_event < %(beforeEvent)i',
|
|
'host_channels.revoke_event < %(beforeEvent)i',
|
|
'host_channels.active', 'host_channels.channel_id',
|
|
'channels.name', 'host_channels.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'host_channels.creator_id', 'creator.name', 'host.name',
|
|
'host_channels.host_id', 'host_channels.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'host_channels.revoker_id', 'revoker.name'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN host ON host_id = host.id',
|
|
'LEFT OUTER JOIN channels ON channel_id = channels.id'])
|
|
self.assertEqual(query.values, {'beforeEvent': 99, 'channel_id': 22})
|
|
|
|
def test_user_and_before_key(self):
|
|
self.get_user.return_value = {'id': 159, 'name': 'test-user'}
|
|
kojihub.query_history(user='test-user', before=1681925020)
|
|
self.assertEqual(len(self.queries), 4)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['cg_users'])
|
|
self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s',
|
|
'users.id = %(affected_user_id)i'])
|
|
self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s',
|
|
'cg_users.active', 'cg_users.cg_id',
|
|
'content_generator.name', 'cg_users.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'cg_users.creator_id', 'creator.name',
|
|
'cg_users.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'cg_users.revoker_id', 'revoker.name', 'users.name',
|
|
'cg_users.user_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN users ON user_id = users.id',
|
|
'LEFT OUTER JOIN content_generator ON cg_id = content_generator.id'])
|
|
self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40+00:00'})
|
|
|
|
query = self.queries[1]
|
|
self.assertEqual(query.tables, ['tag_package_owners'])
|
|
self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s',
|
|
'owner.id = %(affected_user_id)i'])
|
|
self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s',
|
|
'tag_package_owners.active',
|
|
'tag_package_owners.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'tag_package_owners.creator_id', 'creator.name',
|
|
'tag_package_owners.owner', 'owner.name', 'package.name',
|
|
'tag_package_owners.package_id',
|
|
'tag_package_owners.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'tag_package_owners.revoker_id', 'revoker.name',
|
|
'tag.name', 'tag_package_owners.tag_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN package ON package_id = package.id',
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id',
|
|
'LEFT OUTER JOIN users AS owner ON owner = owner.id'])
|
|
self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40+00:00'})
|
|
|
|
query = self.queries[2]
|
|
self.assertEqual(query.tables, ['user_groups'])
|
|
self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s',
|
|
'usergroup.id = %(affected_user_id)i'])
|
|
self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s',
|
|
'user_groups.active', 'user_groups.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'user_groups.creator_id', 'creator.name',
|
|
'usergroup.name', 'user_groups.group_id',
|
|
'user_groups.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'user_groups.revoker_id', 'revoker.name',
|
|
'users.name', 'user_groups.user_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN users ON user_id = users.id',
|
|
'users AS usergroup ON group_id = usergroup.id'])
|
|
self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40+00:00'})
|
|
|
|
query = self.queries[3]
|
|
self.assertEqual(query.tables, ['user_perms'])
|
|
self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s',
|
|
'users.id = %(affected_user_id)i'])
|
|
self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s',
|
|
'user_perms.active', 'user_perms.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'user_perms.creator_id', 'creator.name',
|
|
'user_perms.perm_id', 'permission.name',
|
|
'user_perms.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'user_perms.revoker_id', 'revoker.name', 'users.name',
|
|
'user_perms.user_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN users ON user_id = users.id',
|
|
'LEFT OUTER JOIN permissions AS permission ON perm_id = permission.id'])
|
|
self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40+00:00'})
|
|
|
|
def test_permission_and_after_key(self):
|
|
self.get_perm_id.return_value = 66
|
|
kojihub.query_history(permission='test-perms', after=1681925020)
|
|
self.assertEqual(len(self.queries), 2)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['tag_config'])
|
|
self.assertEqual(query.clauses, ['ev1.time > %(after)s OR ev2.time > %(after)s',
|
|
'permission.id = %(perm_id)i'])
|
|
self.assertEqual(query.columns, ['ev1.time > %(after)s', 'ev2.time > %(after)s',
|
|
'tag_config.active', 'tag_config.arches',
|
|
'tag_config.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'tag_config.creator_id', 'creator.name',
|
|
'tag_config.locked', 'tag_config.maven_include_all',
|
|
'tag_config.maven_support', 'tag_config.perm_id',
|
|
'permission.name', 'tag_config.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'tag_config.revoker_id', 'revoker.name', 'tag.name',
|
|
'tag_config.tag_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id',
|
|
'LEFT OUTER JOIN permissions AS permission ON perm_id = permission.id'])
|
|
self.assertEqual(query.values, {'after': '2023-04-19 17:23:40+00:00', 'perm_id': 66})
|
|
|
|
query = self.queries[1]
|
|
self.assertEqual(query.tables, ['user_perms'])
|
|
self.assertEqual(query.clauses, ['ev1.time > %(after)s OR ev2.time > %(after)s',
|
|
'permission.id = %(perm_id)i'])
|
|
self.assertEqual(query.columns, ['ev1.time > %(after)s', 'ev2.time > %(after)s',
|
|
'user_perms.active', 'user_perms.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'user_perms.creator_id', 'creator.name',
|
|
'user_perms.perm_id', 'permission.name',
|
|
'user_perms.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'user_perms.revoker_id', 'revoker.name', 'users.name',
|
|
'user_perms.user_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN users ON user_id = users.id',
|
|
'LEFT OUTER JOIN permissions AS permission ON perm_id = permission.id'])
|
|
self.assertEqual(query.values, {'after': '2023-04-19 17:23:40+00:00', 'perm_id': 66})
|
|
|
|
def test_cg_key(self):
|
|
self.lookup_name.return_value = {'id': 147}
|
|
kojihub.query_history(cg='test-cg')
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['cg_users'])
|
|
self.assertEqual(query.clauses, ['content_generator.id = %(cg_id)i'])
|
|
self.assertEqual(query.columns, ['cg_users.active', 'cg_users.cg_id',
|
|
'content_generator.name', 'cg_users.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'cg_users.creator_id', 'creator.name',
|
|
'cg_users.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'cg_users.revoker_id', 'revoker.name', 'users.name',
|
|
'cg_users.user_id'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN users ON user_id = users.id',
|
|
'LEFT OUTER JOIN content_generator ON cg_id = content_generator.id'])
|
|
self.assertEqual(query.values, {'cg_id': 147})
|
|
|
|
def test_external_repo_key(self):
|
|
self.get_external_repo_id.return_value = 49
|
|
kojihub.query_history(external_repo='test-ext-repo')
|
|
self.assertEqual(len(self.queries), 3)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['external_repo_config'])
|
|
self.assertEqual(query.clauses, ['external_repo.id = %(external_repo_id)i'])
|
|
self.assertEqual(query.columns, ['external_repo_config.active',
|
|
'external_repo_config.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'external_repo_config.creator_id', 'creator.name',
|
|
'external_repo.name',
|
|
'external_repo_config.external_repo_id',
|
|
'external_repo_config.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'external_repo_config.revoker_id', 'revoker.name',
|
|
'external_repo_config.url'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN external_repo ON external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'external_repo_id': 49})
|
|
|
|
query = self.queries[1]
|
|
self.assertEqual(query.tables, ['external_repo_data'])
|
|
self.assertEqual(query.clauses, ['external_repo.id = %(external_repo_id)i'])
|
|
self.assertEqual(query.columns, ['external_repo_data.active',
|
|
'external_repo_data.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'external_repo_data.creator_id', 'creator.name',
|
|
'external_repo_data.data',
|
|
'external_repo.name',
|
|
'external_repo_data.external_repo_id',
|
|
'external_repo_data.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'external_repo_data.revoker_id', 'revoker.name',
|
|
])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN external_repo ON external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'external_repo_id': 49})
|
|
|
|
query = self.queries[2]
|
|
self.assertEqual(query.tables, ['tag_external_repos'])
|
|
self.assertEqual(query.clauses, ['external_repo.id = %(external_repo_id)i'])
|
|
self.assertEqual(query.columns, ['tag_external_repos.active',
|
|
'tag_external_repos.arches',
|
|
'tag_external_repos.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'tag_external_repos.creator_id', 'creator.name',
|
|
'external_repo.name',
|
|
'tag_external_repos.external_repo_id',
|
|
'tag_external_repos.merge_mode',
|
|
'tag_external_repos.priority',
|
|
'tag_external_repos.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'tag_external_repos.revoker_id', 'revoker.name',
|
|
'tag.name', 'tag_external_repos.tag_id'
|
|
])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id',
|
|
'LEFT OUTER JOIN external_repo ON external_repo_id = external_repo.id'])
|
|
self.assertEqual(query.values, {'external_repo_id': 49})
|
|
|
|
def test_build_target_key(self):
|
|
self.get_build_target_id.return_value = 55
|
|
kojihub.query_history(build_target='test-target')
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['build_target_config'])
|
|
self.assertEqual(query.clauses, ['build_target.id = %(build_target_id)i'])
|
|
self.assertEqual(query.columns, ['build_target_config.active',
|
|
'build_target_config.build_tag', 'build_tag.name',
|
|
'build_target.name',
|
|
'build_target_config.build_target_id',
|
|
'build_target_config.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'build_target_config.creator_id', 'creator.name',
|
|
'build_target_config.dest_tag', 'dest_tag.name',
|
|
'build_target_config.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'build_target_config.revoker_id', 'revoker.name'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN build_target ON build_target_id = build_target.id',
|
|
'LEFT OUTER JOIN tag AS build_tag ON build_tag = build_tag.id',
|
|
'LEFT OUTER JOIN tag AS dest_tag ON dest_tag = dest_tag.id'])
|
|
self.assertEqual(query.values, {'build_target_id': 55})
|
|
|
|
def test_xkey_key(self):
|
|
kojihub.query_history(xkey='test-key')
|
|
self.assertEqual(len(self.queries), 1)
|
|
query = self.queries[0]
|
|
self.assertEqual(query.tables, ['tag_extra'])
|
|
self.assertEqual(query.clauses, ['tag_extra.key = %(key)s'])
|
|
self.assertEqual(query.columns, ['tag_extra.active', 'tag_extra.create_event',
|
|
"date_part('epoch', ev1.time) AS create_ts",
|
|
'tag_extra.creator_id', 'creator.name', 'tag_extra.key',
|
|
'tag_extra.revoke_event',
|
|
"date_part('epoch', ev2.time) AS revoke_ts",
|
|
'tag_extra.revoker_id', 'revoker.name', 'tag.name',
|
|
'tag_extra.tag_id', 'tag_extra.value'])
|
|
self.assertEqual(query.joins,
|
|
["events AS ev1 ON ev1.id = create_event",
|
|
"LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event",
|
|
"users AS creator ON creator.id = creator_id",
|
|
"LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id",
|
|
'LEFT OUTER JOIN tag ON tag_id = tag.id'])
|
|
self.assertEqual(query.values, {'key': 'test-key'})
|