# -*- coding: utf-8 -*-
# Copyright (C) 2012-2018 Red Hat, Inc.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# the GNU General Public License v.2, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY expressed or implied, including the implied warranties of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
# Public License for more details. You should have received a copy of the
# GNU General Public License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA. Any Red Hat trademarks that are incorporated in the
# source code or documentation are not subject to the GNU General Public
# License and may only be used or replicated with the express permission of
# Red Hat, Inc.
#
from __future__ import absolute_import
from __future__ import unicode_literals
import contextlib
import logging
import os
import re
import shutil
import sys
import tempfile
import unittest
from functools import reduce
import hawkey
import hawkey.test
import libdnf.transaction
import dnf
import dnf.conf
import dnf.cli.cli
import dnf.cli.demand
import dnf.cli.option_parser
import dnf.comps
import dnf.exceptions
import dnf.goal
import dnf.i18n
import dnf.package
import dnf.persistor
import dnf.pycomp
import dnf.repo
import dnf.sack
if dnf.pycomp.PY3:
from unittest import mock
from unittest.mock import MagicMock, mock_open
else:
from tests import mock
from tests.mock import MagicMock
def mock_open(mock=None, data=None):
if mock is None:
mock = MagicMock(spec=file)
handle = MagicMock(spec=file)
handle.write.return_value = None
if data is None:
handle.__enter__.return_value = handle
else:
handle.__enter__.return_value = data
mock.return_value = handle
return mock
logger = logging.getLogger('dnf')
skip = unittest.skip
TRACEBACK_RE = re.compile(
r'(Traceback \(most recent call last\):\n'
r'(?: File "[^"\n]+", line \d+, in \w+\n'
r'(?: .+\n)?)+'
r'\S.*\n)')
REASONS = {
'hole': 'group',
'pepper': 'group',
'right': 'dep',
'tour': 'group',
'trampoline': 'group',
}
# @System.repo doesn't provide sha1header/pkgid
# the checksum is computed from an empty string
RPMDB_CHECKSUM = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
TOTAL_RPMDB_COUNT = 10
SYSTEM_NSOLVABLES = TOTAL_RPMDB_COUNT
MAIN_NSOLVABLES = 9
UPDATES_NSOLVABLES = 4
AVAILABLE_NSOLVABLES = MAIN_NSOLVABLES + UPDATES_NSOLVABLES
TOTAL_GROUPS = 5
TOTAL_NSOLVABLES = SYSTEM_NSOLVABLES + AVAILABLE_NSOLVABLES
# testing infrastructure
def dnf_toplevel():
return os.path.normpath(os.path.join(__file__, '../../'))
def repo(reponame):
return os.path.join(REPO_DIR, reponame)
def resource_path(path):
this_dir = os.path.dirname(__file__)
return os.path.join(this_dir, path)
REPO_DIR = resource_path('repos')
COMPS_PATH = os.path.join(REPO_DIR, 'main_comps.xml')
NONEXISTENT_FILE = resource_path('does-not/exist')
TOUR_44_PKG_PATH = resource_path('repos/rpm/tour-4-4.noarch.rpm')
TOUR_50_PKG_PATH = resource_path('repos/rpm/tour-5-0.noarch.rpm')
TOUR_51_PKG_PATH = resource_path('repos/rpm/tour-5-1.noarch.rpm')
USER_RUNDIR = '/tmp/dnf-user-rundir'
# often used query
def installed_but(sack, *args):
q = sack.query().filter(reponame__eq=hawkey.SYSTEM_REPO_NAME)
return reduce(lambda query, name: query.filter(name__neq=name), args, q)
# patching the stdout
@contextlib.contextmanager
def patch_std_streams():
with mock.patch('sys.stdout', new_callable=dnf.pycomp.StringIO) as stdout, \
mock.patch('sys.stderr', new_callable=dnf.pycomp.StringIO) as stderr:
yield (stdout, stderr)
@contextlib.contextmanager
def wiretap_logs(logger_name, level, stream):
"""Record *logger_name* logs of at least *level* into the *stream*."""
logger = logging.getLogger(logger_name)
orig_level = logger.level
logger.setLevel(level)
handler = logging.StreamHandler(stream)
orig_handlers = logger.handlers
logger.handlers = []
logger.addHandler(handler)
try:
yield stream
finally:
logger.removeHandler(handler)
logger.setLevel(orig_level)
logger.handlers = orig_handlers
def command_configure(cmd, args):
parser = dnf.cli.option_parser.OptionParser()
args = [cmd._basecmd] + args
parser.parse_main_args(args)
parser.parse_command_args(cmd, args)
return cmd.configure()
def command_run(cmd, args):
command_configure(cmd, args)
return cmd.run()
class Base(dnf.Base):
def __init__(self, *args, **kwargs):
with mock.patch('dnf.rpm.detect_releasever', return_value=69):
super(Base, self).__init__(*args, **kwargs)
# mock objects
def mock_comps(history, seed_history):
comps = dnf.comps.Comps()
comps._add_from_xml_filename(COMPS_PATH)
if seed_history:
name = 'Peppers'
pkg_types = dnf.comps.MANDATORY
swdb_group = history.group.new(name, name, name, pkg_types)
for pkg_name in ['hole', 'lotus']:
swdb_group.addPackage(pkg_name, True, dnf.comps.MANDATORY)
history.group.install(swdb_group)
name = 'somerset'
pkg_types = dnf.comps.MANDATORY
swdb_group = history.group.new(name, name, name, pkg_types)
for pkg_name in ['pepper', 'trampoline', 'lotus']:
swdb_group.addPackage(pkg_name, True, dnf.comps.MANDATORY)
history.group.install(swdb_group)
name = 'sugar-desktop-environment'
pkg_types = dnf.comps.ALL_TYPES
swdb_env = history.env.new(name, name, name, pkg_types)
for group_id in ['Peppers', 'somerset']:
swdb_env.addGroup(group_id, True, dnf.comps.MANDATORY)
history.env.install(swdb_env)
return comps
def mock_logger():
return mock.create_autospec(logger)
class _BaseStubMixin(object):
"""A reusable class for creating `dnf.Base` stubs.
See also: hawkey/test/python/__init__.py.
Note that currently the used TestSack has always architecture set to
"x86_64". This is to get the same behavior when running unit tests on
different arches.
"""
def __init__(self, *extra_repos, **config_opts):
super(_BaseStubMixin, self).__init__(FakeConf(**config_opts))
for r in extra_repos:
repo = MockRepo(r, self.conf)
repo.enable()
self._repos.add(repo)
self._repo_persistor = FakePersistor()
self._ds_callback = mock.Mock()
self._history = None
self._closed = False
self._closing = False
def add_test_dir_repo(self, id_, cachedir):
"""Add a repository located in a directory in the tests."""
repo = dnf.repo.Repo(id_, cachedir)
repo.baseurl = ['file://%s/%s' % (REPO_DIR, repo.id)]
self.repos.add(repo)
return repo
def close(self):
self._closing = True
super(_BaseStubMixin, self).close()
@property
def history(self):
if self._history:
return self._history
else:
self._history = super(_BaseStubMixin, self).history
if not self._closing:
# don't reset db on close, it causes several tests to fail
self._history.reset_db()
return self._history
@property
def sack(self):
if self._sack:
return self._sack
return self.init_sack()
def _build_comps_solver(self):
return dnf.comps.Solver(self.history, self._comps,
REASONS.get)
def _activate_persistor(self):
pass
def init_sack(self):
# Create the Sack, tell it how to build packages, passing in the Package
# class and a Base reference.
self._sack = TestSack(REPO_DIR, self)
self._sack.load_system_repo()
for repo in self.repos.iter_enabled():
if repo.__class__ is dnf.repo.Repo:
self._add_repo_to_sack(repo)
else:
fn = "%s.repo" % repo.id
self._sack.load_test_repo(repo.id, fn)
self._sack._configure(self.conf.installonlypkgs)
self._goal = dnf.goal.Goal(self._sack)
return self._sack
def mock_cli(self):
stream = dnf.pycomp.StringIO()
logger = logging.getLogger('test')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler(stream))
return mock.Mock(base=self, log_stream=stream, logger=logger,
demands=dnf.cli.demand.DemandSheet())
def read_mock_comps(self, seed_history=True):
self._comps = mock_comps(self.history, seed_history)
return self._comps
def read_all_repos(self, opts=None):
for repo in self.repos.values():
repo._configure_from_options(opts)
def set_debuglevel(self, level):
self.conf._set_value('debuglevel', level, dnf.conf.PRIO_RUNTIME)
class BaseCliStub(_BaseStubMixin, dnf.cli.cli.BaseCli):
"""A class mocking `dnf.cli.cli.BaseCli`."""
def __init__(self, *extra_repos, **config_opts):
"""Initialize the base."""
super(BaseCliStub, self).__init__(*extra_repos, **config_opts)
self.output.term = MockTerminal()
class DemandsStub(object):
pass
class CliStub(object):
"""A class mocking `dnf.cli.Cli`."""
def __init__(self, base):
"""Initialize the CLI."""
self.base = base
self.cli_commands = {}
self.demands = DemandsStub()
self.logger = logging.getLogger()
self.register_command(dnf.cli.commands.HelpCommand)
def redirect_logger(self, stdout=None, stderr=None):
return
def redirect_repo_progress(self, fo=sys.stderr):
return
def register_command(self, command):
"""Register given *command*."""
self.cli_commands.update({alias: command for alias in command.aliases})
class MockOutput(object):
def __init__(self):
self.term = MockTerminal()
def setup_progress_callbacks(self):
return (None, None)
class MockPackage(object):
def __init__(self, nevra, repo=None):
self.baseurl = None
self._chksum = (None, None)
self.downloadsize = None
self._header = None
self.location = '%s.rpm' % nevra
self.repo = repo
self.reponame = None if repo is None else repo.id
self.str = nevra
self.buildtime = 0
nevra = hawkey.split_nevra(nevra)
self.name = nevra.name
self.epoch = nevra.epoch
self.version = nevra.version
self.release = nevra.release
self.arch = nevra.arch
self.evr = '%(epoch)d:%(version)s-%(release)s' % vars(self)
self.pkgtup = (self.name, self.arch, str(self.epoch), self.version,
self.release)
def __str__(self):
return self.str
def localPkg(self):
return os.path.join(self.repo.pkgdir, os.path.basename(self.location))
def returnIdSum(self):
return self._chksum
class MockRepo(dnf.repo.Repo):
def _valid(self):
return None
class MockQuery(dnf.query.Query):
def __init__(self, query):
self.pkgs = [MockPackage(str(p)) for p in query.run()]
self.i = 0
self.n = len(self.pkgs)
def __getitem__(self, key):
if key < self.n:
return self.pkgs[key]
else:
raise KeyError()
def __iter__(self):
return self
def __len__(self):
return self.n
def filter(self, pkg):
self.pkgs = []
self.pkgs.extend(pkg)
self.n = len(self.pkgs)
return self
def next(self):
return self.__next__()
def __next__(self):
if self.i < self.n:
i = self.i
self.i += 1
return self.pkgs[i]
else:
raise StopIteration()
def run(self):
return self.pkgs
class MockTerminal(object):
def __init__(self):
self.MODE = {'bold': '', 'normal': ''}
self.columns = 80
self.real_columns = 80
self.reinit = mock.Mock()
def bold(self, s):
return s
class TestSack(hawkey.test.TestSackMixin, dnf.sack.Sack):
def __init__(self, repo_dir, base):
hawkey.test.TestSackMixin.__init__(self, repo_dir)
dnf.sack.Sack.__init__(self,
arch=hawkey.test.FIXED_ARCH,
pkgcls=dnf.package.Package,
pkginitval=base,
make_cache_dir=True)
class MockBase(_BaseStubMixin, Base):
"""A class mocking `dnf.Base`."""
def mock_sack(*extra_repos):
return MockBase(*extra_repos).sack
class FakeConf(dnf.conf.Conf):
def __init__(self, **kwargs):
super(FakeConf, self).__init__()
self.substitutions['releasever'] = 'Fedora69'
options = [
('assumeyes', None),
('best', False),
('cachedir', dnf.const.TMPDIR),
('clean_requirements_on_remove', False),
('color', 'never'),
('color_update_installed', 'normal'),
('color_update_remote', 'normal'),
('color_list_available_downgrade', 'dim'),
('color_list_available_install', 'normal'),
('color_list_available_reinstall', 'bold'),
('color_list_available_upgrade', 'bold'),
('color_list_installed_extra', 'bold'),
('color_list_installed_newer', 'bold'),
('color_list_installed_older', 'bold'),
('color_list_installed_reinstall', 'normal'),
('color_update_local', 'bold'),
('debug_solver', False),
('debuglevel', 2),
('defaultyes', False),
('disable_excludes', []),
('diskspacecheck', True),
('exclude', []),
('includepkgs', []),
('install_weak_deps', True),
('history_record', False),
('installonly_limit', 0),
('installonlypkgs', ['kernel']),
('installroot', '/tmp/dnf-test-installroot/'),
('ip_resolve', None),
('multilib_policy', 'best'),
('obsoletes', True),
('persistdir', dnf.const.PERSISTDIR),
('transformdb', False),
('protected_packages', ["dnf"]),
('plugins', False),
('showdupesfromrepos', False),
('tsflags', []),
('strict', True),
] + list(kwargs.items())
for optname, val in options:
self._set_value(optname, val, dnf.conf.PRIO_DEFAULT)
# TODO: consolidate with dnf.cli.Cli._read_conf_file()
for opt in ('cachedir', 'logdir', 'persistdir'):
# don't prepend installroot if option was specified by user
# TODO: is this desired? ^^^ (tests won't pass without it ATM)
if opt in kwargs:
continue
self.prepend_installroot(opt)
try:
os.makedirs(self.persistdir)
except:
pass
@property
def releasever(self):
return self.substitutions['releasever']
class FakePersistor(object):
reset_last_makecache = False
expired_to_add = set()
def get_expired_repos(self):
return set()
def since_last_makecache(self):
return None
def save(self):
pass
# object matchers for asserts
class ObjectMatcher(object):
"""Class allowing partial matching of objects."""
def __init__(self, type_=None, attrs=None):
"""Initialize a matcher instance."""
self._type = type_
self._attrs = attrs
def __eq__(self, other):
"""Test whether this object is equal to the *other* one."""
if self._type is not None:
if type(other) is not self._type:
return False
if self._attrs:
for attr, value in self._attrs.items():
if value != getattr(other, attr):
return False
return True
def __ne__(self, other):
"""Test whether this object is not equal to the *other* one."""
return not self == other
def __repr__(self):
"""Compute the "official" string representation of this object."""
args_strs = []
if self._type is not None:
args_strs.append('type_=%s' % repr(self._type))
if self._attrs:
attrs_str = ', '.join('%s: %s' % (dnf.i18n.ucd(attr), repr(value))
for attr, value in self._attrs.items())
args_strs.append('attrs={%s}' % attrs_str)
return '%s(%s)' % (type(self).__name__, ", ".join(args_strs))
# test cases:
class TestCase(unittest.TestCase):
if not dnf.pycomp.PY3:
assertCountEqual = unittest.TestCase.assertItemsEqual
assertRegex = unittest.TestCase.assertRegexpMatches
def assertEmpty(self, collection):
return self.assertEqual(len(collection), 0)
def assertFile(self, path):
"""Assert the given path is a file."""
return self.assertTrue(os.path.isfile(path))
def assertLength(self, collection, length):
return self.assertEqual(len(collection), length)
def assertPathDoesNotExist(self, path):
return self.assertFalse(os.access(path, os.F_OK))
def assertStartsWith(self, string, what):
return self.assertTrue(string.startswith(what))
def assertTracebackIn(self, end, string):
"""Test that a traceback ending with line *end* is in the *string*."""
traces = (match.group() for match in TRACEBACK_RE.finditer(string))
self.assertTrue(any(trace.endswith(end) for trace in traces))
def assertTransEqual(self, trans_pkgs, list):
return self.assertCountEqual([pkg.name for pkg in trans_pkgs], list)
class DnfBaseTestCase(TestCase):
# create base with specified test repos
REPOS = []
# initialize mock sack
INIT_SACK = False
# initialize self.base._transaction
INIT_TRANSACTION = False
# False: self.base = MockBase()
# True: self.base = BaseCliStub()
BASE_CLI = False
# None: self.cli = None
# "init": self.cli = dnf.cli.cli.Cli(self.base)
# "mock": self.cli = self.base.mock_cli()
# "stub": self.cli = StubCli(self.base)
CLI = None
COMPS = False
COMPS_SEED_HISTORY = False
COMPS_SOLVER = False
def setUp(self):
self._installroot = tempfile.mkdtemp(prefix="dnf_test_installroot_")
if self.BASE_CLI:
self.base = BaseCliStub(*self.REPOS, installroot=self._installroot)
else:
self.base = MockBase(*self.REPOS, installroot=self._installroot)
if self.CLI is None:
self.cli = None
elif self.CLI == "init":
self.cli = dnf.cli.cli.Cli(self.base)
elif self.CLI == "mock":
self.cli = self.base.mock_cli()
elif self.CLI == "stub":
self.cli = CliStub(self.base)
else:
raise ValueError("Invalid CLI value: {}".format(self.CLI))
if self.COMPS:
self.base.read_mock_comps(seed_history=self.COMPS_SEED_HISTORY)
if self.INIT_SACK:
self.base.init_sack()
if self.INIT_TRANSACTION:
self.base._transaction = self.base.history.rpm
if self.COMPS_SOLVER:
self.solver = dnf.comps.Solver(self.history, self.comps, REASONS.get)
else:
self.solver = None
def tearDown(self):
self.base.close()
if self._installroot.startswith("/tmp/"):
shutil.rmtree(self._installroot)
@property
def comps(self):
return self.base.comps
@property
def goal(self):
return self.base._goal
@property
def history(self):
return self.base.history
@property
def sack(self):
return self.base.sack
def _swdb_begin(self, tsis=None):
# history.beg() replaces persistor.commit()
tsis = tsis or []
self.history.beg("", [], tsis)
def _swdb_end(self, tsis=None):
for tsi in self.history._swdb.getItems():
if tsi.getState() == libdnf.transaction.TransactionItemState_UNKNOWN:
tsi.setState(libdnf.transaction.TransactionItemState_DONE)
self.history.end("")
self.history.close()
def _swdb_commit(self, tsis=None):
self._swdb_begin(tsis)
self._swdb_end()
self.history.close()
class ResultTestCase(DnfBaseTestCase):
allow_erasing = False
def _get_installed(self, base):
try:
base.resolve(self.allow_erasing)
except dnf.exceptions.DepsolveError:
self.fail()
installed = set(base.sack.query().installed())
for r in base._transaction.remove_set:
installed.remove(r)
installed.update(base._transaction.install_set)
return installed
def assertResult(self, base, pkgs):
"""Check whether the system contains the given pkgs.
pkgs must be present. Any other pkgs result in an error. Pkgs are
present if they are in the rpmdb and are not REMOVEd or they are
INSTALLed.
"""
self.assertCountEqual(self._get_installed(base), pkgs)
def installed_removed(self, base):
try:
base.resolve(self.allow_erasing)
except dnf.exceptions.DepsolveError:
self.fail()
installed = base._transaction.install_set
removed = base._transaction.remove_set
return installed, removed