#!/usr/bin/python3 -u
#
# autopkgtest is a tool for testing Debian binary packages
#
# autopkgtest is Copyright (C) 2006-2016 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 atexit
import json
import os
import re
import shlex
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import traceback
from collections import Counter

from debian.deb822 import Deb822
from debian.debian_support import Version

# support running out of git and from packaged install
our_base = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if os.path.isdir(os.path.join(our_base, "virt")):
    sys.path.insert(0, os.path.join(our_base, "lib"))
    vserver_dir = os.path.join(our_base, "virt")
    os.environ["PATH"] = vserver_dir + ":" + os.environ.get("PATH", "")
else:
    sys.path.insert(0, "/usr/share/autopkgtest/lib")

import adt_binaries
import adtlog
import testdesc
from adt_testbed import (
    Testbed,
    TestbedPath,
)
from autopkgtest_args import parse_args
from autopkgtest_timeouts import operation_timeout

# ---------- global variables

testbed: Testbed | None = None
blamed: list[str] = []


# ---------- convenience functions


def files_from_dsc(dsc_path: str) -> list[str]:
    """Get files from a .dsc or a .changes

    Return list of files, including the directory of dsc_path.
    """
    try:
        files = testdesc.parse_rfc822(dsc_path).__next__()["Files"].split()
    except (StopIteration, KeyError):
        adtlog.badpkg(f"{dsc_path} is invalid and does not contain Files:")

    dsc_dir = os.path.dirname(dsc_path)

    return [os.path.join(dsc_dir, f) for f in files if "." in f and "_" in f]


def blame(m):
    adtlog.debug(f"blame += {m}")
    blamed.append(m)


def setup_trace(*, logfile, summary):
    if logfile is not None:
        # tee stdout/err into log file
        (fd, fifo_log) = tempfile.mkstemp(prefix="autopkgtest-fifo.")
        os.close(fd)
        os.unlink(fifo_log)
        os.mkfifo(fifo_log)
        atexit.register(os.unlink, fifo_log)
        start = str(int(time.time()))

        out_tee = subprocess.Popen(["tee", fifo_log], stdin=subprocess.PIPE)
        with open("/dev/null", "wb") as devnull:
            err_tee = subprocess.Popen(
                ["tee", fifo_log, "-a", "/dev/stderr"],
                stdin=subprocess.PIPE,
                stdout=devnull,
            )

        # In order to correctly timestamp individual lines we need to
        # disable stdin buffering via -Winteractive.
        with open(logfile, "wb") as logfile_stream:
            log_awk = subprocess.Popen(
                [
                    "mawk",
                    "-Winteractive",
                    r'{ printf("%3ds %s\n", systime()-' + start + ", $0 );}",
                    fifo_log,
                ],
                stdout=logfile_stream,
            )

        adtlog.enable_colors = False
        os.dup2(out_tee.stdin.fileno(), sys.stdout.fileno())
        os.dup2(err_tee.stdin.fileno(), sys.stderr.fileno())

        def cleanup():
            os.close(sys.stdout.fileno())
            os.close(out_tee.stdin.fileno())
            out_tee.wait()
            os.close(sys.stderr.fileno())
            os.close(err_tee.stdin.fileno())
            err_tee.wait()
            log_awk.wait()

        atexit.register(cleanup)

    adtlog.summary_stream = open(summary, "w+b", 0)  # noqa: SIM115


def run_tests(
    tests, *, tree, binary_registry, env, shell, build_parallel, fail_fast, deadline
):
    """Run the selected tests and return their combined exit status."""
    errorcode = 0

    # The parser has already excluded architecture mismatches and unsupported
    # restrictions. Apply the remaining static skips before proceeding to
    # actually attempting test runs. In particular we want this to happen
    # before grouping of tests with the same test dependencies, as we don't
    # want skipped tests to count.
    runnable_tests = []
    for t in tests:
        if (
            testbed.test_arch_is_foreign
            and "skip-foreign-architecture" in t.restrictions
        ):
            errorcode |= 2
            adtlog.report(
                t.name, "SKIP test arch is foreign and skip-foreign-architecture set"
            )
        else:
            runnable_tests.append(t)
    tests = runnable_tests

    snapshots_enabled = "snapshot" in testbed.caps
    deps_to_prefetch = ()
    if snapshots_enabled:
        # Group tests with the same test dependencies to make the most use of the
        # "dependencies" snapshot. Note that this keeps the first test first.
        tests_by_deps = {}
        for t in tests:
            tests_by_deps.setdefault(t.depends, []).append(t)
        test_groups = list(tests_by_deps.values())

        # Compute dependencies to prefetch.
        if len(tests_by_deps) > 1:
            dependency_freq = Counter(
                dep for deps in tests_by_deps for dep in set(deps)
            )
            representative = max(
                tests_by_deps,
                key=lambda deps: sum(dependency_freq[dep] - 1 for dep in set(deps)),
            )
            deps_to_prefetch = tuple(
                dep for dep in representative if dependency_freq[dep] > 1
            )
            if deps_to_prefetch:
                adtlog.debug(
                    "Selected shared dependencies from test "
                    f"{tests_by_deps[representative][0].name}: {deps_to_prefetch}"
                )
    else:
        # Without snapshots, each test needs its own fresh testbed.
        test_groups = [[t] for t in tests]

    # Prepare the shared baseline before starting any individual test.
    # This is saved in the "ready" snapshot.
    if snapshots_enabled and tests:
        binary_registry.publish()
        testbed.prefetch_deps(deps_to_prefetch)
        if len(test_groups) > 1:
            testbed.snapshot_create("ready")

    any_positive = False

    def skip_if_run_aborted(remaining_tests):
        nonlocal errorcode
        if fail_fast and (errorcode & 4):
            reason = "previous test failed and --fail-fast is set"
        elif deadline is not None and time.monotonic() >= deadline:
            reason = "global timeout exceeded"
        else:
            return False
        errorcode |= 2
        for t in remaining_tests:
            adtlog.report(t.name, f"SKIP {reason}")
        return True

    for group_index, group in enumerate(test_groups):
        if skip_if_run_aborted(group):
            continue

        # Set up a clean testbed and install dependencies once for the group.
        first_test = group[0]
        adtlog.info(f"test {first_test.name}: preparing testbed")
        if snapshots_enabled:
            if group_index > 0:
                testbed.snapshot_restore("ready")
        else:
            testbed.revert()
            binary_registry.publish()

        try:
            testbed.install_test_dependencies(
                first_test.depends,
                real_package_deps=first_test.package_under_test_depends,
            )
        except adtlog.BadPackageError as e:
            for t in group:
                if "skip-not-installable" in t.restrictions:
                    errorcode |= 2
                    adtlog.report(
                        t.name, "SKIP installation fails and skip-not-installable set"
                    )
                else:
                    errorcode |= 12
                    adtlog.report(t.name, "FAIL badpkg")
                    adtlog.preport("blame: " + " ".join(blamed))
                    adtlog.preport("badpkg: " + str(e))
            continue

        # Only multi-test groups need a snapshot of their installed dependencies.
        if len(group) > 1:
            testbed.snapshot_create("dependencies")

        for t_index, t in enumerate(group):
            if t_index > 0:
                if skip_if_run_aborted([t]):
                    continue
                adtlog.info(f"test {t.name}: preparing testbed")
                testbed.snapshot_restore("dependencies")

            try:
                testbed.satisfy_restrictions(t.name, t.restrictions)
            except testdesc.Unsupported as unsupported:
                errorcode |= 2
                unsupported.report()
                continue

            testbed.run_test(
                tree=tree,
                test=t,
                extra_env=env,
                shell=shell,
                build_parallel=build_parallel,
                deadline=deadline,
            )
            if t.skipped or (not t.result and "flaky" in t.restrictions):
                errorcode |= 2
            elif not t.result:
                errorcode |= 4
            elif "superficial" not in t.restrictions:
                # A superficial test passing is merely a neutral result,
                # not a positive result
                any_positive = True

        if len(group) > 1:
            testbed.snapshot_delete("dependencies")

    if errorcode in (0, 2) and not any_positive:
        # If we have skipped or ignored every non-superficial test, set
        # the same exit status as if we didn't have any tests
        errorcode = 8

    return errorcode


def create_testinfo(vserver_args, *, env, work_dir):
    info = {"virt_server": shlex.join(vserver_args)}

    if testbed.initial_kernel_version:
        info["kernel_version"] = testbed.initial_kernel_version
    if testbed.test_kernel_versions:
        info["test_kernel_versions"] = testbed.test_kernel_versions
    if env:
        info["custom_environment"] = env
    if testbed.nproc:
        info["nproc"] = testbed.nproc
    if testbed.cpu_model:
        info["cpu_model"] = testbed.cpu_model
    if testbed.cpu_flags:
        info["cpu_flags"] = testbed.cpu_flags

    with open(os.path.join(work_dir, "testinfo.json"), "w") as f:
        json.dump(info, f, indent=2)


def print_exception(ei, msgprefix=""):
    (et, e, _tb) = ei
    try:
        if msgprefix:
            adtlog.error(msgprefix)
        if et is adtlog.BadPackageError:
            adtlog.preport("blame: " + " ".join(blamed))
            adtlog.preport("badpkg: " + e.args[0])
            adtlog.error("erroneous package: " + e.args[0])
            adtlog.psummary("erroneous package: " + e.args[0])
            return 12
        elif et is adtlog.TestbedFailure:
            adtlog.error("testbed failure: " + e.args[0])
            adtlog.psummary("testbed failure: " + e.args[0])
            return 16
        elif et is adtlog.AutopkgtestError:
            adtlog.psummary(e.args[0])
            adtlog.error(e.args[0])
            return 20
        else:
            adtlog.error("unexpected error:")
            adtlog.psummary("quitting: unexpected error, see log")
            traceback.print_exc(None, sys.stderr)
            return 20
    except OSError as exc:
        print(f"Issue while logging exception: {exc}", file=sys.stderr)
        print(f"Exception was '{e}'", file=sys.stderr)
        return 16


def cleanup(*, binary_registry, keep_output, work_dir):
    """Clean up the testbed and local temporary files."""
    try:
        if testbed is not None:
            if binary_registry is not None:
                binary_registry.revert_apt_config()
            testbed.stop()
    finally:
        # Local temp-dir removal must run even if testbed teardown failed.
        if not keep_output and work_dir is not None:
            shutil.rmtree(work_dir, ignore_errors=True)


def signal_handler(signum, frame):
    adtlog.error(f"Received signal {signum}, cleaning up...")
    signal.signal(signum, signal.SIG_DFL)
    try:
        # don't call cleanup() here, reverting apt takes too long
        if testbed:
            testbed.stop()
    finally:
        os.kill(os.getpid(), signum)


# ---------- source preparation


def deb_package_name(deb):
    """Return package name from a .deb"""

    try:
        return subprocess.check_output(
            ["dpkg-deb", "--field", deb, "Package"], text=True
        ).strip()
    except subprocess.CalledProcessError as e:
        adtlog.badpkg(f"failed to parse binary package: {e}")


def find_source_version_to_download(src):
    """Honoring pinning, find the version of the source package to download.

    The version for "apt-get source pkg=version" is determined honoring the
    current apt pinning. This is done by putting all binary packages built from
    the source package in one "pool". This includes binary packages build from
    src:pkg from all the configured suites/pockets. In this pool we then look
    for the packages with the highest priority, and among these we select the
    highest version. This is the same logic followed by apt to determine which
    package to install, as documented in apt_preferences(5).
    """

    result = testbed.execute(
        ["apt-cache", "showsrc", "--only-source", src],
        check=False,
        capture_output=True,
    )

    if result.returncode != 0 or (result.stdout == "" and result.stderr != ""):
        adtlog.badpkg(f"apt-cache showsrc didn't succeed: {result.stderr}")
    # To make sure we have at least one paragraph that iter_paragraphs can
    # work on.
    assert ":" in result.stdout, "unexpected apt-cache output"

    # Somewhere in the next loop, we need the os of the testbed; Currently all
    # official Debian architectures are Linux, but we have hurd and we had
    # kfreebsd in the ports. So, support other os' but let's optimize for linux
    # and only query the testbed if we're not on linux.
    os = "linux"
    if not testbed.initial_kernel_version.startswith("Linux"):
        os = testbed.execute(
            ["dpkg-architecture", "-q", "DEB_HOST_ARCH_OS"],
            check=True,
            capture_output=True,
            allow_stderr=False,
        ).stdout.strip()

    # The output apt-get showsrc follows the ordering of sources.
    # Ordering by package version makes the following logic easier.
    candidate_packages = sorted(
        Deb822.iter_paragraphs(result.stdout),
        key=lambda k: Version(k.get("Version", "~")),
    )
    binpkgs_all = set()
    for paragr in candidate_packages:
        # This version should never be the outcome anyways, but let's
        # quite early.
        if "Extra-Source-Only" in paragr:
            continue

        # Very old source packages don't have Package-List: yet, fall
        # back to Binary: (Binary: is generally not sufficient as it
        # gets truncated for long lists, although that problem should
        # fade out as well).
        if "Package-List" not in paragr:
            binpkgs_all.update(set(paragr.get("Binary").split(", ")))
        else:
            package_list = paragr.get("Package-List").strip().split("\n")
            for bin in package_list:
                # package deb section optional arch=arch1,arch2
                # package udeb debian-installer optional arch=arch1 profile=!noudeb
                parts = bin.split()

                # only consider packages with Package-Type: deb, and skip for
                # for example udeb packages, which are not even installable with
                # the sources we set up.
                if parts[1] != "deb":
                    continue

                # Package-List entries created by dpkg-source from dpkg
                # (<< 1.17.7) lack the architecture (and build-profiles)
                # information. In this case let's assume the package is
                # installable on the arch are are interested in.
                if len(parts) < 5:
                    binpkgs_all.add(parts[0])
                    continue

                assert parts[4].startswith("arch=")
                archs = parts[4].split("=")[1].split(",")
                if any(
                    x in archs
                    for x in [
                        "all",
                        "any",
                        "any-" + testbed.test_arch,
                        testbed.test_arch,
                        os + "-any",
                    ]
                ):
                    binpkgs_all.add(parts[0])

    if not binpkgs_all:
        adtlog.badpkg(f"no binaries found for src:{src}")

    max_priority = -32768  # priorities are signed shorts
    max_src_version = Version("~")
    for bin in binpkgs_all:
        version, priority, _ = testbed.get_candidate(bin)
        if not version:
            adtlog.warning(f"No APT candidate found for {bin}, skipping")
            continue

        if priority < max_priority:
            continue

        # Now get the version of the source of this candidate as it might
        # be different than the binary version.
        result = testbed.execute(
            ["apt-cache", "show", bin + "=" + str(version)],
            check=False,
            stdout=subprocess.PIPE,
        )
        if result.returncode != 0:
            adtlog.warning(
                f"apt-cache show {bin}={version} failed with exit status "
                f"{result.returncode}, skipping candidate"
            )
            continue
        if result.stdout == "":
            adtlog.warning(
                f"apt-cache show {bin}={version} returned no data, skipping candidate"
            )
            continue
        pkg_info = Deb822(result.stdout)
        # If there's no Source item, the source and binary are named the
        # same and have the same version.
        pkg_info_src = pkg_info.get("Source", bin).split()
        # Check that the binary is still from this source. Inconsistencies may
        # happen when two source package build the same binary package, as
        # described in LP: #2066290.
        if pkg_info_src[0] != src:
            continue
        # Get the version (absent if same as the binary)
        if len(pkg_info_src) > 1:
            # source version comes between brackets
            version = Version(pkg_info_src[1][1:-1])

        if priority == max_priority:
            max_src_version = max(version, max_src_version)
        else:
            # Here we may be setting max_src_version to a lower value.
            # The "max" in max_src_version is to be intended as "the
            # max version among those with the highest priority".
            max_src_version = version
            max_priority = priority

    return src + "=" + str(max_src_version)


def acquire_source_from_dsc(dsc, source_dir, *, env, deadline):
    """Copy a .dsc and its source artifacts to the testbed and extract them."""
    dsc_tb = os.path.join(testbed.scratch, os.path.basename(dsc))
    TestbedPath(testbed, dsc, dsc_tb).copydown(chown=True)
    for part in files_from_dsc(dsc):
        TestbedPath(
            testbed, part, os.path.join(testbed.scratch, os.path.basename(part))
        ).copydown(chown=True)

    testbed.install_packages(("dpkg-dev",), mark_auto=True)
    result = testbed.execute(
        ["dpkg-source", "-x", dsc_tb, source_dir],
        check=False,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = f"source extraction failed with exit code {result.returncode}"
        adtlog.error(message)
        testbed.debug_shell(cwd=testbed.scratch, extra_env=env)
        adtlog.badpkg(message)


def acquire_source_from_tree(source, source_dir, *, env, deadline):
    """Copy a source tree to the testbed and apply its patches."""
    TestbedPath(testbed, source, source_dir).copydown(chown=True)
    testbed.install_packages(("dpkg-dev",), mark_auto=True)
    result = testbed.execute(
        ["dpkg-source", "--before-build", source_dir],
        check=False,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = f"source tree preparation failed with exit code {result.returncode}"
        adtlog.error(message)
        testbed.debug_shell(cwd=source_dir, extra_env=env)
        adtlog.badpkg(message)


def acquire_source_from_apt(source, source_dir, *, env, deadline):
    """Download the pinned source version and extract it in the testbed."""
    source_version = find_source_version_to_download(source)
    testbed.install_packages(("dpkg-dev",), mark_auto=True)

    try:
        testbed.run_apt_command(
            action="source",
            what=[source_version],
            extra_opts=["--download-only", "--only-source"],
            cwd=testbed.scratch,
            xenv=env,
        )
    except adtlog.AptDownloadError:
        message = f"APT failed to download source package {source_version}"
        adtlog.error(message)
        testbed.debug_shell(cwd=testbed.scratch, extra_env=env)
        testbed.bomb(message)
    except adtlog.AptPermanentError:
        message = f"APT source acquisition failed for {source_version}"
        adtlog.error(message)
        testbed.debug_shell(cwd=testbed.scratch, extra_env=env)
        adtlog.badpkg(message)

    result = testbed.execute(
        [
            "sh",
            "-ec",
            'exec dpkg-source -x "$1"_*.dsc "$2"',
            "extract-apt-source",
            source,
            source_dir,
        ],
        check=False,
        cwd=testbed.scratch,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = f"source extraction failed with exit code {result.returncode}"
        adtlog.error(message)
        testbed.debug_shell(cwd=testbed.scratch, extra_env=env)
        adtlog.badpkg(message)


def acquire_source_from_git(source, source_dir, *, env, deadline):
    """Clone a Git source in the testbed, optionally checking out a refspec."""
    url, _, branch = source.partition("#")
    testbed.install_packages(("git", "ca-certificates"), mark_auto=True)

    command = ["git", "clone", url, source_dir]
    result = testbed.execute(
        command,
        check=False,
        cwd=testbed.scratch,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = (
            f"Git source acquisition failed with exit code {result.returncode}: "
            f"{shlex.join(command)}"
        )
        adtlog.error(message)
        testbed.debug_shell(cwd=testbed.scratch, extra_env=env)
        adtlog.badpkg(message)

    if not branch:
        return

    command = ["git", "fetch", "-fu", "origin", f"{branch}:testbranch"]
    result = testbed.execute(
        command,
        check=False,
        cwd=source_dir,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = (
            f"Git source acquisition failed with exit code {result.returncode}: "
            f"{shlex.join(command)}"
        )
        adtlog.error(message)
        testbed.debug_shell(cwd=source_dir, extra_env=env)
        adtlog.badpkg(message)

    command = ["git", "checkout", "testbranch"]
    result = testbed.execute(
        command,
        check=False,
        cwd=source_dir,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = (
            f"Git source acquisition failed with exit code {result.returncode}: "
            f"{shlex.join(command)}"
        )
        adtlog.error(message)
        testbed.debug_shell(cwd=source_dir, extra_env=env)
        adtlog.badpkg(message)


def parse_tests(source_dir, *, opts):
    """Parse test descriptions using the run's options and testbed capabilities."""
    try:
        return testdesc.parse_debian_source(
            srcdir=source_dir,
            testbed_caps=testbed.caps,
            test_arch=testbed.test_arch,
            test_arch_is_foreign=testbed.test_arch_is_foreign,
            control_path=opts.override_control,
            auto_control=opts.auto_control,
            ignore_restrictions=opts.ignore_restrictions,
            only_tests=opts.only_tests,
        )
    except testdesc.InvalidControl as e:
        adtlog.badpkg(str(e))


def run_binary_build(
    source_dir: str,
    *,
    env: list[str],
    gainroot: str | None,
    build_parallel: str | None,
    deadline: float | None,
) -> None:
    """Install build dependencies and build binary packages in the testbed."""
    assert testbed is not None
    build_argv = [
        "dpkg-buildpackage",
        "-us",
        "-uc",
        "-b",
        "--source-option=--no-unapply-patches",
    ]
    assert testbed.nproc
    deb_build_profiles = ["noudeb", "nocheck"]
    # The "noddebs" option is deprecated in favor of "noautodbgsym",
    # but we keep it for compatibility with old debhelper versions.
    deb_build_options = [
        "nocheck",
        "noautodbgsym",
        "noddebs",
        f"parallel={build_parallel or testbed.nproc}",
    ]
    extra_build_deps: list[str] = []

    if gainroot:
        build_argv.append(f"-r{gainroot}")
    elif testbed.user or "root-on-testbed" not in testbed.caps:
        extra_build_deps.append("fakeroot")

    if testbed.test_arch_is_foreign:
        build_argv.append(f"--host-arch={testbed.test_arch}")
        deb_build_profiles.append("cross")
        extra_build_deps.append(f"crossbuild-essential-{testbed.test_arch}:native")
        # Install libc-dev and libstdc++-dev as a workaround for #815172. Also see:
        # https://salsa.debian.org/debian/sbuild/-/blob/d5fad46720b19e/lib/Sbuild/Build.pm#L838
        extra_build_deps.append(f"libc-dev:{testbed.test_arch}")
        extra_build_deps.append(f"libstdc++-dev:{testbed.test_arch}")

    build_deps_env = env.copy()
    for i, env_assignment in enumerate(build_deps_env):
        if env_assignment.startswith("DEB_BUILD_PROFILES="):
            build_deps_env[i] += " nocheck noudeb"
            break
    else:
        build_deps_env.append("DEB_BUILD_PROFILES=nocheck noudeb")
    testbed.install_build_deps_for_package(
        src_package_path=source_dir,
        extra_deps=extra_build_deps,
        xenv=build_deps_env,
    )

    # Keep patches applied for tests via --no-unapply-patches above.
    # Append the testbed's environment last, including autopkgtest --env values.
    adtlog.info("building binary packages")
    result = testbed.execute(
        [
            "sh",
            "-ec",
            """
                DEB_BUILD_OPTIONS="$1 $DEB_BUILD_OPTIONS"
                DEB_BUILD_PROFILES="$2 $DEB_BUILD_PROFILES"
                export DEB_BUILD_OPTIONS DEB_BUILD_PROFILES
                shift 2
                exec "$@"
            """,
            "binary-build",
            " ".join(deb_build_options),
            " ".join(deb_build_profiles),
            *build_argv,
        ],
        check=False,
        cwd=source_dir,
        user=testbed.user,
        xenv=env,
        stdout=sys.stderr,
        timeout=operation_timeout("build", deadline),
    )
    if result.returncode != 0:
        message = f"binary build failed with exit code {result.returncode}"
        adtlog.error(message)
        testbed.debug_shell(cwd=source_dir, extra_env=env)
        adtlog.badpkg(message)


def record_tested_package_version(source_dir, *, version_file=None):
    """Log the tested package version and save it in the output directory."""
    try:
        with open(os.path.join(source_dir, "debian", "changelog")) as f:
            m = re.match(r"(\S+) \(([^)]+)\)", f.readline())
    except OSError:
        m = None
    if not m:
        adtlog.badpkg("cannot parse debian/changelog")
    testpkg_name, testpkg_version = m.groups()
    adtlog.info(f"testing package {testpkg_name} version {testpkg_version}")
    if version_file is not None:
        with open(version_file, "w") as f:
            f.write(f"{testpkg_name} {testpkg_version}\n")


def copy_full_source(tree: TestbedPath) -> None:
    """Replace the runner's source copy with the complete testbed tree."""
    adtlog.info("copying full source tree from testbed to runner")
    shutil.rmtree(tree.host)
    tree.copyup()


def acquire_source(
    source_kind: str, source_path: str, source_dir: str, *, env, deadline
):
    """Acquire the requested kind of source in the testbed."""
    if source_kind == "source":
        acquire = acquire_source_from_dsc
    elif source_kind == "source-tree":
        acquire = acquire_source_from_tree
    elif source_kind == "apt-source":
        acquire = acquire_source_from_apt
    elif source_kind == "git-source":
        acquire = acquire_source_from_git
    else:
        adtlog.bomb("unknown source kind: " + source_kind)

    acquire(source_path, source_dir, env=env, deadline=deadline)


def prepare_source_for_testing(
    *, source_kind, source_path, binary_registry, opts, work_dir, deadline
):
    """Acquire source, build when required, and return the tree and discovery results."""
    blame(source_path)
    snapshots_enabled = "snapshot" in testbed.caps

    # Acquire the source tree to test
    source_dir = os.path.join(testbed.scratch, "tests-tree")
    acquire_source(
        source_kind, source_path, source_dir, env=opts.env, deadline=deadline
    )

    # Copy debian/ from testbed to host to access package metadata and test definitions
    tests_tree = TestbedPath(testbed, os.path.join(work_dir, "tests-tree"), source_dir)
    atexit.register(shutil.rmtree, tests_tree.host, ignore_errors=True)
    # debian/ may be a symlink.
    result = testbed.execute(
        ["readlink", "-fn", "--", os.path.join(tests_tree.tb, "debian")],
        check=True,
        stdout=subprocess.PIPE,
    )
    debian_tree = TestbedPath(
        testbed,
        os.path.join(tests_tree.host, "debian"),
        result.stdout,
    )
    debian_tree.copyup()

    # For simple validation we can stop here.
    if opts.validate:
        tests, some_skipped = parse_tests(tests_tree.host, opts=opts)
        return tests_tree, tests, some_skipped

    # Record the version of the package under test
    version_file = None
    if opts.output_dir:
        version_file = os.path.join(work_dir, "testpkg-version")
    record_tested_package_version(tests_tree.host, version_file=version_file)

    # autodep8 generates tests from the full source, not just debian/
    autodep8_requested = opts.auto_control and testdesc.needs_autodep8(tests_tree.host)
    if autodep8_requested:
        copy_full_source(tests_tree)

    # Discover tests
    tests, some_skipped = parse_tests(tests_tree.host, opts=opts)

    # Decide whether to build
    if opts.built_binaries:
        adtlog.info("build needed for binaries")
        build_needed = True
    elif any("build-needed" in test.restrictions for test in tests):
        adtlog.info("build needed for tests")
        build_needed = True
    else:
        adtlog.info("build not needed")
        build_needed = False

    # Build, then take the built tree to the runner
    if build_needed:
        if snapshots_enabled:
            testbed.snapshot_create("pre-build")
        run_binary_build(
            source_dir,
            env=opts.env,
            gainroot=opts.gainroot,
            build_parallel=opts.build_parallel,
            deadline=deadline,
        )
        if opts.built_binaries:
            binary_registry.collect_built(tests_tree.tb)
        copy_full_source(tests_tree)

        if snapshots_enabled:
            # Roll back build changes and replace the unbuilt tree with the built one.
            testbed.snapshot_restore("pre-build")
            testbed.execute(
                ["rm", "-rf", "--", tests_tree.tb],
                check=True,
                capture_output=True,
                allow_stderr=False,
            )
            adtlog.info("copying built source tree to testbed")
            tests_tree.copydown(chown=True)
            testbed.snapshot_delete("pre-build")
    elif not snapshots_enabled and not autodep8_requested:
        # Without snapshots, each test needs a fresh copydown from the runner
        copy_full_source(tests_tree)

    return tests_tree, tests, some_skipped


def select_tests(
    tests: list[testdesc.Test],
    *,
    source_kind: str,
    source_path: str,
    only_tests: list[str] | None,
    skip_tests: list[str] | None,
) -> list[testdesc.Test]:
    """Apply test-name selection, then exclude explicitly skipped tests."""
    if only_tests:
        adtlog.debug(
            f"only running {only_tests} for package {source_kind} {source_path}"
        )
        tests = [t for t in tests if t.name in only_tests]
        if not tests:
            adtlog.error(
                f"{source_kind} {source_path} has no test matching --test-name {', '.join(only_tests)}"
            )

    if skip_tests:
        adtlog.debug(
            f'filtering out "{" ".join(skip_tests)}" for package {source_kind} {source_path}'
        )
        tests = [t for t in tests if t.name not in skip_tests]

    return tests


def test_source_package(
    *,
    source_kind: str,
    source_path: str,
    binary_registry,
    opts,
    work_dir: str,
    deadline: float | None,
) -> int:
    """Prepare the single source and test it using the registered binaries."""

    adtlog.info(f"@@@@@@@@@@@@@@@@@@@@ {source_kind} {source_path}")

    tests_tree, tests, skipped = prepare_source_for_testing(
        source_kind=source_kind,
        source_path=source_path,
        binary_registry=binary_registry,
        opts=opts,
        work_dir=work_dir,
        deadline=deadline,
    )
    if opts.validate:
        adtlog.report("*", "Test specification is valid")
        return 0

    tests = select_tests(
        tests,
        source_kind=source_kind,
        source_path=source_path,
        only_tests=opts.only_tests,
        skip_tests=opts.skip_tests,
    )
    if not tests and not skipped:
        adtlog.report("*", "SKIP no tests in this package")

    errorcode = run_tests(
        tests,
        tree=tests_tree,
        binary_registry=binary_registry,
        env=opts.env,
        shell=opts.shell,
        build_parallel=opts.build_parallel,
        fail_fast=opts.fail_fast,
        deadline=deadline,
    )
    if skipped and errorcode != 8:
        errorcode |= 2

    assert adtlog.summary_stream is not None
    adtlog.summary_stream.flush()
    if adtlog.verbosity >= 1:
        adtlog.summary_stream.seek(0)
        adtlog.info("@@@@@@@@@@@@@@@@@@@@ summary")
        sys.stderr.buffer.write(adtlog.summary_stream.read())
    adtlog.summary_stream.close()
    adtlog.summary_stream = None

    return errorcode


def main():
    global testbed
    try:
        (opts, vserver_args) = parse_args()
    except SystemExit:
        # argparser exits with error 2 by default, but we have a different
        # meaning for that already
        sys.exit(20)

    binary_registry = None
    work_dir = None
    keep_output = opts.output_dir is not None

    # ensure proper cleanup on signals
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGQUIT, signal_handler)

    try:
        if keep_output:
            os.makedirs(opts.output_dir, exist_ok=True)
            if os.listdir(opts.output_dir):
                adtlog.bomb(f'--output-dir "{opts.output_dir}" is not empty')
            work_dir = opts.output_dir
        else:
            work_dir = tempfile.mkdtemp(prefix="autopkgtest.output.")
            os.chmod(work_dir, 0o755)

        logfile = opts.logfile
        if logfile is None and keep_output:
            logfile = os.path.join(work_dir, "log")
        summary = opts.summary
        if summary is None:
            summary = os.path.join(work_dir, "summary")
        setup_trace(logfile=logfile, summary=summary)
        deadline = None
        if opts.timeout > 0:
            deadline = time.monotonic() + opts.timeout
        testbed = Testbed(
            vserver_argv=vserver_args,
            output_dir=work_dir,
            user=opts.user,
            shell_fail=opts.shell_fail,
            setup_commands=opts.setup_commands,
            setup_commands_boot=opts.setup_commands_boot,
            add_apt_pockets=opts.apt_pocket,
            copy_files=opts.copy,
            enable_apt_fallback=opts.enable_apt_fallback,
            needs_internet=opts.needs_internet,
            add_apt_sources=opts.add_apt_sources,
            add_apt_releases=opts.add_apt_releases,
            pin_packages=opts.pin_packages,
            apt_default_release=opts.apt_default_release,
            apt_upgrade=opts.apt_upgrade,
            test_arch=opts.test_architecture,
        )
        testbed.start()
        testbed.open()
        binary_registry = adt_binaries.DebBinaries(testbed, work_dir)
        for path in opts.provided_binary_paths:
            blame("arg:" + path)
            pkg = deb_package_name(path)
            blame("deb:" + pkg)
            binary_registry.register(path, pkg)
        errorcode = test_source_package(
            source_kind=opts.source_kind,
            source_path=opts.source_path,
            binary_registry=binary_registry,
            opts=opts,
            work_dir=work_dir,
            deadline=deadline,
        )
    except Exception:  # noqa: BLE001
        errorcode = print_exception(sys.exc_info(), "")
    if work_dir is not None and testbed is not None:
        try:
            create_testinfo(vserver_args, env=opts.env, work_dir=work_dir)
        except Exception:  # noqa: BLE001
            errorcode = print_exception(sys.exc_info(), "")
    try:
        cleanup(
            binary_registry=binary_registry, keep_output=keep_output, work_dir=work_dir
        )
    except Exception:  # noqa: BLE001
        # A testbed failure (16) means the testbed may be unusable, so we
        # cannot assume we can even tear it down; preserve that code and stay
        # quiet. Any other prior state is reported as a generic cleanup error.
        if errorcode != 16:
            print_exception(
                sys.exc_info(),
                f"{os.path.basename(sys.argv[0])}\n: error cleaning up:\n",
            )
            errorcode = 20
    sys.exit(errorcode)


main()
