from __future__ import absolute_import import time from datetime import datetime from dateutil.tz import tzutc import unittest from six.moves import StringIO try: from unittest import mock except ImportError: import mock import koji from koji_cli.commands import anon_handle_list_history from . import utils class TestListHistory(utils.CliTestCase): def setUp(self): self.options = mock.MagicMock() self.options.debug = False self.session = mock.MagicMock() self.maxDiff = None self.ensure_connection = mock.patch('koji_cli.commands.ensure_connection').start() self.error_format = """Usage: %s list-history [options] (Specify the --help global option for a list of other help options) %s: error: {message} """ % (self.progname, self.progname) def tearDown(self): mock.patch.stopall() @staticmethod def get_expected_date_active_action(item, act='add', utc=False): if act == 'add': if utc: dt = datetime.fromtimestamp(item['create_ts'], tzutc()) else: dt = datetime.fromtimestamp(item['create_ts']) if item['active']: active = ' [still active]' else: active = '' else: if utc: dt = datetime.fromtimestamp(item['revoke_ts'], tzutc()) else: dt = datetime.fromtimestamp(item['revoke_ts']) active = '' expected_date = time.asctime(dt.timetuple()) return expected_date, active def get_expected_channel(self, item, utc=False): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act, utc) if act == 'add': action = 'added to' else: action = 'removed from' expected = expected + '{} host {} {} channel {} by {}{}\n'.format( expected_date, item['host.name'], action, item['channels.name'], item['creator_name'], active) return expected def get_expected_host(self, item): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act) if act == 'add': action = 'new host' else: action = 'host deleted' expected = expected + "{} {}: {} by {}{}\n".format( expected_date, action, item['host.name'], item['creator_name'], active) return expected def get_expected_package_owner(self, item): expected_date, active = self.get_expected_date_active_action(item) expected = "{} package owner {} set for {} in {} by {}{}\n".format( expected_date, item['owner.name'], item['package.name'], item['tag.name'], item['creator_name'], active) return expected def get_expected_tag_listing(self, item): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act) if act == 'add': action = 'tagged into' else: action = 'untagged from' expected = expected + "{} {}-{}-{} {} {} by {}{}\n".format( expected_date, item['name'], item['version'], item['release'], action, item['tag.name'], item['creator_name'], active) return expected def get_expected_tag_config(self, item): expected_date, active = self.get_expected_date_active_action(item) expected = "{} new tag: {} by {}{}\n".format( expected_date, item['tag.name'], item['creator_name'], active) return expected def get_expected_tag_extra(self, item): expected_date, active = self.get_expected_date_active_action(item) expected = "{} added tag option {} for tag {} by {}{}\n".format( expected_date, item['key'], item['tag.name'], item['creator_name'], active) return expected def get_expected_build_target(self, item): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act) if act == 'add': action = 'new build target' else: action = 'build target deleted' expected = expected + "{} {}: {} by {}{}\n".format( expected_date, action, item['build_target.name'], item['creator_name'], active) return expected def get_expected_tag_packages(self, item): expected_date, active = self.get_expected_date_active_action(item) expected = "{} package list entry created: {} in {} by {}{}\n".format( expected_date, item['package.name'], item['tag.name'], item['creator_name'], active) return expected def get_expected_tag_inheritance(self, item): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act) if act == 'add': action = 'added' else: action = 'removed' expected = \ expected + "{} inheritance line {}->{} {} by {}{}\n".format( expected_date, item['tag.name'], item['parent.name'], action, item['creator_name'], active) return expected def get_expected_user_perm(self, item): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act) if act == 'add': action = 'granted to' else: action = 'revoked for' expected = expected + "{} permission {} {} {} by {}{}\n".format( expected_date, item['permission.name'], action, item['user.name'], item['creator_name'], active) return expected def get_expected_cg_user(self, item): if item['active']: list_active = ['add'] else: list_active = ['add', 'remove'] expected = '' for act in list_active: expected_date, active = self.get_expected_date_active_action( item, act) if act == 'add': action = 'added to' else: action = 'removed from' expected = \ expected + "{} user {} {} content generator {} by {}" \ "{}\n".format(expected_date, item['user.name'], action, item['content_generator.name'], item['creator_name'], active) return expected def get_expected_ext_repo(self, item): expected_date, active = self.get_expected_date_active_action(item) expected = "{} new external repo: {} by {}{}\n".format( expected_date, item['external_repo.name'], item['creator_name'], active) return expected def get_expected_tag_ext_repo(self, item): expected_date, active = self.get_expected_date_active_action(item) expected = "{} external repo entry for {} added to tag {} " \ "by {}{}\n".format(expected_date, item['external_repo.name'], item['tag.name'], item['creator_name'], active) return expected @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_channel_utc(self, stdout): # test case when channel is still active channel_name = 'default' # when channel name is created dict_history = { 'host_channels': [ {'active': True, 'channel_id': 1, 'channels.name': 'default', 'create_event': 3, 'create_ts': 1612355089.887727, 'creator_id': 1, 'creator_name': 'kojiadmin', 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_channel(dict_history['host_channels'][0], utc=True) anon_handle_list_history(self.options, self.session, ['--channel', channel_name, '--utc']) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_channel(self, stdout): # test case when channel is still active channel_name = 'default' # when channel name is created dict_history = { 'host_channels': [ {'active': True, 'channel_id': 1, 'channels.name': 'default', 'create_event': 3, 'create_ts': 1612355089.887727, 'creator_id': 1, 'creator_name': 'kojiadmin', 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_channel(dict_history['host_channels'][0]) anon_handle_list_history(self.options, self.session, ['--channel', channel_name]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # when channel name is dropped dict_history = { 'host_channels': [ {'active': None, 'channel_id': 1, 'channels.name': 'default', 'create_event': 3, 'create_ts': 1612355089.887727, 'creator_id': 1, 'creator_name': 'kojiadmin', 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': 8, 'revoke_ts': 1612355099.887727, 'revoker_id': 1, 'revoker_name': 'kojiadmin'}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_channel(dict_history['host_channels'][0]) anon_handle_list_history(self.options, self.session, ['--channel', channel_name]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when channel is not existing self.session.untaggedBuilds.return_value = {} self.session.getChannel.return_value = None arguments = ['--channel', channel_name] self.assert_system_exit( anon_handle_list_history, self.options, self.session, arguments, stderr="No such channel: %s" % channel_name + "\n", stdout='', activate_session=None, exit_code=1) self.ensure_connection.assert_not_called() @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_host(self, stdout): host_name = 'kojibuilder' # new host dict_history = { 'host_config': [ {'active': True, 'arches': 'x86_64', 'capacity': 2.0, 'comment': None, 'create_event': 2, 'create_ts': 1612355089.886359, 'creator_id': 1, 'creator_name': 'kojiadmin', 'description': None, 'enabled': True, 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_host(dict_history['host_config'][0]) anon_handle_list_history(self.options, self.session, ['--host', host_name]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # dropped dict_history = { 'host_config': [ {'active': None, 'arches': 'x86_64', 'capacity': 2.0, 'comment': None, 'create_event': 2, 'create_ts': 1612355089.886359, 'creator_id': 1, 'creator_name': 'kojiadmin', 'description': None, 'enabled': True, 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': 3, 'revoke_ts': 1612355099.886359, 'revoker_id': 1, 'revoker_name': 'kojiadmin'}] } host_name = 'kojibuilder' self.session.queryHistory.return_value = dict_history expected = self.get_expected_host(dict_history['host_config'][0]) anon_handle_list_history(self.options, self.session, ['--host', host_name]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when host is not existing self.session.untaggedBuilds.return_value = {} self.session.getHost.return_value = None arguments = ['--host', host_name] self.assert_system_exit( anon_handle_list_history, self.options, self.session, arguments, stderr="No such host: %s" % host_name + "\n", stdout='', activate_session=None, exit_code=1) self.ensure_connection.assert_not_called() @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_build(self, stdout): build_nvr = 'test-build-1.1-11' # when build is tagged dict_history = { 'tag_listing': [ {'active': True, 'build.state': 1, 'build_id': 6, 'create_event': 585, 'create_ts': 1613744957.14465, 'creator_id': 1, 'creator_name': 'kojiadmin', 'epoch': 7, 'name': 'test-build', 'release': '11', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination', 'tag_id': 13, 'version': '1.1'}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_tag_listing(dict_history['tag_listing'][0]) anon_handle_list_history(self.options, self.session, ['--build', build_nvr]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # when build is untagged dict_history = { 'tag_listing': [ {'active': None, 'build.state': 1, 'build_id': 6, 'create_event': 585, 'create_ts': 1613744957.14465, 'creator_id': 1, 'creator_name': 'kojiadmin', 'epoch': 7, 'name': 'test-build', 'release': '11', 'revoke_event': 590, 'revoke_ts': 1613744967.14465, 'revoker_id': 1, 'revoker_name': 'kojiadmin', 'tag.name': 'destination-tag', 'tag_id': 13, 'version': '1.1'}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_tag_listing(dict_history['tag_listing'][0]) anon_handle_list_history(self.options, self.session, ['--build', build_nvr]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when build nvr is not existing expected = "No matching build found: %s" % build_nvr + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--build', build_nvr]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when build has wrong format build = 'test-build' expected = "invalid format: %s" % build + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--build', build_nvr]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_package(self, stdout): dict_history = { 'tag_listing': [ {'active': True, 'build.state': 1, 'build_id': 4, 'create_event': 424, 'create_ts': 1613736474.42776, 'creator_id': 1, 'creator_name': 'kojiadmin', 'epoch': 7, 'name': 'pkg-name', 'release': '11', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination-test-tag', 'tag_id': 10, 'version': '1.1'}], 'tag_package_owners': [ {'active': True, 'create_event': 418, 'create_ts': 1613736220.79199, 'creator_id': 1, 'creator_name': 'kojiadmin', 'owner': 1, 'owner.name': 'kojiadmin', 'package.name': 'pkg-name', 'package_id': 4, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination-test-tag', 'tag_id': 10}], 'tag_packages': [ {'active': True, 'blocked': False, 'create_event': 418, 'create_ts': 1613736220.79199, 'creator_id': 1, 'creator_name': 'kojiadmin', 'extra_arches': '', 'package.name': 'pkg-name', 'package_id': 4, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination-test-tag', 'tag_id': 10}] } package = 'pkg-name' self.session.queryHistory.return_value = dict_history expected = self.get_expected_package_owner( dict_history['tag_package_owners'][0]) expected = expected + self.get_expected_tag_packages( dict_history['tag_packages'][0]) expected = expected + self.get_expected_tag_listing( dict_history['tag_listing'][0]) anon_handle_list_history(self.options, self.session, ['--package', package]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when package s not existing expected = "No such entry in table package: %s" % package + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--package', package]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_tag(self, stdout): dict_history = { 'tag_config': [ {'active': True, 'arches': 'x86_64', 'create_event': 6, 'create_ts': 1612872591.313584, 'creator_id': 1, 'creator_name': 'kojiadmin', 'locked': False, 'maven_include_all': False, 'maven_support': False, 'perm_id': None, 'permission.name': None, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'test-tag', 'tag_id': 2}], 'tag_extra': [ {'active': True, 'create_event': 6, 'create_ts': 1612872591.313584, 'creator_id': 1, 'creator_name': 'kojiadmin', 'key': 'mock.package_manager', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'test-tag', 'tag_id': 2, 'value': '"dnf"'}] } tag = 'test-tag' self.session.queryHistory.return_value = dict_history expected = self.get_expected_tag_config(dict_history['tag_config'][0]) expected = expected + self.get_expected_tag_extra( dict_history['tag_extra'][0]) anon_handle_list_history(self.options, self.session, ['--tag', tag]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when tag s not existing expected = "No such entry in table tag: %s" % tag + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--tag', tag]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_editor(self, stdout): dict_history = { 'build_target_config': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'build_tag': 1, 'build_tag.name': 'test-tag', 'build_target.name': 'build-target-test-tag', 'build_target_id': 1, 'create_event': 8, 'create_ts': 1612872656.231499, 'creator_id': 1, 'creator_name': 'kojiadmin', 'dest_tag': 3, 'dest_tag.name': 'destination-test-tag', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}], 'host_channels': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'channel_id': 1, 'channels.name': 'default', 'create_event': 3, 'create_ts': 1612355089.887727, 'creator_id': 1, 'creator_name': 'kojiadmin', 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}], 'host_config': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'arches': 'x86_64', 'capacity': 2.0, 'comment': None, 'create_event': 2, 'create_ts': 1612355089.886359, 'creator_id': 1, 'creator_name': 'kojiadmin', 'description': None, 'enabled': True, 'host.name': 'kojibuilder', 'host_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}], 'tag_config': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'arches': '', 'create_event': 5, 'create_ts': 1612871243.593475, 'creator_id': 1, 'creator_name': 'kojiadmin', 'locked': False, 'maven_include_all': False, 'maven_support': False, 'perm_id': None, 'permission.name': None, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'test-tag', 'tag_id': 1}], 'tag_extra': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'create_event': 6, 'create_ts': 1612872591.313584, 'creator_id': 1, 'creator_name': 'kojiadmin', 'key': 'mock.package_manager', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'test-tag', 'tag_id': 1, 'value': '"dnf"'}], 'tag_inheritance': [ {'_created_by': True, '_revoked_by': True, 'active': None, 'create_event': 16, 'create_ts': 1613545870.370125, 'creator_id': 1, 'creator_name': 'kojiadmin', 'intransitive': False, 'maxdepth': None, 'noconfig': False, 'parent.name': 'parent-test-tag', 'parent_id': 4, 'pkg_filter': '', 'priority': 1, 'revoke_event': 31, 'revoke_ts': 1613545887.513565, 'revoker_id': 1, 'revoker_name': 'kojiadmin', 'tag.name': 'test-tag', 'tag_id': 5}], 'tag_package_owners': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'create_event': 9, 'create_ts': 1612872778.934647, 'creator_id': 1, 'creator_name': 'kojiadmin', 'owner': 1, 'owner.name': 'kojiadmin', 'package.name': 'koji', 'package_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination-test-tag', 'tag_id': 3}], 'tag_packages': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'blocked': False, 'create_event': 9, 'create_ts': 1612872778.934647, 'creator_id': 1, 'creator_name': 'kojiadmin', 'extra_arches': '', 'package.name': 'koji', 'package_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination-test-tag', 'tag_id': 3}], 'user_perms': [ {'_created_by': True, '_revoked_by': None, 'active': True, 'create_event': 1, 'create_ts': 1612355089.882428, 'creator_id': 1, 'creator_name': 'kojiadmin', 'perm_id': 1, 'permission.name': 'admin', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'user.name': 'kojiadmin', 'user_id': 1}] } editor = 'kojiadmin' self.session.queryHistory.return_value = dict_history expected = self.get_expected_user_perm(dict_history['user_perms'][0]) expected = expected + self.get_expected_host( dict_history['host_config'][0]) expected = expected + self.get_expected_channel( dict_history['host_channels'][0]) expected = expected + self.get_expected_tag_config( dict_history['tag_config'][0]) expected = expected + self.get_expected_tag_extra( dict_history['tag_extra'][0]) expected = expected + self.get_expected_build_target( dict_history['build_target_config'][0]) expected = expected + self.get_expected_package_owner( dict_history['tag_package_owners'][0]) expected = expected + self.get_expected_tag_packages( dict_history['tag_packages'][0]) expected = expected + self.get_expected_tag_inheritance( dict_history['tag_inheritance'][0]) anon_handle_list_history(self.options, self.session, ['--editor', editor]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when tag is not existing expected = "No such user: %s" % editor + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--editor', editor]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_user(self, stdout): dict_history = { 'cg_users': [], 'tag_package_owners': [ {'active': True, 'create_event': 9, 'create_ts': 1613730799.934647, 'creator_id': 1, 'creator_name': 'kojiadmin', 'owner': 1, 'owner.name': 'kojiadmin', 'package.name': 'koji', 'package_id': 1, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'destination-test-tag', 'tag_id': 3}], 'user_groups': [], 'user_perms': [ {'active': True, 'create_event': 1, 'create_ts': 1612355089.882428, 'creator_id': 1, 'creator_name': 'kojiadmin', 'perm_id': 1, 'permission.name': 'admin', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'user.name': 'kojiadmin', 'user_id': 1}, {'active': None, 'create_event': 6, 'create_ts': 1613730777.437744, 'creator_id': 1, 'creator_name': 'kojiadmin', 'perm_id': 4, 'permission.name': 'dist-repo', 'revoke_event': 7, 'revoke_ts': 1613730790.797031, 'revoker_id': 1, 'revoker_name': 'kojiadmin', 'user.name': 'kojiadmin', 'user_id': 1}] } username = 'kojiadmin' self.session.queryHistory.return_value = dict_history expected = '' for item in dict_history['user_perms']: expected = expected + self.get_expected_user_perm(item) expected = expected + self.get_expected_package_owner( dict_history['tag_package_owners'][0]) anon_handle_list_history(self.options, self.session, ['--user', username]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when user is not existing expected = "No such user: %s" % username + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--user', username]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_permissions(self, stdout): permission = 'dist-repo' # when user has permission dict_history = { 'tag_config': [], 'user_perms': [ {'active': True, 'create_event': 20, 'create_ts': 1613730777.437744, 'creator_id': 1, 'creator_name': 'kojiadmin', 'perm_id': 4, 'permission.name': 'dist-repo', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'user.name': 'kojiadmin', 'user_id': 1}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_user_perm(dict_history['user_perms'][0]) anon_handle_list_history(self.options, self.session, ['--permission', permission]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # when user does not have permission dict_history = { 'tag_config': [], 'user_perms': [ {'active': None, 'create_event': 25, 'create_ts': 1613730787.437744, 'creator_id': 1, 'creator_name': 'kojiadmin', 'perm_id': 4, 'permission.name': 'dist-repo', 'revoke_event': 27, 'revoke_ts': 1613730797.797031, 'revoker_id': 1, 'revoker_name': 'kojiadmin', 'user.name': 'kojiadmin', 'user_id': 1}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_user_perm(dict_history['user_perms'][0]) anon_handle_list_history(self.options, self.session, ['--permission', permission]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when perm is not existing expected = "No such entry in table permissions: %s" % permission + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--permission', permission]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_cg(self, stdout): cg = 'test-cg' # when cg is new dict_history = { 'cg_users': [ {'active': True, 'cg_id': 1, 'content_generator.name': 'test-cg', 'create_event': 425, 'create_ts': 1613736494.11504, 'creator_id': 1, 'creator_name': 'kojiadmin', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'user.name': 'kojiadmin', 'user_id': 1}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_cg_user(dict_history['cg_users'][0]) anon_handle_list_history(self.options, self.session, ['--cg', cg]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # when cg is dropped dict_history = { 'cg_users': [ {'active': None, 'cg_id': 2, 'content_generator.name': 'test-cg', 'create_event': 430, 'create_ts': 1613736499.11504, 'creator_id': 1, 'creator_name': 'kojiadmin', 'revoke_event': 440, 'revoke_ts': 1613736510.11504, 'revoker_id': 1, 'revoker_name': 'kojiadmin', 'user.name': 'kojiadmin', 'user_id': 1}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_cg_user(dict_history['cg_users'][0]) anon_handle_list_history(self.options, self.session, ['--cg', cg]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when cg is not existing expected = "No such entry in table content_generator: %s" % cg + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--cg', cg]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_external_repo(self, stdout): dict_history = { 'external_repo_config': [ {'active': True, 'create_event': 342, 'create_ts': 1613736061.68861, 'creator_id': 1, 'creator_name': 'kojiadmin', 'external_repo.name': 'external-repo-test', 'external_repo_id': 5, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'url': 'https://kojipkgs.fedoraproject.org/repos/' 'f32-build/latest/$arch/'}], 'tag_external_repos': [ {'active': True, 'create_event': 343, 'create_ts': 1613736061.72882, 'creator_id': 1, 'creator_name': 'kojiadmin', 'external_repo.name': 'external-repo-test', 'external_repo_id': 5, 'merge_mode': 'koji', 'priority': 5, 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'test-tag', 'tag_id': 9}] } external_repo = 'external-repo-test' self.session.queryHistory.return_value = dict_history expected = self.get_expected_ext_repo( dict_history['external_repo_config'][0]) expected = expected + self.get_expected_tag_ext_repo( dict_history['tag_external_repos'][0]) anon_handle_list_history(self.options, self.session, ['--external-repo', external_repo]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when external repo is not existing expected = "No such entry in table external_repo: " \ "%s" % external_repo + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--external-repo', external_repo]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_build_target(self, stdout): build_target = 'test-build-target' # test when build target is new dict_history = { 'build_target_config': [ {'active': True, 'build_tag': 10, 'build_tag.name': 'test-tag', 'build_target.name': 'test-build-target', 'build_target_id': 5, 'create_event': 420, 'create_ts': 1613736230.7615, 'creator_id': 1, 'creator_name': 'kojiadmin', 'dest_tag': 11, 'dest_tag.name': 'destination-test-tag', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_build_target( dict_history['build_target_config'][0]) anon_handle_list_history(self.options, self.session, ['--build-target', build_target]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test when build target is dropped dict_history = { 'build_target_config': [ {'active': None, 'build_tag': 9, 'build_tag.name': 'test-tag', 'build_target.name': 'test-build-target', 'build_target_id': 4, 'create_event': 417, 'create_ts': 1613736220.7615, 'creator_id': 1, 'creator_name': 'kojiadmin', 'dest_tag': 10, 'dest_tag.name': 'destination-test-tag', 'revoke_event': 420, 'revoke_ts': 1613736225.7615, 'revoker_id': 1, 'revoker_name': 'kojiadmin'}] } self.session.queryHistory.return_value = dict_history expected = self.get_expected_build_target( dict_history['build_target_config'][0]) anon_handle_list_history(self.options, self.session, ['--build-target', build_target]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) self.ensure_connection.reset_mock() # test case when build target is not existing expected = "No such entry in table build_target: " \ "%s" % build_target + "\n" self.session.queryHistory.side_effect = koji.GenericError(expected) with self.assertRaises(koji.GenericError) as ex: anon_handle_list_history(self.options, self.session, ['--build-target', build_target]) self.assertEqual(str(ex.exception), expected) self.ensure_connection.assert_called_once_with(self.session, self.options) @mock.patch('sys.stdout', new_callable=StringIO) def test_list_history_xkey(self, stdout): dict_history = { 'tag_extra': [ {'active': True, 'create_event': 586, 'create_ts': 1613744977.02986, 'creator_id': 1, 'creator_name': 'kojiadmin', 'key': 'extra-key-test', 'revoke_event': None, 'revoke_ts': None, 'revoker_id': None, 'revoker_name': None, 'tag.name': 'tag-test', 'tag_id': 14, 'value': '"extra-value-test"'}] } xkey = 'extra-key-test' self.session.queryHistory.return_value = dict_history expected = self.get_expected_tag_extra(dict_history['tag_extra'][0]) anon_handle_list_history(self.options, self.session, ['--xkey', xkey]) self.assert_console_message(stdout, expected) self.ensure_connection.assert_called_once_with(self.session, self.options) def test_list_history_without_limited_and_all(self): arguments = [] self.assert_system_exit( anon_handle_list_history, self.options, self.session, arguments, stderr=self.format_error_message("Please specify an option to limit the query"), stdout='', activate_session=None, exit_code=2) self.ensure_connection.assert_not_called() def test_list_history_with_arg(self): arguments = ['args'] self.assert_system_exit( anon_handle_list_history, self.options, self.session, arguments, stderr=self.format_error_message("This command takes no arguments"), stdout='', activate_session=None, exit_code=2) self.ensure_connection.assert_not_called() def test_handle_list_history_help(self): self.assert_help( anon_handle_list_history, """Usage: %s list-history [options] (Specify the --help global option for a list of other help options) Options: -h, --help show this help message and exit --build=BUILD Only show data for a specific build --package=PACKAGE Only show data for a specific package --tag=TAG Only show data for a specific tag --editor=USER, --by=USER Only show entries modified by user --user=USER Only show entries affecting a user --permission=PERMISSION Only show entries relating to a given permission --cg=CG Only show entries relating to a given content generator --external-repo=EXTERNAL_REPO, --erepo=EXTERNAL_REPO Only show entries relating to a given external repo --build-target=BUILD_TARGET, --target=BUILD_TARGET Only show entries relating to a given build target --group=GROUP Only show entries relating to a given group --host=HOST Only show entries related to given host --channel=CHANNEL Only show entries related to given channel --xkey=XKEY Only show entries related to given tag extra key --before=BEFORE Only show entries before this time, time is specified as timestamp or date/time in any format which can be parsed by dateutil.parser. e.g. "2020-12-31 12:35" or "December 31st 12:35" --after=AFTER Only show entries after timestamp (same format as for --before) --before-event=EVENT_ID Only show entries before event --after-event=EVENT_ID Only show entries after event --watch Monitor history data --active Only show entries that are currently active --revoked Only show entries that are currently revoked --context Show related entries -s SHOW, --show=SHOW Show data from selected tables -v, --verbose Show more detail -e, --events Show event ids --all Allows listing the entire global history --utc Shows time in UTC timezone """ % self.progname) if __name__ == '__main__': unittest.main()