# -*- 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)?)+'
    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_releasevers', return_value=(69, None, None)):
            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)
        self._goal.protect_running_kernel = self.conf.protect_running_kernel
        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"]),
            ('protect_running_kernel', True),
            ('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

    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