pungi/tests/test_pkgset_pkgsets.py
Lubomír Sedlář 2bb65408ee [pkgset] Add tests for KojiPackageSet
There is a lot of mock objects needed: we bypass calls to Koji, use a
mock FileCache that does not need valid RPMs on disk and avoid any
multithreading.

The test data in tests/fixtures/tagged-rpms.json comes from Koji. It is
filtered down to only a few packages to make it manageable.

Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
2016-04-22 13:35:17 +02:00

220 lines
7.0 KiB
Python
Executable File

#!/usr/bin/env python2
# -*- coding: utf-8 -*-
import mock
import os
import sys
import unittest
import json
import functools
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from pungi.phases.pkgset import pkgsets
from tests import helpers
class MockPathInfo(object):
def __init__(self, topdir):
self.topdir = topdir
def build(self, build_info):
return self.topdir
def get_filename(self, rpm_info):
return '{name}@{version}@{release}@{arch}'.format(**rpm_info)
def signed(self, rpm_info, sigkey):
return os.path.join('signed', sigkey, self.get_filename(rpm_info))
def rpm(self, rpm_info):
return os.path.join('rpms', self.get_filename(rpm_info))
@functools.total_ordering
class MockFile(object):
def __init__(self, path):
if path.startswith('/tmp'):
# Drop /tmp/something/ from path
path = path.split('/', 3)[-1]
self.path = path
self.file_name = os.path.basename(path)
self.name, self.version, self.release, self.arch = self.file_name.split('@')
self.sourcerpm = self.name
def __hash__(self):
return hash(self.path)
def __repr__(self):
return self.path
def __eq__(self, other):
try:
return self.path == other.path
except AttributeError:
return self.path == other
def __le__(self, other):
try:
return self.path < other.path
except AttributeError:
return self.path < other
class MockFileCache(dict):
"""Mock for kobo.pkgset.FileCache.
It gets data from filename and does not touch filesystem.
"""
def __init__(self, _wrapper):
super(MockFileCache, self).__init__()
def add(self, file_path):
obj = MockFile(file_path)
self[file_path] = obj
return obj
class FakePool(object):
"""This class will be substituted for ReaderPool.
It implements the same interface, but uses only the last added worker to
process all tasks sequentially.
"""
def __init__(self, package_set, logger=None):
self.queue = []
self.worker = None
self.package_set = package_set
def log_warning(self, *args, **kwargs):
pass
@property
def queue_total(self):
return len(self.queue)
def queue_put(self, item):
self.queue.append(item)
def add(self, worker):
self.worker = worker
def start(self):
for i, item in enumerate(self.queue):
self.worker.process(item, i)
def stop(self):
pass
@mock.patch('pungi.phases.pkgset.pkgsets.ReaderPool', new=FakePool)
@mock.patch('kobo.pkgset.FileCache', new=MockFileCache)
class TestKojiPkgset(helpers.PungiTestCase):
def setUp(self):
super(TestKojiPkgset, self).setUp()
with open(os.path.join(helpers.FIXTURE_DIR, 'tagged-rpms.json')) as f:
self.tagged_rpms = json.load(f)
self.path_info = MockPathInfo(self.topdir)
self.koji_wrapper = mock.Mock()
self.koji_wrapper.koji_proxy.listTaggedRPMS.return_value = self.tagged_rpms
self.koji_wrapper.koji_module.pathinfo = self.path_info
def _touch_files(self, filenames):
for filename in filenames:
helpers.touch(os.path.join(self.topdir, filename))
def assertPkgsetEqual(self, actual, expected):
for k, v1 in expected.iteritems():
self.assertIn(k, actual)
v2 = actual.pop(k)
self.assertItemsEqual(v1, v2)
self.assertEqual({}, actual, msg='Some architectures were missing')
def test_all_arches(self):
self._touch_files([
'rpms/pungi@4.1.3@3.fc25@noarch',
'rpms/pungi@4.1.3@3.fc25@src',
'rpms/bash@4.3.42@4.fc24@i686',
'rpms/bash@4.3.42@4.fc24@x86_64',
'rpms/bash@4.3.42@4.fc24@src',
'rpms/bash-debuginfo@4.3.42@4.fc24@i686',
'rpms/bash-debuginfo@4.3.42@4.fc24@x86_64',
])
pkgset = pkgsets.KojiPackageSet(self.koji_wrapper, [None])
result = pkgset.populate('f25')
self.assertEqual(
self.koji_wrapper.koji_proxy.mock_calls,
[mock.call.listTaggedRPMS('f25', event=None, inherit=True, latest=True)])
self.assertPkgsetEqual(result,
{'src': ['rpms/pungi@4.1.3@3.fc25@src',
'rpms/bash@4.3.42@4.fc24@src'],
'noarch': ['rpms/pungi@4.1.3@3.fc25@noarch'],
'i686': ['rpms/bash@4.3.42@4.fc24@i686',
'rpms/bash-debuginfo@4.3.42@4.fc24@i686'],
'x86_64': ['rpms/bash@4.3.42@4.fc24@x86_64',
'rpms/bash-debuginfo@4.3.42@4.fc24@x86_64']})
def test_only_one_arch(self):
self._touch_files([
'rpms/bash@4.3.42@4.fc24@x86_64',
'rpms/bash-debuginfo@4.3.42@4.fc24@x86_64',
])
pkgset = pkgsets.KojiPackageSet(self.koji_wrapper, [None], arches=['x86_64'])
result = pkgset.populate('f25')
self.assertEqual(
self.koji_wrapper.koji_proxy.mock_calls,
[mock.call.listTaggedRPMS('f25', event=None, inherit=True, latest=True)])
self.assertPkgsetEqual(result,
{'x86_64': ['rpms/bash-debuginfo@4.3.42@4.fc24@x86_64',
'rpms/bash@4.3.42@4.fc24@x86_64']})
def test_find_signed_with_preference(self):
self._touch_files([
'signed/cafebabe/bash@4.3.42@4.fc24@x86_64',
'signed/deadbeef/bash@4.3.42@4.fc24@x86_64',
'signed/deadbeef/bash-debuginfo@4.3.42@4.fc24@x86_64',
])
pkgset = pkgsets.KojiPackageSet(self.koji_wrapper, ['cafebabe', 'deadbeef'], arches=['x86_64'])
result = pkgset.populate('f25')
self.assertEqual(
self.koji_wrapper.koji_proxy.mock_calls,
[mock.call.listTaggedRPMS('f25', event=None, inherit=True, latest=True)])
self.assertPkgsetEqual(result,
{'x86_64': ['signed/cafebabe/bash@4.3.42@4.fc24@x86_64',
'signed/deadbeef/bash-debuginfo@4.3.42@4.fc24@x86_64']})
def test_find_signed_fallback_unsigned(self):
self._touch_files([
'signed/cafebabe/bash@4.3.42@4.fc24@x86_64',
'rpms/bash-debuginfo@4.3.42@4.fc24@x86_64',
])
pkgset = pkgsets.KojiPackageSet(self.koji_wrapper, ['cafebabe', None], arches=['x86_64'])
result = pkgset.populate('f25')
self.assertEqual(
self.koji_wrapper.koji_proxy.mock_calls,
[mock.call.listTaggedRPMS('f25', event=None, inherit=True, latest=True)])
self.assertPkgsetEqual(result,
{'x86_64': ['rpms/bash-debuginfo@4.3.42@4.fc24@x86_64',
'signed/cafebabe/bash@4.3.42@4.fc24@x86_64']})
if __name__ == "__main__":
unittest.main()