# testdesc is part of autopkgtest
# autopkgtest is a tool for testing Debian binary packages
#
# autopkgtest is Copyright (C) 2006-2014 Canonical Ltd.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty 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., 675 Mass Ave, Cambridge, MA 02139, USA.
#
# See the file CREDITS for a full list of credits information (often
# installed as /usr/share/doc/autopkgtest/CREDITS).

import string
import re
import errno
import os.path
import subprocess
import tempfile
from typing import Dict, FrozenSet, Iterable, List, Optional, Tuple, Union

import debian.deb822
import debian.debian_support
import debian.debfile

import adtlog

#
# Abstract test representation
#

known_restrictions = frozenset(
    {
        "allow-stderr",
        "breaks-testbed",
        "build-needed",
        "flaky",
        "isolation-container",
        "isolation-machine",
        "needs-internet",
        "needs-reboot",
        "needs-recommends",
        "needs-root",
        "needs-sudo",
        "rw-build-tree",
        "skip-foreign-architecture",
        "skip-not-installable",
        "skippable",
        "superficial",
    }
)


# Keys are restrictions
# Values are sets of either:
# - str: a capability that is required by the restriction
# - set of two or more str: at least one of these caps is required
#   by the restriction
RESTRICTIONS_REQUIRE_CAPS: Dict[str, FrozenSet[Union[str, FrozenSet[str]]]] = {
    "breaks-testbed": frozenset({"revert-full-system"}),
    "needs-internet": frozenset({"has_internet"}),
    "needs-reboot": frozenset({"reboot"}),
    "needs-root": frozenset({"root-on-testbed"}),
    "isolation-container": frozenset(
        {
            frozenset({"isolation-container", "isolation-machine"}),
        }
    ),
    "isolation-machine": frozenset({"isolation-machine"}),
}


# Keys are restrictions
# Values are sets of dependency strings
RESTRICTIONS_IMPLY_DEPS = {
    "needs-sudo": frozenset({"passwd", "sudo"}),
}


PKGDATADIR = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))


class Unsupported(Exception):
    """Test cannot be run in the testbed"""

    def __init__(self, testname, message):
        self.testname = testname
        self.message = message

    def __str__(self):
        return "Unsupported test %s: %s" % (self.testname, self.message)

    def report(self):
        adtlog.report(self.testname, "SKIP %s" % self.message)


class InvalidControl(Exception):
    """Test has invalid control data"""

    def __init__(self, testname, message):
        self.testname = testname
        self.message = message

    def __str__(self):
        return "InvalidControl test %s: %s" % (self.testname, self.message)

    def report(self):
        adtlog.report(self.testname, "BROKEN %s" % self.message)


class Test:
    """Test description.

    This is only a representation of the metadata, it does not have any
    actions.
    """

    def __init__(
        self,
        name: str,
        path: Optional[str],
        command: Optional[str],
        restrictions: Iterable[str],
        features: Iterable[str],
        depends: List[List[str]],
        package_under_test_depends: List[List[str]],
    ) -> None:
        """Create new test description

        A test must have either "path" or "command", the respective other value
        must be None.

        @name: Test name
        @path: path to the test's executable, relative to source tree
        @command: shell command for the test code
        @restrictions, @features: sets of strings, as in README.package-tests
        @depends: string list of test dependencies (packages)
        @package_under_test_depends: string list with the subset of @depends
        that come from the source package we are testing
        """
        if "/" in name:
            raise Unsupported(name, "test name may not contain / character")

        if not ((path is None) ^ (command is None)):
            raise InvalidControl(name, "Test must have either path or command")

        self.name = name
        self.path = path
        self.command = command
        self.restrictions = set(restrictions)
        self.features = set(features)
        self.depends = list(depends)
        self.package_under_test_depends = package_under_test_depends

        for r in self.restrictions:
            for d in RESTRICTIONS_IMPLY_DEPS.get(r, frozenset()):
                if d not in self.depends:
                    self.depends.append([d])

        # None while test hasn't run yet; True: pass, False: fail
        self.result = None
        self.skipped = False
        adtlog.debug(
            'Test defined: name %s path %s command "%s" '
            "restrictions %s features %s depends %s "
            % (name, path, command, sorted(restrictions), sorted(features), depends)
        )

    def passed(self):
        """Mark test as passed"""

        self.result = True
        if "superficial" in self.restrictions:
            adtlog.report(self.name, "PASS (superficial)")
        else:
            adtlog.report(self.name, "PASS")

    def set_skipped(self, reason):
        """Mark test as skipped"""
        # This isn't called skipped() to avoid clashing with the boolean
        # attribute.

        self.skipped = True
        self.result = True
        adtlog.report(self.name, "SKIP " + reason)

    def failed(self, reason):
        """Mark test as failed"""

        self.result = False
        if "flaky" in self.restrictions:
            adtlog.report(self.name, "FLAKY " + reason)
        else:
            adtlog.report(self.name, "FAIL " + reason)

    def check_testbed_compat(
        self,
        caps: List[str],
        ignore_restrictions: Optional[List[str]] = None,
    ) -> None:
        """Check for restrictions incompatible with test bed capabilities.

        Raise Unsupported exception if there are any.
        """
        effective = set(self.restrictions)
        if ignore_restrictions:
            effective -= set(ignore_restrictions)
        provided = set(caps)

        for r in effective:
            if r not in known_restrictions:
                raise Unsupported(self.name, "unknown restriction %s" % r)

        for r in effective:
            need_caps = RESTRICTIONS_REQUIRE_CAPS.get(r, frozenset())

            for c in need_caps:
                if isinstance(c, str):
                    if c not in caps:
                        raise Unsupported(
                            self.name,
                            ('Test restriction "%s" requires testbed capability "%s"')
                            % (r, c),
                        )
                else:
                    # must have at least one of the capabilities
                    assert isinstance(c, frozenset)
                    assert len(c) > 1

                    if not (c & provided):
                        cap_names = sorted('"%s"' % n for n in c)
                        needed = ", ".join(cap_names[:-1])
                        needed += " and/or " + cap_names[-1]

                        raise Unsupported(
                            self.name,
                            ('Test restriction "%s" requires testbed capability %s')
                            % (r, needed),
                        )


#
# Parsing for Debian source packages
#


def parse_rfc822(path):
    """Parse Debian-style RFC822 file

    Yield dictionaries with the keys/values.
    """
    try:
        f = open(path, encoding="UTF-8")
    except (IOError, OSError) as oe:
        if oe.errno != errno.ENOENT:
            raise
        return

    # filter out comments, python-debian doesn't do that
    # (http://bugs.debian.org/743174)
    lines = []
    for line in f:
        # completely ignore ^# as that breaks continuation lines
        if line.startswith("#"):
            continue
        # filter out comments which don't start on first column (Debian
        # #743174); entirely remove line if all that's left is whitespace, as
        # that again breaks continuation lines
        if "#" in line:
            line = line.split("#", 1)[0]
            if not line.strip():
                continue
        lines.append(line)
    f.close()

    for p in debian.deb822.Deb822.iter_paragraphs(lines):
        r = {}
        for field, value in p.items():
            # un-escape continuation lines
            v = "".join(value.split("\n")).replace("  ", " ")
            field = string.capwords(field)
            r[field] = v
        yield r


def _split_and_strip(string: str, delimiter: str = ",") -> List[str]:
    """Split the provided string on delimiter, strip the resulting
    elements and return a list of the non-empty ones."""

    return [x for x in (x.strip() for x in string.split(delimiter)) if x]


def _debian_check_unknown_fields(name, record):
    unknown_keys = set(record.keys()).difference(
        {
            "Tests",
            "Test-command",
            "Restrictions",
            "Features",
            "Depends",
            "Tests-directory",
            "Classes",
            "Architecture",
        }
    )
    if unknown_keys:
        raise Unsupported(name, "unknown field %s" % unknown_keys.pop())


def _architecture_is_concerned(
    architecture: str,
    architecture_restrictions: List[str],
) -> bool:
    """Wrapper around python-debian's architecture_is_concerned() with perl
    fallback in case python-debian is too old."""

    # TODO: Add Depends: python3-debian (>= 0.1.48) to bin:autopkgtest and drop this.

    if hasattr(debian.debian_support, "DpkgArchTable") and hasattr(
        debian.debian_support.DpkgArchTable, "architecture_is_concerned"
    ):
        dpkg_arch_table = debian.debian_support.DpkgArchTable.load_arch_table()
        return dpkg_arch_table.architecture_is_concerned(
            architecture,
            architecture_restrictions,
        )
    else:
        res = subprocess.run(
            [
                os.path.join(PKGDATADIR, "lib", "arch-is-concerned.pl"),
                architecture,
                " ".join(architecture_restrictions),
            ]
        )
        return res.returncode == 0


def _debian_packages_from_source(
    srcdir: str,
    test_arch: str,
    test_arch_is_foreign: bool,
) -> Tuple[List[str], List[str]]:
    """Parse source tree in srcdir and return tuple with: list of all binary
    deb packages built from that source, and the same list but restricted to
    packages that build for test_arch."""

    packages: List[str] = []
    packages_for_test_arch: List[str] = []

    for st in parse_rfc822(os.path.join(srcdir, "debian/control")):
        if "Package" not in st:
            # source stanza
            continue
        # filter out udebs and similar stuff which aren't "real" debs
        if (
            st.get("Xc-package-type", "deb") != "deb"
            or st.get("Package-type", "deb") != "deb"
        ):
            continue
        binarydeb = st["Package"]
        arch = st["Architecture"]
        packages.append(binarydeb)
        if arch == "all":
            packages_for_test_arch.append(binarydeb)
        else:
            arch_list = arch.split()
            if _architecture_is_concerned(test_arch, arch_list):
                if test_arch_is_foreign:
                    binarydeb += ":" + test_arch
                packages_for_test_arch.append(binarydeb)

    return (packages, packages_for_test_arch)


def _filter_variables(deps: Iterable[str]) -> List[str]:
    deplist = []
    for dep in deps:
        if re.match(r"^\${", dep):
            continue
        dep = re.sub(r"\(.*\${\S*}.*\)", "", dep)
        deplist.append(dep.strip())
    return deplist


def _debian_recommends_from_source(srcdir: str) -> List[str]:
    deps = []
    for st in parse_rfc822(os.path.join(srcdir, "debian/control")):
        if "Recommends" in st:
            deps += _split_and_strip(st["Recommends"])

    # Because we parse the source d/control file instead of the binary
    # control file, we may run into variables. See bug 1008206 for
    # some discussion.
    deps = _filter_variables(deps)

    # The following check is a bit silly as there were apparently no
    # recommends, but still the test asked for it. Let's not fail on
    # that, but also let's not add the empty string.
    if deps == [""]:
        deps = []

    return deps


def _debian_build_deps_from_source(srcdir: str) -> List[List[str]]:
    deps = []
    for st in parse_rfc822(os.path.join(srcdir, "debian/control")):
        for key in ("Build-depends", "Build-depends-indep", "Build-depends-arch"):
            if key in st:
                altgr = _split_and_strip(st[key])
                deps += [_split_and_strip(g, "|") for g in altgr]

    # @builddeps@ should always imply build-essential
    deps.append(["build-essential"])
    return deps


dep_re = re.compile(
    r"(?P<package>[a-z0-9+-.]+)(?::[a-z0-9_-]+)?\s*"
    r"(\((?P<relation><<|<=|>=|=|>>)\s*(?P<version>[^\)]*)\))?"
    r"(\s*\[(?P<arch>[a-z0-9+-.! ]+)\])?$"
)


def _debian_check_dep(testname, dep):
    """Check a single Debian dependency"""

    dep = dep.strip()
    m = dep_re.match(dep)
    if not m:
        raise InvalidControl(
            testname, "Test Depends field contains an invalid dependency `%s'" % dep
        )
    if m.group("version"):
        try:
            debian.debian_support.NativeVersion(m.group("version"))
        except ValueError:
            raise InvalidControl(
                testname,
                "Test Depends field contains "
                "dependency `%s' with an "
                "invalid version" % dep,
            )
        except AttributeError:
            # too old python-debian, skip the check
            pass

    return (m.group("package"), m.group("version"))


def _expand_test_depends(
    testname: str,
    dep_str: str,
    srcdir: str,
    test_arch: str,
    test_arch_is_foreign: bool,
) -> Tuple[List[List[str]], List[List[str]]]:
    """Split test dependencies (comma separated), validate their syntax, and
    expand @, @builddeps@ and @recommends@. Returns list of dependencies
    and the subset of those dependencies that come from the source
    package under test.

    If test_arch_is_foreign is True, then :test_arch architecture qualifiers
    are added to packages that are not Architecture: all.

    This may raise an InvalidControl exception if there are invalid
    dependencies.
    """
    deps = []
    package_under_test_deps = []
    (my_packages, my_packages_for_test_arch) = _debian_packages_from_source(
        srcdir,
        test_arch,
        test_arch_is_foreign,
    )

    for alt_group_str in _split_and_strip(dep_str):
        adtlog.debug("processing dependency %s" % alt_group_str)
        if alt_group_str == "@":
            for d in my_packages_for_test_arch:
                adtlog.debug("synthesised dependency %s from @" % d)
                deps.append([d])
                package_under_test_deps.append([d])
        elif alt_group_str == "@builddeps@":
            for daltgr in _debian_build_deps_from_source(srcdir):
                adtlog.debug(
                    "synthesised dependency %s from @builddeps@" % " | ".join(daltgr)
                )
                deps.append(daltgr)
        elif alt_group_str == "@recommends@":
            for d in _debian_recommends_from_source(srcdir):
                adtlog.debug("synthesised dependency %s from @recommends@" % d)
                deps.append([d])
        else:
            alt_group_list = []
            package_under_test_alternatives = []
            for dep in _split_and_strip(alt_group_str, "|"):
                (pkg, version) = _debian_check_dep(testname, dep)
                if (
                    test_arch_is_foreign
                    and f"{pkg}:{test_arch}" in my_packages_for_test_arch
                ):
                    dep = dep.replace(pkg, f"{pkg}:{test_arch}", 1)
                    pkg += f":{test_arch}"
                alt_group_list.append(dep)
                if pkg in my_packages:
                    package_under_test_alternatives.append(pkg)
            if len(alt_group_list) == len(package_under_test_alternatives):
                # We depend on x|y|z, and all of x, y and z are binary
                # packages built by the package under test
                adtlog.debug(
                    "marked alternatives %s as part of the package under test"
                    % package_under_test_alternatives
                )
                package_under_test_deps.append(package_under_test_alternatives)
            deps.append(alt_group_list)

    if test_arch_is_foreign and ["build-essential"] in deps:
        # Ensure we are installing the testbed-native version of
        # build-essential by specifying :native. This is needed because the
        # build-essential package is not Multi-Arch: foreign, so we need to
        # make sure we are always picking up the package for the testbed
        # architecture, not for the foreign test architecture.
        deps.remove(["build-essential"])
        deps.append(["build-essential:native"])
        # Install the right crossbuild-essential package.
        deps.append([f"crossbuild-essential-{test_arch}:native"])
        # Add install libc-dev and libstdc++-dev, as a workaround for #815172.
        deps.append([f"libc-dev:{test_arch}"])
        deps.append([f"libstdc++-dev:{test_arch}"])

    return (deps, package_under_test_deps)


def _autodep8(srcdir):
    """Generate control file with autodep8"""

    f = tempfile.NamedTemporaryFile(prefix="autodep8.")
    try:
        autodep8 = subprocess.Popen(
            ["autodep8"], cwd=srcdir, stdout=f, stderr=subprocess.PIPE
        )
    except OSError as e:
        adtlog.debug("autodep8 not available (%s)" % e)
        return None

    err = autodep8.communicate()[1].decode()
    if autodep8.returncode == 0:
        f.flush()
        f.seek(0)
        ctrl = f.read().decode()
        adtlog.debug("autodep8 generated control: -----\n%s\n-------" % ctrl)
        return f

    f.close()
    adtlog.debug(
        "autodep8 failed to generate control (exit status %i): %s"
        % (autodep8.returncode, err)
    )
    return None


def _matches_architecture(host_arch, arch_wildcard):
    try:
        subprocess.check_call(
            [
                "perl",
                "-mDpkg::Arch",
                "-e",
                "exit(!Dpkg::Arch::debarch_is(shift, shift))",
                host_arch,
                arch_wildcard,
            ]
        )
    except subprocess.CalledProcessError as e:
        # returns 1 if host_arch is not matching arch_wildcard; other
        # errors shouldn't be ignored
        if e.returncode != 1:
            raise
        return False
    return True


def _check_architecture(name, test_arch, architectures):
    """Check if test_arch is supported by the test architectures

    The architecture list comes in two variants, positive: only this
    arch is supported (arch may be a wildcard) and negative: this arch
    is not supported (arch may be a wildcard). If there is any
    positive arch, every arch not explicitly listed is skipped. Debian
    Policy 7.1 explains that for (Build-)Depends it's not allowed to
    mix positive and negative, so let's not do either. The list can
    also be empty. Empty and ["any"] are the same, "all" isn't
    allowed.
    """

    if "all" in architectures:
        raise Unsupported(name, "Arch 'all' not allowed in Architecture field")

    if len(architectures) == 0 or architectures == ["any"]:
        return

    any_negative = False
    any_positive = False
    for arch in architectures:
        if arch[0] == "!":
            any_negative = True
            if _matches_architecture(test_arch, arch[1:]):
                raise Unsupported(
                    name,
                    "Test declares architecture as not " + "supported: %s" % test_arch,
                )
        if arch[0] != "!":
            any_positive = True

    if any_positive:
        if any_negative:
            raise Unsupported(
                name,
                "It is not permitted for some archs to "
                + "be prepended by an exclamation mark while "
                + "others aren't",
            )
        arch_matched = False
        for arch in architectures:
            if _matches_architecture(test_arch, arch):
                arch_matched = True

        if not arch_matched:
            raise Unsupported(
                name,
                "Test lists explicitly supported "
                + "architectures, but the current architecture "
                + "%s isn't listed." % test_arch,
            )


def parse_debian_source(
    srcdir: str,
    testbed_caps: List[str],
    test_arch: str,
    test_arch_is_foreign: bool,
    control_path: Optional[str] = None,
    auto_control: bool = True,
    ignore_restrictions: Optional[List[str]] = None,
    only_tests: Optional[List[str]] = None,
) -> Tuple[List[Test], bool]:
    """Parse test descriptions from a Debian DEP-8 source dir

    @ignore_restrictions: If we would skip the test due to these restrictions,
                          run it anyway

    You can specify an alternative path for the control file (default:
    srcdir/debian/tests/control).

    Return (list of Test objects, some_skipped). If this encounters any invalid
    restrictions, fields, or test restrictions which cannot be met by the given
    testbed capabilities, the test will be skipped (and reported so), and not
    be included in the result.

    This may raise an InvalidControl exception.
    """
    some_skipped = False
    command_counter = 0
    tests = []
    if not ignore_restrictions:
        ignore_restrictions = []
    if not control_path:
        control_path = os.path.join(srcdir, "debian", "tests", "control")
        dtc_exists = os.path.exists(control_path)
        try_autodep8 = False

        if auto_control:
            if not dtc_exists:
                try_autodep8 = True
            else:
                dcontrol_path = os.path.join(srcdir, "debian", "control")
                for record in parse_rfc822(dcontrol_path):
                    testsuite = record.get("Testsuite", "")
                    if "autopkgtest-pkg-" in testsuite:
                        try_autodep8 = True
                    # We only want to look at the source section
                    break

        if try_autodep8:
            control = _autodep8(srcdir)
            if control is not None:
                control_path = control.name
            elif not dtc_exists:
                return ([], False)
        elif not dtc_exists:
            adtlog.debug("auto_control is disabled, and no regular tests")
            return ([], False)

    for record in parse_rfc822(control_path):
        command = None
        try:
            restrictions = set(record.get("Restrictions", "").replace(",", " ").split())

            # needs-recommends is deprecated, but until removed, let's
            # replace it with @recommends@
            if "needs-recommends" in restrictions:
                record["Depends"] = record.get("Depends", "@") + ",@recommends@"

            feature_test_name = None
            features = set()
            record_features = record.get("Features", "").replace(",", " ").split()
            for feature in record_features:
                details = feature.split("=", 1)
                if details[0] != "test-name":
                    features.add(feature)
                    continue
                if len(details) != 2:
                    # No value, i.e. a bare 'test-name'
                    raise InvalidControl("*", "test-name feature with no argument")
                if feature_test_name is not None:
                    raise InvalidControl("*", "only one test-name feature allowed")
                feature_test_name = details[1]
                features.add(feature)
            architectures = record.get("Architecture", "").replace(",", " ").split()

            if "Tests" in record:
                test_names = record["Tests"].replace(",", " ").split()
                if len(test_names) == 0:
                    raise InvalidControl("*", '"Tests" field is empty')
                (depends, package_under_test_depends) = _expand_test_depends(
                    test_names[0],
                    record.get("Depends", "@"),
                    srcdir,
                    test_arch,
                    test_arch_is_foreign,
                )
                if "Test-command" in record:
                    raise InvalidControl(
                        "*", 'Only one of "Tests" or "Test-Command" may be given'
                    )
                if feature_test_name is not None:
                    raise InvalidControl(
                        "*", "test-name feature incompatible with Tests"
                    )
                test_dir = record.get("Tests-directory", "debian/tests")

                for n in test_names:
                    try:
                        _debian_check_unknown_fields(n, record)
                        _check_architecture(n, test_arch, architectures)

                        test = Test(
                            n,
                            os.path.join(test_dir, n),
                            None,
                            restrictions,
                            features,
                            depends,
                            package_under_test_depends,
                        )
                        test.check_testbed_compat(testbed_caps, ignore_restrictions)
                    except Unsupported as u:
                        if (not only_tests) or n in only_tests:
                            u.report()
                            some_skipped = True
                    else:
                        tests.append(test)
            elif "Test-command" in record:
                command = record["Test-command"]
                (depends, package_under_test_depends) = _expand_test_depends(
                    command,
                    record.get("Depends", "@"),
                    srcdir,
                    test_arch,
                    test_arch_is_foreign,
                )
                if feature_test_name is None:
                    command_counter += 1
                    name = "command%i" % command_counter
                else:
                    name = feature_test_name
                _debian_check_unknown_fields(name, record)
                _check_architecture(name, test_arch, architectures)
                test = Test(
                    name,
                    None,
                    command,
                    restrictions,
                    features,
                    depends,
                    package_under_test_depends,
                )
                test.check_testbed_compat(testbed_caps, ignore_restrictions)
                tests.append(test)
            else:
                raise InvalidControl("*", 'missing "Tests" or "Test-Command" field')
        except Unsupported as u:
            if not only_tests:
                u.report()
                some_skipped = True

    return (tests, some_skipped)