fm-orchestrator/tests/test_utils/test_ursine.py

# Copyright (c) 2017  Red Hat, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import copy

from mock import patch, Mock

from module_build_service import conf, glib
from module_build_service.utils import ursine
from tests import make_module, clean_database


class TestFindModuleKojiTags:
    """Test ursine.find_module_koji_tags"""

    @patch.object(conf, 'koji_tag_prefixes', new=['module'])
    def test_find_out_all_module_koji_tags(self):
        session = Mock()
        session.getFullInheritance.return_value = [
            {'name': 'module-tag1-s-v-c'},
            {'name': 'module-tag2-s-v-c'},
            {'name': 'tag-1'},
        ]

        expected_tags = ['module-tag1-s-v-c', 'module-tag2-s-v-c']

        tags = ursine.find_module_koji_tags(session, 'tag-a-build')
        assert expected_tags == tags

    @patch.object(conf, 'koji_tag_prefixes', new=['module'])
    def test_return_empty_if_no_module_koji_tags(self):
        session = Mock()
        session.getFullInheritance.return_value = [
            {'name': 'tag-1'}, {'name': 'tag-2'},
        ]

        tags = ursine.find_module_koji_tags(session, 'tag-a-build')
        assert [] == tags


class TestFindUrsineRootTags:
    """Test ursine.find_build_tags_from_external_repos"""

    def setup_method(self):
        self.koji_session = Mock()
        self.koji_session.getTag.side_effect = lambda name: \
            None if name == 'X-build' else {'name': name}

    def test_find_build_tags(self):
        with patch.object(conf, 'koji_external_repo_url_prefix',
                          new='http://example.com/brewroot/'):
            tags = ursine.find_build_tags_from_external_repos(self.koji_session, [
                {
                    'external_repo_name': 'tag-1-external-repo',
                    'url': 'http://example.com/brewroot/repos/tag-1-build/latest/$arch/'
                },
                {
                    'external_repo_name': 'tag-2-external-repo',
                    'url': 'http://example.com/brewroot/repos/tag-2-build/latest/$arch/'
                },
            ])

            assert ['tag-1-build', 'tag-2-build'] == tags

    def test_return_emtpy_if_no_match_external_repo_url(self):
        with patch.object(conf, 'koji_external_repo_url_prefix',
                          new='http://example.com/brewroot/'):
            tags = ursine.find_build_tags_from_external_repos(self.koji_session, [
                {
                    'external_repo_name': 'tag-1-external-repo',
                    'url': 'https://another-site.org/repos/tag-1-build/latest/$arch/'
                },
                {
                    'external_repo_name': 'tag-2-external-repo',
                    'url': 'https://another-site.org/repos/tag-2-build/latest/$arch/'
                },
            ])

            assert [] == tags

    def test_some_tag_is_not_koji_tag(self):
        with patch.object(conf, 'koji_external_repo_url_prefix',
                          new='http://example.com/brewroot/'):
            tags = ursine.find_build_tags_from_external_repos(self.koji_session, [
                {
                    'external_repo_name': 'tag-1-external-repo',
                    'url': 'http://example.com/brewroot/repos/tag-1-build/latest/$arch/'
                },
                {
                    'external_repo_name': 'tag-2-external-repo',
                    'url': 'http://example.com/brewroot/repos/X-build/latest/$arch/'
                },
            ])

            assert ['tag-1-build'] == tags


class TestGetModulemdsFromUrsineContent:
    """Test ursine.get_modulemds_from_ursine_content"""

    def setup_method(self):
        clean_database(False)

    def teardown_method(self, test_method):
        clean_database()

    @patch('module_build_service.builder.KojiModuleBuilder.KojiClientSession')
    def test_return_empty_if_no_ursine_build_tag_is_found(self, ClientSession):
        session = ClientSession.return_value

        # No module koji_tag in ursine content yet. This will result in empty
        # ursine modulemds is returned.
        session.getFullInheritance.return_value = [
            {'name': 'tag-1.0-build'},
        ]
        session.getExternalRepoList.return_value = [
            {
                'external_repo_name': 'tag-1.0-external-repo',
                'url': 'http://example.com/repos/tag-4-build/latest/$arch/'
            }
        ]

        modulemds = ursine.get_modulemds_from_ursine_content('tag')
        assert [] == modulemds

    @patch.object(conf, 'koji_tag_prefixes', new=['module'])
    @patch('module_build_service.builder.KojiModuleBuilder.KojiClientSession')
    def test_get_modulemds(self, ClientSession):
        session = ClientSession.return_value

        # Ensure to to get build tag for further query of ursine content.
        # For this test, the build tag is tag-4-build
        session.getExternalRepoList.return_value = [
            {
                'external_repo_name': 'tag-1.0-external-repo',
                'url': 'http://example.com/repos/tag-4-build/latest/$arch/'
            }
        ]

        # Ensure to return module tags from ursine content of fake build tag
        # specified in above external repo's url.
        def mock_getFullInheritance(tag):
            if tag == 'tag-4-build':
                return [
                    {'name': 'tag-1.0-build'},
                    # Below two modules should be returned and whose modulemd
                    # should be also queried from database.
                    {'name': 'module-name1-s-2020-c'},
                    {'name': 'module-name2-s-2021-c'},
                ]
            raise ValueError('{} is not handled by test.'.format(tag))
        session.getFullInheritance.side_effect = mock_getFullInheritance

        # Defaults to DB resolver, so create fake module builds and store them
        # into database to ensure they can be queried.
        mmd_name1s2020c = make_module(
            'name1:s:2020:c',
            xmd={'mbs': {'koji_tag': 'module-name1-s-2020-c'}})
        mmd_name2s2021c = make_module(
            'name2:s:2021:c',
            xmd={'mbs': {'koji_tag': 'module-name2-s-2021-c'}})

        koji_tag = 'tag'  # It's ok to use arbitrary tag name.
        with patch.object(conf, 'koji_external_repo_url_prefix', new='http://example.com/'):
            modulemds = ursine.get_modulemds_from_ursine_content(koji_tag)

        test_nsvcs = [item.dup_nsvc() for item in modulemds]
        test_nsvcs.sort()

        expected_nsvcs = [mmd_name1s2020c.mmd().dup_nsvc(),
                          mmd_name2s2021c.mmd().dup_nsvc()]
        expected_nsvcs.sort()

        session.getExternalRepoList.assert_called_once_with(koji_tag)
        assert expected_nsvcs == test_nsvcs


class TestRecordStreamCollisionModules:
    """Test ursine.record_stream_collision_modules"""

    @patch.object(conf, 'base_module_names', new=['platform'])
    @patch.object(ursine, 'find_stream_collision_modules')
    def test_nothing_changed_if_no_base_module_is_in_buildrequires(
            self, find_stream_collision_modules):
        xmd = {
            'mbs': {
                'buildrequires': {
                    'modulea': {'stream': 'master'}
                }
            }
        }
        fake_mmd = make_module('name1:s:2020:c', xmd=xmd, store_to_db=False)
        original_xmd = glib.from_variant_dict(fake_mmd.get_xmd())

        with patch.object(ursine, 'log') as log:
            ursine.handle_stream_collision_modules(fake_mmd)
            assert 2 == log.info.call_count
            find_stream_collision_modules.assert_not_called()

        assert original_xmd == glib.from_variant_dict(fake_mmd.get_xmd())

    @patch.object(conf, 'base_module_names', new=['platform'])
    @patch('module_build_service.utils.ursine.get_modulemds_from_ursine_content')
    def test_mark_handled_even_if_no_modules_in_ursine_content(
            self, get_modulemds_from_ursine_content):
        xmd = {
            'mbs': {
                'buildrequires': {
                    'modulea': {'stream': 'master'},
                    'platform': {'stream': 'master', 'koji_tag': 'module-rhel-8.0-build'},
                }
            }
        }
        fake_mmd = make_module('name1:s:2020:c', xmd=xmd, store_to_db=False)
        original_xmd = glib.from_variant_dict(fake_mmd.get_xmd())

        get_modulemds_from_ursine_content.return_value = []

        with patch.object(ursine, 'log') as log:
            ursine.handle_stream_collision_modules(fake_mmd)
            assert 2 == log.info.call_count

        expected_xmd = copy.deepcopy(original_xmd)
        # Ensure stream_collision_modules is set.
        expected_xmd['mbs']['buildrequires']['platform']['stream_collision_modules'] = ''
        expected_xmd['mbs']['buildrequires']['platform']['ursine_rpms'] = ''
        assert expected_xmd == glib.from_variant_dict(fake_mmd.get_xmd())

    @patch.object(conf, 'base_module_names', new=['platform', 'project-platform'])
    @patch('module_build_service.utils.ursine.get_modulemds_from_ursine_content')
    @patch('module_build_service.resolver.GenericResolver.create')
    @patch('module_build_service.builder.KojiModuleBuilder.KojiClientSession')
    def test_add_collision_modules(self, ClientSession, resolver_create,
                                   get_modulemds_from_ursine_content):
        xmd = {
            'mbs': {
                'buildrequires': {
                    'modulea': {'stream': 'master'},
                    'foo': {'stream': '1'},
                    'bar': {'stream': '2'},
                    'platform': {'stream': 'master', 'koji_tag': 'module-rhel-8.0-build'},
                    'project-platform': {
                        'stream': 'master', 'koji_tag': 'module-project-1.0-build'
                    },
                }
            }
        }
        fake_mmd = make_module('name1:s:2020:c',
                               xmd=xmd, store_to_db=False)

        def mock_get_ursine_modulemds(koji_tag):
            if koji_tag == 'module-rhel-8.0-build':
                return [
                    # This is the one
                    make_module('modulea:10:20180813041838:5ea3b708',
                                store_to_db=False),
                    make_module('moduleb:1.0:20180113042038:6ea3b105',
                                store_to_db=False),
                ]
            if koji_tag == 'module-project-1.0-build':
                return [
                    # Both of them are the collided modules
                    make_module('bar:6:20181013041838:817fa3a8',
                                store_to_db=False),
                    make_module('foo:2:20180113041838:95f078a1',
                                store_to_db=False),
                ]

        get_modulemds_from_ursine_content.side_effect = mock_get_ursine_modulemds

        # Mock for finding out built rpms
        def mock_get_module(name, stream, version, context, strict=True):
            return {
                'modulea:10:20180813041838:5ea3b708': {
                    'koji_tag': 'module-modulea-10-20180813041838-5ea3b708',
                },
                'bar:6:20181013041838:817fa3a8': {
                    'koji_tag': 'module-bar-6-20181013041838-817fa3a8',
                },
                'foo:2:20180113041838:95f078a1': {
                    'koji_tag': 'module-foo-2-20180113041838-95f078a1',
                },
            }['{}:{}:{}:{}'.format(name, stream, version, context)]

        resolver = resolver_create.return_value
        resolver._get_module.side_effect = mock_get_module

        def mock_listTaggedRPMS(tag, latest):
            return {
                'module-modulea-10-20180813041838-5ea3b708': [[
                    {'name': 'pkg1', 'version': '1.0', 'release': '1.fc28',
                     'epoch': None},
                ]],
                'module-bar-6-20181013041838-817fa3a8': [[
                    {'name': 'pkg2', 'version': '2.0', 'release': '1.fc28',
                     'epoch': None},
                ]],
                'module-foo-2-20180113041838-95f078a1': [[
                    {'name': 'pkg3', 'version': '3.0', 'release': '1.fc28',
                     'epoch': None},
                ]],
            }[tag]

        koji_session = ClientSession.return_value
        koji_session.listTaggedRPMS.side_effect = mock_listTaggedRPMS

        ursine.handle_stream_collision_modules(fake_mmd)

        xmd = glib.from_variant_dict(fake_mmd.get_xmd())
        buildrequires = xmd['mbs']['buildrequires']

        assert (['modulea:10:20180813041838:5ea3b708'] ==
                buildrequires['platform']['stream_collision_modules'])
        assert (['pkg1-0:1.0-1.fc28'] ==
                buildrequires['platform']['ursine_rpms'])

        modules = sorted(
            buildrequires['project-platform']['stream_collision_modules'])
        expected_modules = ['bar:6:20181013041838:817fa3a8',
                            'foo:2:20180113041838:95f078a1']
        assert expected_modules == modules

        assert (['pkg2-0:2.0-1.fc28', 'pkg3-0:3.0-1.fc28'] ==
                sorted(buildrequires['project-platform']['ursine_rpms']))


class TestFindStreamCollisionModules:
    """Test ursine.find_stream_collision_modules"""

    @patch('module_build_service.utils.ursine.get_modulemds_from_ursine_content')
    def test_no_modulemds_found_from_ursine_content(
            self, get_modulemds_from_ursine_content):
        get_modulemds_from_ursine_content.return_value = []
        assert not ursine.find_stream_collision_modules({}, 'koji_tag')

    @patch('module_build_service.utils.ursine.get_modulemds_from_ursine_content')
    def test_no_collisions_found(self, get_modulemds_from_ursine_content):
        xmd_mbs_buildrequires = {
            'modulea': {'stream': 'master'},
            'moduleb': {'stream': '10'},
        }
        get_modulemds_from_ursine_content.return_value = [
            make_module('moduler:1:1:c1', store_to_db=False),
            make_module('modules:2:1:c2', store_to_db=False),
            make_module('modulet:3:1:c3', store_to_db=False),
        ]
        assert [] == ursine.find_stream_collision_modules(
            xmd_mbs_buildrequires, 'koji_tag')

    @patch('module_build_service.utils.ursine.get_modulemds_from_ursine_content')
    def test_collision_modules_are_found(self, get_modulemds_from_ursine_content):
        xmd_mbs_buildrequires = {
            'modulea': {'stream': 'master'},
            'moduleb': {'stream': '10'},
        }
        fake_modules = [
            make_module('moduler:1:1:c1', store_to_db=False),
            make_module('moduleb:6:1:c2', store_to_db=False),
            make_module('modulet:3:1:c3', store_to_db=False),
        ]
        get_modulemds_from_ursine_content.return_value = fake_modules

        assert [fake_modules[1].dup_nsvc()] == \
            ursine.find_stream_collision_modules(
                xmd_mbs_buildrequires, 'koji_tag')