From c91a6f91bdc8b25aae6b88a1a039031cfe13e94f Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 16 2019 16:39:21 +0000 Subject: [PATCH 1/8] Use Pythonic ways to iterate over dicts Signed-off-by: Nils Philippsen --- diff --git a/robosignatory/tag.py b/robosignatory/tag.py index 4bd2aa4..fb728ed 100644 --- a/robosignatory/tag.py +++ b/robosignatory/tag.py @@ -20,9 +20,7 @@ class TagSigner(object): self.signer = utils.get_signing_helper(**signing_config) self.koji_clients = {} - for instance in self.config['koji_instances']: - instance_info = self.config[ - 'koji_instances'][instance] + for instance, instance_info in self.config['koji_instances'].items(): client = koji.ClientSession(instance_info['url'], instance_info['options']) @@ -144,15 +142,15 @@ class TagSigner(object): sigkey=tag_info['keyid']) log.info('RPMs to sign and move: %s', ['%s (%s, signed: %s)' % - (key, rpms[key]['id'], rpms[key]['signed']) - for key in rpms.keys()]) + (key, rpm['id'], rpm['signed']) + for key, rpm in rpms.items()]) if len(rpms) < 1: log.info('Build contains no rpms, skipping signing and writing') - if all([rpms[rpm]['signed'] for rpm in rpms]) or len(rpms) < 1: + if all([rpm['signed'] for rpm in rpms.values()]) or len(rpms) < 1: log.debug('All RPMs are already signed') else: - to_sign = [key for key in rpms.keys() if not rpms[key]['signed']] + to_sign = [key for key, rpm in rpms.items() if not rpm['signed']] log.debug('RPMs needing signing: %s' % to_sign) cmdline = self.signer.build_sign_cmdline(tag_info['key'], to_sign, @@ -169,8 +167,7 @@ class TagSigner(object): log.info('Build was succesfully signed, telling koji to write with key' ' %s', tag_info['keyid']) - for rpm in rpms: - instance['client'].writeSignedRPM(rpms[rpm]['id'], - tag_info['keyid']) + for rpm in rpms.values(): + instance['client'].writeSignedRPM(rpm['id'], tag_info['keyid']) log.info('Signed RPMs written out') From e49fab367be9705264eac1cf760c12c295061d49 Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 16 2019 16:39:52 +0000 Subject: [PATCH 2/8] Add missing % to interpolate exception message Signed-off-by: Nils Philippsen --- diff --git a/robosignatory/tag.py b/robosignatory/tag.py index fb728ed..1e812d1 100644 --- a/robosignatory/tag.py +++ b/robosignatory/tag.py @@ -52,7 +52,7 @@ class TagSigner(object): if tag_type is None: tag_type = 'plain' elif tag_type not in KNOWN_TAG_TYPES: - raise Exception('Invalid tag type detected: %s', tag_type) + raise Exception('Invalid tag type detected: %s' % tag_type) instance_obj['tags'][tag['from']]['type'] = tag_type self.koji_clients[instance] = instance_obj From c26993e2433e6278f4e6344bd8e7fb89c3ecd27f Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 16 2019 16:41:37 +0000 Subject: [PATCH 3/8] Remove unused variable Signed-off-by: Nils Philippsen --- diff --git a/robosignatory/tag.py b/robosignatory/tag.py index 1e812d1..6d2f06f 100644 --- a/robosignatory/tag.py +++ b/robosignatory/tag.py @@ -82,7 +82,6 @@ class TagSigner(object): log.info('Koji instance not known, skipping') return - instance = self.koji_clients[koji_instance] self.dowork(build_nvr, build_id, tag, koji_instance, skip_tagging=False) From be575408138bfb7a6cef3573ae26161e8ab1a53e Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 16 2019 16:42:14 +0000 Subject: [PATCH 4/8] Fix typo Signed-off-by: Nils Philippsen --- diff --git a/robosignatory/tag.py b/robosignatory/tag.py index 6d2f06f..97093a0 100644 --- a/robosignatory/tag.py +++ b/robosignatory/tag.py @@ -163,7 +163,7 @@ class TagSigner(object): raise Exception('Signing failed') if len(rpms) > 1: - log.info('Build was succesfully signed, telling koji to write with key' + log.info('Build was successfully signed, telling koji to write with key' ' %s', tag_info['keyid']) for rpm in rpms.values(): From 7f5cae29e4f0e77d760258798b0b702f8efcf9e3 Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 16 2019 16:42:40 +0000 Subject: [PATCH 5/8] Migrate to pytest There isn't much nose-specific in the project yet, and it doesn't seem to be actively developed upstream. See: https://nose.readthedocs.io/en/latest/#note-to-users Pytest has many useful features like log capturing, easy parametrizing of tests, so let's use it instead. Signed-off-by: Nils Philippsen --- diff --git a/setup.py b/setup.py index 5ced7b1..5f45871 100644 --- a/setup.py +++ b/setup.py @@ -17,7 +17,7 @@ setup( "click", ], tests_require=[ - "nose", + "pytest", "mock", ], packages=[ diff --git a/tests/test_utils.py b/tests/test_utils.py index a262750..c07b898 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,14 +1,14 @@ import unittest -from nose.tools import raises, eq_ +from pytest import raises import robosignatory.utils class TestUtils(unittest.TestCase): - @raises(KeyError) def test_no_such_helper(self): - robosignatory.utils.get_signing_helper(backend='wat', ) + with raises(KeyError): + robosignatory.utils.get_signing_helper(backend='wat', ) def test_get_sigul_helper(self): helper = robosignatory.utils.get_signing_helper( @@ -16,7 +16,7 @@ class TestUtils(unittest.TestCase): user='ralph', passphrase_file='/tmp/wide-open.txt', ) - eq_(type(helper), robosignatory.utils.SigulHelper) + assert type(helper) == robosignatory.utils.SigulHelper def test_simple_echo_helper(self): helper = robosignatory.utils.get_signing_helper( @@ -25,4 +25,4 @@ class TestUtils(unittest.TestCase): passphrase_file='/tmp/wide-open.txt', ) cmdline = helper.build_cmdline('wat') - eq_(cmdline, ["echo", "build_cmdline: ('wat',) {}"]) + assert cmdline == ["echo", "build_cmdline: ('wat',) {}"] From 83c4d54122d91889fdfc7556ceadcda26d6e9a26 Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 21 2019 19:01:55 +0000 Subject: [PATCH 6/8] Test most of the TagSigner functionality Signed-off-by: Nils Philippsen --- diff --git a/tests/test_tag.py b/tests/test_tag.py new file mode 100644 index 0000000..dfb6476 --- /dev/null +++ b/tests/test_tag.py @@ -0,0 +1,315 @@ +import copy +import logging +import mock + +import _pytest.logging +from fedora_messaging.api import Message +from pytest import raises, mark + +from robosignatory.tag import TagSigner + + +if not hasattr(_pytest.logging.LogCaptureFixture, 'messages'): # noqa + # monkey-patch missing messages property + class MyLogCaptureFixture(_pytest.logging.LogCaptureFixture): + @property + def messages(self): + return [r.getMessage() for r in self.records] + + +TEST_CONFIG = { + 'signing': { + 'backend': 'echo', + }, + 'koji_instances': { + 'test': { + 'url': 'https://koji.example.com', + 'mbs_user': 'mbs_user', + 'options': { + 'authmethod': 'kerberos', + 'principal': 'test@EXAMPLE.COM', + }, + 'tags': [ + { + 'from': 'f31-pending', + 'to': 'f31', + 'key': 'fedora-31', + 'keyid': 'deadbeef', + }, + { + 'from': 'f30-signing-pending', + 'to': 'f30-updates-testing-pending', + 'key': 'fedora-30', + 'keyid': 'OU812I81B4U', + 'type': 'plain', + }, + { + 'from': 'f30-modular-signing-pending', + 'to': 'f30-modular-updates-testing-pending', + 'key': 'fedora-30', + 'keyid': 'OU812I81B4U', + 'type': 'modular', + } + ], + }, + }, + 'ostree_refs': {}, + 'coreos': { + 'bucket': 'testing', + 'key': 'testing', + 'aws': { + 'access_key': 'testing', + 'access_secret': 'testing', + 'region': 'us-east-1', + } + }, +} + + +class MockUtils(mock.MagicMock): + + builds = [ + {'id': 1, + 'nvr': 'foo-1-1.fc31', + 'rpms': [ + {'id': 100, + 'nvr': 'foo-1-1.fc31', + 'arch': 'x64_64', + 'sigkey': '1234'}, + {'id': 101, + 'nvr': 'foo-libs-1-1.fc31', + 'arch': 'x64_64', + 'sigkey': '1234'}, + ]}, + {'id': 2, + 'nvr': 'bar-0.11-1.fc31', + 'rpms': [ + {'id': 200, + 'nvr': 'bar-0.11-1.fc31', + 'arch': 'x64_64', + 'sigkey': '2345'}, + {'id': 201, + 'nvr': 'bar-docs-0.11-1.fc31', + 'arch': 'noarch', + 'sigkey': '2345'}, + ]}, + {'id': 3, + 'nvr': 'gnu-10.0-5.fc31', + 'rpms': [ + {'id': 300, + 'nvr': 'gnu-10.0-5.fc31', + 'arch': 'x64_64'}, + ]}, + ] + + def _get_build(self, build_id): + for b in self.builds: + if b['id'] == build_id: + return b + + def get_rpms(self, koji_client, build_nvr, build_id, sigkey=None): + rpminfo = {} + for rpm in self._get_build(build_id)['rpms']: + info = {'id': rpm['id']} + if sigkey: + info['signed'] = rpm['sigkey'] == sigkey + rpminfo['{}.{}'.format(rpm['nvr'], rpm['arch'])] = info + return rpminfo + + def get_signing_helper(self, **signing_config): + return mock.MagicMock(signing_config=signing_config) + + def run_command(self, cmdline): + return 0, "", "" + + +class DummyContext(object): + + def __enter__(self): + pass + + def __exit__(self, *args): + pass + + +@mock.patch( + 'robosignatory.consumer.fedora_messaging.config.conf', + {'consumer_config': TEST_CONFIG} +) +class TestTagSigner(object): + """Test the Koji tag signer class""" + + test_msg = { + 'topic': 'org.fedoraproject.prod.buildsys.tag', + 'body': { + 'name': 'foo', 'version': '1', 'release': '1.fc31', + 'build_id': 1, 'tag': 'f31-pending', + 'instance': 'test', + }, + } + + def setup_method(self, method): + # Patch in a mock koji session. + with mock.patch('robosignatory.tag.koji.ClientSession'): + self.tag_signer = TagSigner(config=TEST_CONFIG) + + # Ensure methods can muck around with the test message contents and not + # disturb each other. + self.test_msg = copy.deepcopy(type(self).test_msg) + + @property + def instance_obj(self): + return self.tag_signer.koji_clients['test'] + + @property + def koji_client(self): + return self.instance_obj['client'] + + # Test creating the tag signer object with various (working & non-working) + # configurations. + + def test_init(self): + """Test that the tag signer can be created""" + # The signer gets created in .setUp(). + + assert self.instance_obj['mbs_user'] == 'mbs_user' + + self.koji_client.krb_login.assert_called_once_with( + principal='test@EXAMPLE.COM') + self.koji_client.ssl_login.assert_not_called() + + def test_init_with_ssl(self): + """Test that a signer using SSL authentication can be created""" + # We overwrite the signer here to inject non-standard configuration. + test_config = copy.deepcopy(TEST_CONFIG) + test_config['koji_instances']['test']['options'] = { + 'authmethod': 'ssl', 'cert': 'cert', 'serverca': 'serverca', + } + with mock.patch('robosignatory.tag.koji.ClientSession'): + self.tag_signer = TagSigner(test_config) + + self.koji_client.ssl_login.assert_called_once_with( + 'cert', None, 'serverca') + self.koji_client.krb_login.assert_not_called() + + def test_init_with_unknown_authmethod(self): + """Test that unknown authmethods are caught""" + # We overwrite the signer here to inject non-standard configuration. + test_config = copy.deepcopy(TEST_CONFIG) + test_config['koji_instances']['test']['options'] = { + 'authmethod': 'unknown' + } + with mock.patch( + 'robosignatory.tag.koji.ClientSession' + ) as koji_client, raises( + Exception, match='Only SSL and kerberos authmethods supported' + ): + self.tag_signer = TagSigner(test_config) + + koji_client.ssl_login.assert_not_called() + koji_client.krb_login.assert_not_called() + + def test_init_with_duplicate_from_tag(self): + """Test behavior with duplicate from tag""" + # We overwrite the signer here to inject non-standard configuration. + test_config = copy.deepcopy(TEST_CONFIG) + tags = test_config['koji_instances']['test']['tags'] + tags.insert(1, tags[0]) + with mock.patch( + 'robosignatory.tag.koji.ClientSession' + ), raises( + Exception, + match='From detected twice: {}'.format(tags[0]['from']) + ): + self.tag_signer = TagSigner(test_config) + + def test_init_with_unknown_tag_type(self): + """Test behavior unknown tag type""" + # We overwrite the signer here to inject non-standard configuration. + test_config = copy.deepcopy(TEST_CONFIG) + tags = test_config['koji_instances']['test']['tags'] + tags[0]['type'] = 'unknown' + with mock.patch( + 'robosignatory.tag.koji.ClientSession' + ), raises( + Exception, match='Invalid tag type detected: unknown' + ): + self.tag_signer = TagSigner(test_config) + + # Test well-formed messages. + + @mock.patch('robosignatory.tag.utils', new_callable=MockUtils()) + @mark.parametrize('type_,error', + ((None, None), ('plain', None), ('modular', None), + ('modular', 'non-mbs-owner'))) + def test_build_messages(self, utils, caplog, type_, error): + """Test build messages for various types.""" + caplog.set_level(logging.DEBUG) + + body = self.test_msg['body'] + instance = body['instance'] + build_nvr = '{name}-{version}-{release}'.format(**body) + from_tag = body['tag'] + build_id = body['build_id'] + tag_conf = self.instance_obj['tags'][from_tag] + to_tag = tag_conf['to'] + + if type_ == 'modular': + body['tag'] = from_tag = 'f30-modular-signing-pending' + to_tag = 'f30-modular-updates-testing-pending' + if error == 'non-mbs-owner': + build_owner = 'hamburglar' + else: + build_owner = TEST_CONFIG['koji_instances']['test']['mbs_user'] + + self.koji_client.listTagged.return_value = [ + {'owner_name': build_owner, 'nvr': build_nvr, 'build_id': build_id} + ] + + expected_log_msgs = [ + "Build {} ({}) tagged into {} on {}".format( + build_nvr, build_id, from_tag, instance), + ] + + if error == 'non-mbs-owner': + expected_log_msgs.append( + "Build {} has owner {}, which is NOT mbs_user!".format(build_id, build_owner)) + exc_ctx_mgr = raises(Exception, + match="Modular content tag contains invalid owned build") + else: + exc_ctx_mgr = DummyContext() + msg = Message(**self.test_msg) + with exc_ctx_mgr: + self.tag_signer.consume(msg) + + for msg in expected_log_msgs: + assert msg in caplog.messages + + if not error: + self.koji_client.tagBuild.assert_called_once_with(to_tag, build_id, + False, from_tag) + else: + self.koji_client.tagBuild.assert_not_called() + + def test_plain_build_message_with_unconfigured_tag(self, caplog): + """Test the behavior with an unconfigured tag""" + caplog.set_level(logging.DEBUG) + + self.test_msg['body']['tag'] = 'unconfigured' + msg = Message(**self.test_msg) + self.tag_signer.consume(msg) + + assert "Tag not autosigned, skipping" in caplog.messages + + self.koji_client.tagBuild.assert_not_called() + + # Test invalid messages. + + @mock.patch('robosignatory.tag.log') + def test_unknown_koji_instance(self, log): + """Test an unknown Koji instance in the message""" + self.test_msg['body']['instance'] = 'unknown' + msg = Message(**self.test_msg) + self.tag_signer.consume(msg) + + log.info.assert_called_with('Koji instance not known, skipping') From ac24a0d588347d941f0751e2b497f2d9df036a1f Mon Sep 17 00:00:00 2001 From: Aurélien Bompard Date: Aug 26 2019 12:57:14 +0000 Subject: [PATCH 7/8] Use pytest when testing with tox Signed-off-by: Nils Philippsen --- diff --git a/tox.ini b/tox.ini index 32ff2e5..e495a6d 100644 --- a/tox.ini +++ b/tox.ini @@ -4,5 +4,8 @@ envlist = py27,py37,py38 skip_missing_interpreters = True [testenv] +deps = + pytest + mock commands = - python setup.py test + python -m pytest -v From 85ba8f67fa63df3c796f65aeee8c71497b552f57 Mon Sep 17 00:00:00 2001 From: Nils Philippsen Date: Aug 26 2019 14:46:32 +0000 Subject: [PATCH 8/8] Integrate pytest with setuptools Signed-off-by: Nils Philippsen --- diff --git a/setup.py b/setup.py index 5f45871..303835d 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,32 @@ +import distutils + from setuptools import setup +from setuptools.command.test import test + + +class PyTest(test): + user_options = [('pytest-args=', 'a', "Arguments to pass to pytest")] + + def initialize_options(self): + test.initialize_options(self) + self.pytest_args = [] + + def finalize_options(self): + test.finalize_options(self) + self.ensure_string_list('pytest_args') + self.test_args = [] + self.test_suite = True + + def run_tests(self): + # import late because pytest is only required for testing + import pytest + + exitcode = pytest.main(self.pytest_args) + if exitcode: + msg = 'pytest failed!' + self.announce(msg, distutils.log.ERROR) + raise distutils.errors.DistutilsError(msg) + setup( name='robosignatory', @@ -31,4 +59,7 @@ setup( echo = robosignatory.utils:EchoHelper sigul = robosignatory.utils:SigulHelper """, + cmdclass={ + 'test': PyTest, + }, )