# adt_testbed.py is part of autopkgtest
# autopkgtest is a tool for testing Debian binary packages
#
# autopkgtest is Copyright (C) 2006-2015 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 os
import sys
import errno
import json
import time
import traceback
import re
import shlex
import signal
import subprocess
import tempfile
import textwrap
import shutil
from string import Template
import urllib.parse
import urllib.request
from typing import (
    List,
    Optional,
    Set,
    Tuple,
)

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

import adtlog
import VirtSubproc
from VirtSubproc import load_shell_script
from testdesc import Unsupported


timeouts = {"short": 100, "copy": 300, "install": 3000, "test": 10000, "build": 100000}
global_timeout = 0


# When running installed, this is /usr/share/autopkgtest.
# When running uninstalled, this is the source tree.
# Either way, it has a setup-commands subdirectory.
PKGDATADIR = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))


class Testbed:
    SOURCELIST_PATTERN = re.compile(
        r"(?P<type>deb|deb-src)\s+"
        + r"(?:(?P<options>\[[^]]+\])\s+)?"
        + r"(?P<uri>[^\s]+)\s+"
        + r"(?P<suite>[^\s]+)\s+"
        + r"(?P<components>.+)$"
    )

    # sample pattern: ppa:user:token@owner/name:fingerprint
    PPA_PATTERN = re.compile(
        # maybe match literal ppa: at the beginning of string, case insensitive
        r"^(?i:ppa:)?"
        # maybe match user:token for private PPA authentication;
        # for the token accept all the RFC 3986 unreserved characters
        r"((?P<user>[a-z0-9](-?[a-z0-9])+):(?P<token>[a-zA-Z0-9._~-]+)@)?"
        # match PPA owner, following most Launchpad rules for usernames
        r"(?P<owner>[a-z0-9](-?[a-z0-9])+)"
        # match PPA name, following most Launchpad rules for PPA names
        r"/(?P<name>[a-z0-9][a-zA-Z0-9.+-]*)"
        # maybe match fingerprint, discarding leading 0x if present, case insensitive
        r"(?i::(0x)?(?P<fingerprint>[0-9a-f]{40}))?$"
    )

    def __init__(
        self,
        *,
        vserver_argv: List[str],
        output_dir: str,
        test_arch: Optional[str] = None,
        user: str,
        setup_commands: List[str],
        setup_commands_boot: List[str],
        add_apt_pockets: List[str],
        copy_files: List[str],
        pin_packages: List[str],
        add_apt_sources: List[str],
        add_apt_releases: List[str],
        apt_default_release: Optional[str] = None,
        apt_upgrade: bool = False,
        enable_apt_fallback: bool = True,
        shell_fail: bool = False,
        needs_internet: str = "run",
        insecure: bool = False,
    ) -> None:
        self.sp = None
        self.lastsend = None
        self.scratch = None
        self.modified = False
        self._need_reset_apt = False
        self.stop_sent = False
        self.dpkg_arch = None
        self.test_arch = test_arch
        self.exec_cmd = None
        self.output_dir = output_dir
        self.shared_downtmp = None  # testbed's downtmp on the host, if supported
        self.vserver_argv = vserver_argv
        self.user = user
        self.setup_commands = setup_commands
        self.setup_commands_boot = setup_commands_boot
        self.add_apt_pockets = add_apt_pockets
        self.add_apt_sources = add_apt_sources
        self.add_apt_releases = add_apt_releases
        self.pin_packages = pin_packages
        self.default_release = apt_default_release
        self.apt_upgrade = apt_upgrade
        self.copy_files = copy_files
        self.initial_kernel_version = None
        self.insecure = insecure
        # tests might install a different kernel; [(testname, reboot_marker, kver)]
        self.test_kernel_versions: List[Tuple[str, str, str]] = []
        # used for tracking kernel version changes
        self.last_test_name = ""
        self.last_reboot_marker = ""
        self.eatmydata_prefix = None
        self.apt_pin_for_releases: List[str] = []
        self.enable_apt_fallback = enable_apt_fallback
        self.apt_version: Optional[Version] = None
        self.release: Optional[str] = None
        self.needs_internet = needs_internet
        self.shell_fail = shell_fail
        self.nproc = None
        self.cpu_model = None
        self.cpu_flags = None
        self._created_user = False
        self._tried_debug = False
        # False if not tried, True if successful, error message if not
        self._tried_provide_sudo = False
        # used for keeping track of the global timeout, if any
        self.start_time = int(time.time())
        self.skip_remaining_tests = False
        # Absolute path to wrapper.sh on testbed
        self.wrapper_sh = ""

        adtlog.debug("testbed init")

    @property
    def su_user(self):
        return self.user or "root"

    def start(self):
        # log date at least once; to ease finding it
        adtlog.info("starting date and time: %s" % time.strftime("%Y-%m-%d %H:%M:%S%z"))

        # are we running from a checkout?
        root_dir = PKGDATADIR
        if os.path.exists(os.path.join(root_dir, ".git")):
            try:
                head = subprocess.check_output(
                    ["git", "show", "--no-patch", "--oneline"], cwd=root_dir
                )
                head = head.decode("UTF-8").strip()
            except Exception as e:
                adtlog.debug(str(e))
                head = "cannot determine current HEAD"
            adtlog.info("git checkout: %s" % head)
        else:
            adtlog.info("version @version@")

        # log command line invocation for the log
        adtlog.info(
            "host %s; command line: %s"
            % (os.uname()[1], " ".join([shlex.quote(w) for w in sys.argv]))
        )

        self.sp = subprocess.Popen(
            self.vserver_argv,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            universal_newlines=True,
        )
        self.expect("ok", 0)

    def stop(self):
        adtlog.debug("testbed stop")
        if self.stop_sent:
            # avoid endless loop
            return
        self.stop_sent = True

        self.close()
        if self.sp is None:
            return
        ec = self.sp.returncode
        if ec is None:
            self.sp.stdout.close()
            self.send("quit")
            self.sp.stdin.close()
            ec = self.sp.wait()
        if ec:
            self.debug_fail()
            self.bomb("testbed gave exit status %d after quit" % ec)
        self.sp = None

    def open(self):
        adtlog.debug("testbed open, scratch=%s" % self.scratch)
        if self.scratch is not None:
            return
        pl = self.command("open", (), 1)
        self._opened(pl)

    def post_boot_setup(self):
        """Setup after (re)booting the test bed"""

        # provide autopkgtest-reboot command, if reboot is supported; /run is
        # usually "noexec" and /[s]bin might be readonly, so create in /tmp
        if "reboot" in self.caps and "root-on-testbed" in self.caps:
            adtlog.debug("testbed supports reboot, creating /tmp/autopkgtest-reboot")
            reboot = TestbedPath(
                self,
                os.path.join(PKGDATADIR, "lib", "in-testbed", "reboot.sh"),
                "%s/autopkgtest-reboot" % self.scratch,
                False,
            )
            reboot.copydown(mode="0755")
            # TODO: Because we're now creating a symlink, it would be OK
            # to put the symlink in /run again, even though it's noexec,
            # because the actual content is elsewhere
            self.check_exec(
                ["ln", "-fns", reboot.tb, "/tmp/autopkgtest-reboot"],
            )
            # Intentionally not check_exec, and intentionally silencing stderr:
            # /[s]bin might be read-only.
            self.execute(
                ["ln", "-fns", reboot.tb, "/sbin/autopkgtest-reboot"],
                stderr=subprocess.PIPE,
            )

            reboot_prepare = TestbedPath(
                self,
                os.path.join(PKGDATADIR, "lib", "in-testbed", "reboot-prepare.sh"),
                "%s/autopkgtest-reboot-prepare" % self.scratch,
                False,
            )
            reboot_prepare.copydown(mode="0755")
            self.check_exec(
                [
                    "ln",
                    "-fns",
                    reboot_prepare.tb,
                    "/tmp/autopkgtest-reboot-prepare",
                ],
            )

        # record running kernel version
        kver = self.check_exec(["uname", "-srv"], True).strip()
        if not self.initial_kernel_version:
            assert not self.last_test_name
            self.initial_kernel_version = kver
            adtlog.info("testbed running kernel: " + self.initial_kernel_version)
        else:
            if kver != self.initial_kernel_version:
                self.test_kernel_versions.append(
                    (self.last_test_name, self.last_reboot_marker, kver)
                )
                adtlog.info(
                    "testbed running kernel changed: %s (current test: %s%s)"
                    % (
                        kver,
                        self.last_test_name,
                        self.last_reboot_marker
                        and (", last reboot marker: " + self.last_reboot_marker)
                        or "",
                    )
                )

        # get CPU info
        if self.nproc is None:
            cpu_info = self.check_exec(
                ["sh", "-c", "nproc; cat /proc/cpuinfo 2>/dev/null || true"],
                stdout=True,
            ).strip()
            self.nproc = cpu_info.split("\n", 1)[0]
            m = re.search(
                r"^(model.*name|cpu)\s*:\s*(.*)$",
                cpu_info,
                re.MULTILINE | re.IGNORECASE,
            )
            if m:
                self.cpu_model = m.group(2)
            m = re.search(
                r"^(flags|features)\s*:\s*(.*)$", cpu_info, re.MULTILINE | re.IGNORECASE
            )
            if m:
                self.cpu_flags = m.group(2)

        xenv = ["AUTOPKGTEST_IS_SETUP_BOOT_COMMAND=1"]
        if self.user:
            xenv.append("AUTOPKGTEST_NORMAL_USER=" + self.user)
            xenv.append("ADT_NORMAL_USER=" + self.user)

        for c in self.setup_commands_boot:
            rc = self.execute(["sh", "-ec", c], xenv=xenv, kind="install")[0]
            if rc:
                # setup scripts should exit with 100 if it's the package's
                # fault, otherwise it's considered a transient testbed failure
                if self.shell_fail:
                    self.run_shell()
                if rc == 100:
                    self.badpkg("testbed boot setup commands failed with status 100")
                else:
                    self.bomb("testbed boot setup commands failed with status %i" % rc)

    def _opened(self, pl):
        self._tried_provide_sudo = False
        self.scratch = pl[0]
        self.deps_installed = []
        self.apt_pin_for_releases = []
        self.exec_cmd = list(
            map(
                urllib.parse.unquote,
                self.command("print-execute-command", (), 1)[0].split(","),
            )
        )
        self.caps = self.command("capabilities", (), None)
        if self.needs_internet in ["try", "run"]:
            self.caps.append("has_internet")
        adtlog.debug("testbed capabilities: %s" % self.caps)
        for c in self.caps:
            if c.startswith("downtmp-host="):
                self.shared_downtmp = c.split("=", 1)[1]

        wrapper_sh = TestbedPath(
            self,
            os.path.join(PKGDATADIR, "lib", "in-testbed", "wrapper.sh"),
            "%s/wrapper.sh" % self.scratch,
            False,
        )
        wrapper_sh.copydown(mode="0755")
        self.wrapper_sh = wrapper_sh.tb

        # provide a default for --user
        if self.user is None and "root-on-testbed" in self.caps:
            self.user = ""
            for c in self.caps:
                if c.startswith("suggested-normal-user="):
                    self.user = c.split("=", 1)[1]

            if "revert-full-system" in self.caps and not self.user:
                with open(
                    os.path.join(PKGDATADIR, "setup-commands", "create-normal-user"),
                    encoding="UTF-8",
                ) as reader:
                    self.check_exec(["sh", "-euc", reader.read()])
                    self._created_user = True
                    self.user = self.check_exec(
                        ["cat", "/run/autopkgtest-normal-user"],
                        stdout=True,
                    ).strip()
        elif self._created_user:
            # Re-create the same user after reset. We only do this if
            # we created it before the reset, in which case the capabilities
            # should be the same as above and the same username should still
            # be available.
            assert "revert-full-system" in self.caps, self.caps
            assert "root-on-testbed" in self.caps, self.caps
            with open(
                os.path.join(PKGDATADIR, "setup-commands", "create-normal-user"),
                encoding="UTF-8",
            ) as reader:
                self.check_exec(["sh", "-euc", reader.read(), "sh", self.user])

        if (
            self.su_user is not None
            and self.su_user != "root"
            and not any(
                c in ["isolation-machine", "isolation-container"] for c in self.caps
            )
        ):
            # scratch directory is owned by root and not world-writeable.
            # (See the rules in README.virtualisation-server.)
            #
            # But, we like to run things (particularly, tests) as a normal user.
            # That normal user needs to be able to write to this directory.
            #
            # Once upon a time, this was possible because the scratch directory was 1777.
            # Nowadays, we make it possible by making it owned by the normal user.
            #
            # This means that we make root on the testbed completely trust self.user.
            # This is OK because either:
            #   self.user came from the testbed capability suggested-normal-user=
            #   self.user came from the --user command line option
            # and in both cases this trust relationship is documented there.
            self.check_exec(["chown", self.su_user, "--", self.scratch])

        # determine testbed architecture
        self.dpkg_arch = self.check_exec(["dpkg", "--print-architecture"], True).strip()
        adtlog.info("testbed dpkg architecture: " + self.dpkg_arch)

        # the test architecture defaults to the dpkg architecture
        if not self.test_arch:
            self.test_arch = self.dpkg_arch

        self.test_arch_is_foreign = self.test_arch != self.dpkg_arch

        out = self.check_exec(
            ["dpkg-query", "-W", "-f", "${Version}", "apt"], True
        ).strip()
        adtlog.info("testbed apt version: " + out)
        try:
            self.apt_version = Version(out)
        except ValueError:
            self.bomb("can't determine the apt version on the testbed")

        # Foreign arch testing requires `apt-get satisfy`.
        if self.test_arch_is_foreign and self.apt_version < Version("1.9.0"):
            self.bomb(
                "foreign arch testing requires apt >= 1.9.0 on the testbed",
                adtlog.AutopkgtestError,
            )

        # Add foreign test architecture
        if self.test_arch_is_foreign:
            self.check_exec(["dpkg", "--add-architecture", self.test_arch])

        # do we have eatmydata?
        (code, out, err) = self.execute(
            ["sh", "-ec", "command -v eatmydata"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if code == 0:
            adtlog.debug("testbed has eatmydata")
            self.eatmydata_prefix = out.strip()

        # ensure that /etc/apt/preferences.d exists
        self.check_exec(["mkdir", "-p", "/etc/apt/preferences.d"])

        self.run_setup_commands()

        # record package versions of pristine testbed
        if self.output_dir:
            pkglist = TempPath(self, "testbed-packages", autoclean=False)
            self.check_exec(
                [
                    "sh",
                    "-ec",
                    "dpkg-query --show -f '${Package}\\t${Version}\\n' > %s"
                    % pkglist.tb,
                ]
            )
            pkglist.copyup()

        self.post_boot_setup()

    def close(self):
        adtlog.debug("testbed close, scratch=%s" % self.scratch)
        if self.scratch is None:
            return
        self.scratch = None
        if self.sp is None:
            return
        self.command("close")
        self.shared_downtmp = None

    def reboot(self, prepare_only=False):
        """Reboot the testbed"""

        self.command("reboot", prepare_only and ("prepare-only",) or ())
        self.post_boot_setup()

    def run_setup_commands(self):
        """Run --setup-commmands and --copy"""

        adtlog.info("@@@@@@@@@@@@@@@@@@@@ test bed setup")
        for host, tb in self.copy_files:
            adtlog.debug("Copying file %s to testbed %s" % (host, tb))
            TestbedPath(self, host, tb, os.path.isdir(host)).copydown()

        # Pin the default release if explicitly requested.
        if self.default_release:
            self._set_default_release()

        if (
            self.apt_upgrade
            or self.add_apt_sources
            or self.add_apt_pockets
            or self.add_apt_releases
        ):
            # Update APT package index. We can't run this unconditionally because
            # we want to support not-root-on-testned and offline use cases.
            adtlog.info("updating testbed package index (apt update)")
            try:
                self._run_apt_command(action="update")
            except adtlog.AptError:
                self.bomb("failed testbed apt update")

        # Assume the testbed release to be the one that provides the candidate
        # base-files version. This requires the package index to be up-to-date.
        _, _, self.release = self.get_candidate("base-files")
        if self.release:
            adtlog.info(f"testbed release detected to be: {self.release}")
        else:
            adtlog.info(
                "could not determine the testbed release (no APT source for base-files?)"
            )
        if not self.release and (
            self.add_apt_releases
            or self.add_apt_pockets
            or any(self.PPA_PATTERN.match(s) for s in self.add_apt_sources)
        ):
            self.bomb(
                "testbed release required to be known for autopkgtest to run",
                adtlog.AutopkgtestError,
            )

        pin_packages = self.pin_packages.copy()
        add_apt_releases = self.add_apt_releases.copy()
        for pocket in self.add_apt_pockets:
            pocket = pocket.strip()
            if not pocket:
                continue
            (pocket, _, pkglist) = pocket.partition("=")
            pocket_suite = self.release + "-" + pocket

            add_apt_releases.append(pocket_suite)

            if pkglist:
                # create apt pinning for --apt-pocket=pocket=packages
                pin_packages.append(f"{pocket_suite}={pkglist}")

        # clean list and remove duplicates
        add_apt_releases = [r.strip() for r in add_apt_releases]
        add_apt_releases = [r for r in add_apt_releases if r]
        add_apt_releases = list(dict.fromkeys(add_apt_releases))
        # add releases
        for release in add_apt_releases:
            self._add_apt_release_one_line_style(release)
            self._add_apt_release_deb822(release)

            # Set baseline "general form" pin to neutralize NotAutomatic: yes.
            self._set_baseline_pin_priority_for_release(release)

        # To be done after adding add_apt_releases, so that adding releases/pocket
        # is only done based on sources the image came configured with (LP: #2091393).
        for source in self.add_apt_sources:
            if self.SOURCELIST_PATTERN.match(source):
                self._add_apt_source(
                    source, filename="autopkgtest-add-apt-sources.list"
                )
            elif self.PPA_PATTERN.match(source):
                self._add_apt_ppa(source)
            elif (
                os.path.isfile(source)
                and os.access(source, os.R_OK)
                and (source.endswith(".list") or source.endswith(".sources"))
            ):
                with open(source, "r") as s:
                    self._add_apt_source(
                        s.read(),
                        filename=os.path.basename(source),
                    )
            else:
                self.bomb(f"invalid apt source: {source}", adtlog.AutopkgtestError)

        # apt-get update, to be done after adding sources, releases, pockets,
        # but before setting up pinning for apt (<< 1.9.11) compatibility, as
        # it requires src:package expansion via apt-cache dumpavail.
        if self.apt_upgrade or self.add_apt_sources or add_apt_releases:
            adtlog.info("updating testbed package index (apt update)")
            try:
                self._run_apt_command(action="update")
            except adtlog.AptError:
                self.bomb("failed testbed apt update")

        # remove duplicates
        pin_packages = list(dict.fromkeys(pin_packages))
        # create apt pinning for --pin-packages and --apt-pocket, to be done
        # after updating the index (apt-get update) to allow expansion of
        # src:package pins via apt-cache dumpavail.
        for package_set in pin_packages:
            (release, pkglist) = package_set.split("=", 1)
            self._create_apt_pinning_for_packages(release, pkglist)

        # record the mtimes of dirs affecting the boot
        boot_dirs = " ".join(
            [
                "/boot",
                "/boot/efi",
                "/boot/grub",
                "/etc/init",
                "/etc/init.d",
                "/etc/systemd/system",
                "/lib/systemd/system",
            ]
        )
        self.check_exec(
            [
                "bash",
                "-ec",
                r"for d in %s; do [ ! -d $d ] || touch -r $d %s/${d//\//_}.stamp; done"
                % (boot_dirs, self.scratch),
            ]
        )

        xenv = ["AUTOPKGTEST_IS_SETUP_COMMAND=1"]
        if self.user:
            xenv.append("AUTOPKGTEST_NORMAL_USER=" + self.user)
            xenv.append("ADT_NORMAL_USER=" + self.user)

        for c in self.setup_commands:
            rc = self.execute(["sh", "-ec", c], xenv=xenv, kind="install")[0]
            if rc:
                # setup scripts should exit with 100 if it's the package's
                # fault, otherwise it's considered a transient testbed failure
                if self.shell_fail:
                    self.run_shell()
                if rc == 100:
                    self.badpkg("testbed setup commands failed with status 100")
                else:
                    self.bomb("testbed setup commands failed with status %i" % rc)

        if self.apt_upgrade:
            adtlog.info("upgrading testbed (apt dist-upgrade and autopurge)")
            try:
                self._run_apt_command(action="dist-upgrade")
                self._run_apt_command(action="autoremove", extra_opts=["--purge"])
            except adtlog.AptError:
                # If apt fails at this stage always consider it a testbed failure,
                # so bomb() on all AptError exceptions, including AptPermanentFailure.
                self.bomb("failed testbed apt dist-upgrade or autopurge")

        # if the setup commands affected the boot, then reboot
        if self.setup_commands and "reboot" in self.caps:
            boot_affected = self.execute(
                [
                    "bash",
                    "-ec",
                    r"[ ! -e /run/autopkgtest_no_reboot.stamp ] || exit 0;"
                    r"for d in %s; do s=%s/${d//\//_}.stamp;"
                    r"  [ ! -d $d ] || [ `stat -c %%Y $d` = `stat -c %%Y $s` ]; done"
                    % (boot_dirs, self.scratch),
                ]
            )[0]
            if boot_affected:
                adtlog.info("rebooting testbed after setup commands that affected boot")
                self.reboot()

    def reset(self, deps_new):
        """Reset the testbed, if possible and necessary"""

        adtlog.debug(
            "testbed reset: modified=%s, deps_installed=%s, deps_new=%s"
            % (self.modified, self.deps_installed, deps_new)
        )
        if "revert" in self.caps and (
            self.modified or [d for d in self.deps_installed if d not in deps_new]
        ):
            adtlog.debug("testbed reset")
            pl = self.command("revert", (), 1)
            self._opened(pl)
        self.modified = False

    def install_deps(
        self,
        deps_new: List[List[str]],
        shell_on_failure: bool = False,
        package_under_test_deps: Optional[List[List[str]]] = None,
    ) -> None:
        """
        Install test dependencies into testbed.
        """

        adtlog.debug("install_deps: deps_new=%s" % deps_new)

        self.deps_installed = deps_new
        if not deps_new:
            return

        self.satisfy_dependencies_string(
            ", ".join([" | ".join(altgroup) for altgroup in deps_new]),
            "install-deps",
            shell_on_failure=shell_on_failure,
            package_under_test_deps=package_under_test_deps,
        )

    def _provide_sudo(self) -> str:
        """
        Enable the needs-sudo restriction, or return why we can't.
        """

        if "root-on-testbed" in self.caps and not self.user:
            return "Cannot enable needs-sudo restriction: no ordinary user available"

        # This is intentionally similar to how run_test() runs it, to make
        # sure that we can sudo here if and only if we could sudo in the test.
        if "root-on-testbed" in self.caps:
            run_as_user = ["su", "-s", "/bin/bash", self.user, "-c"]
        else:
            run_as_user = ["bash", "-c"]

        # First check whether the test user is in the sudo group
        rc, out, err = self.execute(
            run_as_user + ["id -Gn"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        if rc == 0 and "sudo" in out.split():
            adtlog.debug(
                'User "%s" groups include sudo (%s)' % (self.user, out.strip())
            )
            # Also check whether they need a password (by default yes,
            # but in e.g. Ubuntu cloud images they don't)
            rc, out, err = self.execute(
                run_as_user + ["sudo -n true"],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            adtlog.debug("sudo -n true stdout: %s" % out.strip())
            adtlog.debug("sudo -n true stderr: %s" % err.strip())
            adtlog.debug("sudo -n true status: %d" % rc)
            already_has_sudo = rc == 0
        else:
            adtlog.debug("id -Gn stdout: %s" % out.strip())
            adtlog.debug("id -Gn stderr: %s" % err.strip())
            adtlog.debug("id -Gn status: %d" % rc)
            already_has_sudo = False

        if already_has_sudo:
            adtlog.debug('User "%s" can already sudo without password' % self.user)
            return ""

        if "root-on-testbed" not in self.caps:
            return "Cannot enable needs-sudo restriction: not root"

        if "revert-full-system" not in self.caps:
            return "Cannot enable needs-sudo restriction: cannot revert"

        self.needs_reset()
        adtlog.info('Setting up user "%s" to sudo without password...' % self.user)

        with open(os.path.join(PKGDATADIR, "setup-commands", "enable-sudo")) as reader:
            script = reader.read()

        rc, _, _ = self.execute(
            ["sh", "-euc", script, "enable-sudo", self.user],
        )

        if rc != 0:
            return ("Failed to enable needs-sudo restriction: exit status %d") % rc

        return ""

    def satisfy_restrictions(
        self,
        name: str,
        restrictions: Set[str],
    ) -> None:
        """
        Try to satisfy restrictions that can be set up dynamically.
        Raise Unsupported if unable to do so.
        """

        if "needs-sudo" in restrictions:
            if not self._tried_provide_sudo:
                no_sudo_reason = self._provide_sudo()
                if no_sudo_reason:
                    raise Unsupported(name, no_sudo_reason)
                else:
                    self._tried_provide_sudo = True

    def needs_reset(self):
        # show what caused a reset
        (fname, lineno, function, code) = traceback.extract_stack(limit=2)[0]
        adtlog.debug(
            "needs_reset, previously=%s, requested by %s() line %i"
            % (self.modified, function, lineno)
        )
        self.modified = True

    def bomb(self, m, _type=adtlog.TestbedFailure):
        adtlog.debug("%s %s" % (_type.__name__, m))
        self.stop()
        raise _type(m)

    def badpkg(self, m):
        _type = adtlog.BadPackageError
        adtlog.debug("%s %s" % (_type.__name__, m))
        raise _type(m)

    def debug_fail(self):
        # Avoid recursion, and also avoid repeating ourselves if cleanup
        # fails after we already tried to debug a failure
        if self._tried_debug:
            return

        self._tried_debug = True

        # Many reasons for failure will make auxverb_debug_fail also fail.
        # Ignore that, to preserve the original exception
        try:
            self.command("auxverb_debug_fail")
        except Exception as e:
            adtlog.debug("auxverb_debug_fail failed: %s" % e)

    def send(self, string):
        try:
            adtlog.debug("sending command to virtualisation server: " + string)
            self.sp.stdin.write(string)
            self.sp.stdin.write("\n")
            self.sp.stdin.flush()
            self.lastsend = string
        except Exception as e:
            self.debug_fail()
            self.bomb("cannot send to virtualisation server: %s" % e)

    def expect(self, keyword, nresults):
        line = self.sp.stdout.readline()
        if not line:
            self.debug_fail()
            self.bomb("eof from the virtualisation server")
        if not line.endswith("\n"):
            self.bomb("unterminated line from the virtualisation server")
        line = line.rstrip("\n")
        adtlog.debug("got reply from testbed: " + line)
        ll = line.split()
        if not ll:
            self.bomb("unexpected whitespace-only line from the virtualisation server")
        if ll[0] != keyword:
            self.debug_fail()

            if self.lastsend is None:
                self.bomb("got banner `%s', expected `%s...'" % (line, keyword))
            else:
                self.bomb(
                    "sent `%s', got `%s', expected `%s...'"
                    % (self.lastsend, line, keyword)
                )
        ll = ll[1:]
        if nresults is not None and len(ll) != nresults:
            self.bomb(
                "sent `%s', got `%s' (%d result parameters),"
                " expected %d result parameters"
                % (self.lastsend, line, len(ll), nresults)
            )
        return ll

    def command(self, cmd, args=(), nresults=0, unquote=True):
        # pass args=[None,...] or =(None,...) to avoid more url quoting
        if type(cmd) is str:
            cmd = [cmd]
        if len(args) and args[0] is None:
            args = args[1:]
        else:
            args = list(map(urllib.parse.quote, args))
        al = cmd + args
        self.send(" ".join(al))
        ll = self.expect("ok", nresults)
        if unquote:
            ll = list(map(urllib.parse.unquote, ll))
        return ll

    def execute(self, argv, xenv=[], stdout=None, stderr=None, kind="short"):
        """Run command in testbed.

        The commands stdout/err will be piped directly to autopkgtest and its log
        files, unless redirection happens with the stdout/stderr arguments
        (passed to Popen).

        Return (exit code, stdout, stderr). stdout/err will be None when output
        is not redirected.
        """

        adtlog.debug(
            "testbed command %s, kind %s, sout %s, serr %s, env %s"
            % (argv, kind, stdout and "pipe" or "raw", stderr and "pipe" or "raw", xenv)
        )

        if xenv:
            argv = ["env"] + xenv + argv

        abort_after_timeout = False
        timeout = timeouts[kind]
        timeout_type = "timed out"
        if kind in ["test", "build"] and global_timeout > 0:
            remaining_time = self.start_time + global_timeout - int(time.time())
            if remaining_time <= 0:
                adtlog.error("global timeout reached even before %s" % " ".join(argv))
                raise VirtSubproc.Timeout(global_timeout, abort=True)
            if timeout > remaining_time:
                timeout = remaining_time
                timeout_type = "global timeout reached"
                abort_after_timeout = True

        VirtSubproc.timeout_start(timeout)
        try:
            proc = subprocess.Popen(
                self.exec_cmd + argv,
                stdin=subprocess.DEVNULL,
                stdout=stdout,
                stderr=stderr,
            )
            (out, err) = proc.communicate()
            if out is not None:
                out = out.decode()
            if err is not None:
                err = err.decode()
            VirtSubproc.timeout_stop()
        except VirtSubproc.Timeout as exc:
            # This is a bit of a hack, but what can we do.. we can't kill/clean
            # up sudo processes, we can only hope that they clean up themselves
            # after we stop the testbed
            killtree(proc.pid)
            adtlog.debug(
                "%s on %s %s (kind: %s)" % (timeout_type, self.exec_cmd, argv, kind)
            )
            if "sudo" not in self.exec_cmd:
                try:
                    proc.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    killtree(proc.pid, signal.SIGKILL)
                    proc.wait()
            msg = '%s on command "%s" (kind: %s)' % (timeout_type, " ".join(argv), kind)
            if kind == "test":
                adtlog.error(msg)
                exc.abort = abort_after_timeout
                raise exc
            else:
                self.debug_fail()
                self.bomb(msg)

        adtlog.debug("testbed command exited with code %i" % proc.returncode)

        if proc.returncode in (254, 255):
            self.debug_fail()
            self.bomb("testbed auxverb failed with exit code %i" % proc.returncode)

        return (proc.returncode, out, err)

    def check_exec(self, argv, stdout=False, kind="short", xenv=[]):
        """Run argv in testbed.

        If stdout is True, capture stdout and return it. Otherwise, don't
        redirect and return None.

        argv must succeed and not print any stderr.
        """
        (code, out, err) = self.execute(
            argv,
            stdout=(stdout and subprocess.PIPE or None),
            stderr=subprocess.PIPE,
            kind=kind,
            xenv=xenv,
        )
        if err:
            self.bomb(
                '"%s" failed with stderr "%s"' % (" ".join(argv), err),
                adtlog.AutopkgtestError,
            )
        if code != 0:
            self.bomb(
                '"%s" failed with status %i' % (" ".join(argv), code),
                adtlog.AutopkgtestError,
            )
        return out

    def is_real_package_installed(self, package):
        """Check if a non-virtual package is installed in the testbed"""
        (rc, out, err) = self.execute(
            ["dpkg-query", "--show", "-f", "${Status}", package],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if rc != 0:
            if "no packages found" in err:
                return False
            self.badpkg("Failed to run dpkg-query: %s (exit code %d)" % (err, rc))
        if out == "install ok installed":
            return True
        return False

    def is_virtual_package_installed(self, package):
        """Check if a package is installed in the testbed"""
        (rc, out, err) = self.execute(
            ["dpkg-query", "--show", "-f", "${Status} ${Provides}\n", "*"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if rc != 0:
            self.badpkg("Failed to run dpkg-query: %s (exit code %d)" % (err, rc))
            return False
        if not out:
            return False
        for line in out.splitlines():
            prefix = "install ok installed "
            if not line.startswith(prefix):
                continue
            line = line[len(prefix) :]
            for p in line.split(","):
                (p, _, _) = p.lstrip().partition(" ")  # ' foo (== 1.0)' => 'foo'
                if p == package:
                    return True
        return False

    def _run_apt_command(
        self,
        *,
        action: str,
        what: List[str] = [],
        extra_opts: List[str] = [],
        xenv: List[str] = [],
    ) -> None:
        """Call apt-get with an action (update, install, satisfy, ...) with
        options suitable for installing test dependencies."""

        env = xenv.copy()
        env.append("DEBIAN_FRONTEND=noninteractive")
        env.append("APT_LISTBUGS_FRONTEND=none")
        env.append("APT_LISTCHANGES_FRONTEND=none")

        cmd = []
        if self.eatmydata_prefix:
            cmd.append(self.eatmydata_prefix)
        cmd += [
            "apt-get",
            "--quiet",
            "--assume-yes",
            "-o=APT::Status-Fd=3",
            "-o=APT::Install-Recommends=false",
            f"-o=DPkg::Lock::Timeout={timeouts['copy']}",
            "-o=Dpkg::Options::=--force-confnew",
            "-o=Debug::pkgProblemResolver=true",
        ]
        cmd += extra_opts
        cmd.append(action)
        cmd += what

        # capture status-fd via stderr
        cmd = ["/bin/sh", "-ec", '"$@" 3>&2 2>&1', "run_apt_command"] + cmd

        (rc, _, serr) = self.execute(
            cmd,
            kind="install",
            stderr=subprocess.PIPE,
            xenv=env,
        )
        if rc != 0:
            adtlog.debug("apt-get %s failed; status-fd:\n%s" % (action, serr))
            # check if apt failed during package download, which might be a
            # transient error, so we likely want a "testbed failure".
            if "dlstatus:" in serr and "pmstatus:" not in serr:
                raise adtlog.AptDownloadError
            else:
                raise adtlog.AptPermanentError

    def _install_apt_package_under_test_deps(
        self, package_under_test_deps: List[List[str]]
    ) -> None:
        need_explicit_install = []
        for dep in package_under_test_deps:
            if len(dep) == 1:
                # simple test dependency (no alternatives)
                pkg = dep[0]
                if self.is_real_package_installed(pkg):
                    continue
                if self.is_virtual_package_installed(pkg):
                    need_explicit_install.append(pkg)
                    continue
                adtlog.warning("package %s is not installed though it should be" % pkg)
            else:
                # figure out which test dependency alternative got installed
                installed_virtual_packages = []
                found_a_real_package = False
                for pkg in dep:
                    if self.is_real_package_installed(pkg):
                        # no need to install anything from this set of alternatives
                        found_a_real_package = True
                        break
                    if self.is_virtual_package_installed(pkg):
                        installed_virtual_packages.append(pkg)
                if found_a_real_package:
                    continue
                if not installed_virtual_packages:
                    adtlog.warning(
                        "no alternative in %s is installed though one should be" % dep
                    )
                    continue
                if len(installed_virtual_packages) > 1:
                    adtlog.warning(
                        "more than one test dependency alternative in %s installed as a virtual package, installing the first one (%s) as the real package"
                        % (dep, installed_virtual_packages[0])
                    )
                need_explicit_install.append(installed_virtual_packages[0])

        if need_explicit_install:
            adtlog.debug(
                "installing real packages of test dependencies: %s"
                % need_explicit_install
            )
            self._run_apt_command(action="install", what=need_explicit_install)

    def _generate_satdep_deb(self, deps: str) -> "TestbedPath":
        """create a dummy deb with the deps"""
        pkgdir = tempfile.mkdtemp(prefix="autopkgtest-satdep.")
        debdir = os.path.join(pkgdir, "DEBIAN")
        os.chmod(pkgdir, 0o755)
        os.mkdir(debdir)
        os.chmod(debdir, 0o755)
        with open(os.path.join(debdir, "control"), "w") as f:
            f.write(
                textwrap.dedent(
                    f"""\
                    Package: autopkgtest-satdep
                    Section: oldlibs
                    Priority: extra
                    Maintainer: autogenerated
                    Version: 0
                    Architecture: {self.dpkg_arch}
                    Depends: {deps}
                    Description: satisfy autopkgtest test dependencies
                    """
                )
            )
        deb = TempPath(self, "autopkgtest-satdep.deb")
        subprocess.check_call(
            ["dpkg-deb", "-Zxz", "-b", pkgdir, deb.host], stdout=subprocess.PIPE
        )
        shutil.rmtree(pkgdir)
        return deb

    def install_apt(
        self,
        deps: str,
        shell_on_failure: bool = False,
        package_under_test_deps: Optional[List[List[str]]] = None,
    ) -> None:
        """Install dependencies with apt-get into testbed

        This requires root privileges and a writable file system.
        """
        # Check if apt supports the `apt-get satisfy <string>` syntax
        legacy_apt = self.apt_version < Version("1.9.0")

        if legacy_apt:
            # Fallback to installing test dependencies via a dummy binary .deb.
            #
            # TODO for when we'll be able to assume apt (>= 1.1exp1) on the testbed:
            # 1. Switch from installing deps via `dpkg --unpack autopkgtest-satdep.deb`
            #    + `apt-get --fix-broken` + `apt-get purge autokgtest-satdep` to
            #    `apt-get build-dep autopkgtest-satdep.dsc`.
            #    See: b86576aa47f6f95985c3be46e4a4e20278da4511.
            # 2. Drop the special handling of :native.
            adtlog.debug(
                "Legacy testbed apt, installing dependencies via autopkgtest-satdep.deb"
            )

            # We need to strip :native qualifiers as they are not valid in
            # DEBIAN/control Depends, see deb-control(5), and compare with
            # deb-src-control(5).
            deps = deps.replace(":native", "")

            assert self.dpkg_arch, "testbed architecture unknown"
            try:
                deps = subprocess.run(
                    [
                        os.path.join(PKGDATADIR, "lib", "parse-deps.pl"),
                        deps,
                        self.dpkg_arch,
                    ],
                    check=True,
                    capture_output=True,
                    text=True,
                ).stdout.strip()
            except subprocess.CalledProcessError as e:
                self.bomb(f"failed to run: {e.cmd}")
            adtlog.debug(f"architecture resolved deps: {deps}")

            # generate a dummy package with the deps
            satdep_package: TestbedPath = self._generate_satdep_deb(deps)
            satdep_package.copydown()

        # install the dependencies in the tb; our apt pinning is not very
        # clever wrt. resolving transitional dependencies in the pocket,
        # so we might need to retry without pinning
        while True:
            if legacy_apt:
                self.check_exec(
                    ["dpkg", "--unpack", satdep_package.tb], stdout=subprocess.PIPE
                )

            rc = 0
            try:
                if legacy_apt:
                    self._run_apt_command(action="install", extra_opts=["--fix-broken"])
                elif deps:
                    self._run_apt_command(action="satisfy", what=[deps])
                # For dependencies that are part of the source package under
                # test (for example if '@' was provided as a test dependency),
                # we explicitly install the corresponding binary packages in case
                # the dependencies on those were satisfied by versioned Provides
                if package_under_test_deps:
                    self._install_apt_package_under_test_deps(package_under_test_deps)

            except adtlog.AptDownloadError:
                self.bomb("apt failed to download packages")

            except adtlog.AptPermanentError:
                rc = -1
                if shell_on_failure:
                    self.run_shell()
            else:
                if legacy_apt:
                    # verify that we actually got autopkgtest-satdep installed
                    rc = self.execute(
                        ["dpkg", "--status", "autopkgtest-satdep"],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                    )[0]

            if rc != 0:
                if self.apt_pin_for_releases and self.enable_apt_fallback:
                    release = self.apt_pin_for_releases.pop()
                    adtlog.warning(
                        "Test dependencies are unsatisfiable with using apt pinning. "
                        "Retrying with using all packages from %s" % release
                    )
                    pref_file_basename = f"autopkgtest-{release}.pref"
                    # same replacement we do in _create_apt_pinning_for_packages()
                    pref_file_basename = self._sanitize_filename_for_apt(
                        pref_file_basename
                    )
                    pref_file = f"/etc/apt/preferences.d/{pref_file_basename}"
                    self.check_exec(["rm", pref_file])
                    continue

                if shell_on_failure:
                    self.run_shell()
                if self.enable_apt_fallback:
                    self.badpkg(
                        "Test dependencies are unsatisfiable. A common reason is "
                        "that your testbed is out of date with respect to the "
                        "archive, and you need to use a current testbed or run "
                        "apt-get update or use -U."
                    )
                else:
                    self.badpkg(
                        "Test dependencies are unsatisfiable. A common reason is "
                        "that the requested apt pinning prevented dependencies "
                        "from the non-default suite to be installed. In that "
                        "case you need to add those dependencies to the pinning "
                        "list."
                    )
            break

        if legacy_apt:
            # remove autopkgtest-satdep to avoid confusing tests, but avoid marking our
            # test dependencies for auto-removal
            out = self.check_exec(
                [
                    "apt-get",
                    "--simulate",
                    "--quiet",
                    "-o",
                    "APT::Get::Show-User-Simulation-Note=False",
                    "--auto-remove",
                    "purge",
                    "autopkgtest-satdep",
                ],
                True,
            )
            test_deps = []
            for line in out.splitlines():
                if not line.startswith("Purg "):
                    continue
                pkg = line.split()[1]
                if pkg != "autopkgtest-satdep":
                    test_deps.append(pkg)
            if test_deps:
                adtlog.debug(
                    "Marking test dependencies as manually installed: %s"
                    % " ".join(test_deps)
                )
                # avoid overly long command lines
                batch = 0
                while batch < len(test_deps):
                    self.check_exec(
                        ["apt-mark", "manual", "-qq"] + test_deps[batch : batch + 20]
                    )
                    batch += 20

            self.execute(["dpkg", "--purge", "autopkgtest-satdep"])

    def install_build_deps_for_package(
        self, src_package_path, shell_on_failure=False, xenv=[]
    ) -> None:
        """Install build dependencies via apt-get build-dep.

        This requires apt >= 1.1~exp1 (Debian >= 9, Ubuntu >= 16.04).
        The caller function is expected to check this.
        """
        # apt-get build-dep --simulate prints a "Inst ..." line for each package to be installed.
        # If none of those is present, it means no package actually needs to be installed.
        # We do this because with --simulate we can run apt-get as non-root; without --simulate
        # apt-get build-dep always requires root, even if no package is installed.
        apt_bd_simulate_cmd = [
            "apt-get",
            "-q",
            "--simulate",
            "build-dep",
            src_package_path,
        ]
        rc, out, _ = self.execute(
            apt_bd_simulate_cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            xenv=xenv,
        )
        bd_simulate_out = out.splitlines()

        if rc != 0:
            adtlog.error(
                f"Failed to resolve build-deps for {src_package_path}; apt-get --simulate output:"
            )
            for line in bd_simulate_out:
                adtlog.error("| " + line)
            if shell_on_failure:
                self.run_shell()
            # If we fail here we can blame the package => error out via badpkg().
            self.badpkg("Can't resolve build dependencies on testbed")

        needed_bd = [line for line in bd_simulate_out if re.match(r"^Inst ", line)]

        if not needed_bd:
            adtlog.debug("No build dependencies to install")
            return

        if "root-on-testbed" not in self.caps:
            adtlog.error(
                "Missing build dependencies, apt-get build-dep --simulate output:"
            )
            for line in bd_simulate_out:
                adtlog.error("| " + line)
            if shell_on_failure:
                self.run_shell()
            adtlog.bomb("Can't install build dependencies (must be root).")

        try:
            self._run_apt_command(
                action="build-dep",
                what=[src_package_path],
                xenv=xenv,
            )
        except adtlog.AptError:
            if shell_on_failure:
                self.run_shell()
            self.bomb("Failure installing build dependencies on testbed")

    def satisfy_dependencies_string(
        self,
        deps: str,
        what: str,
        shell_on_failure: bool = False,
        package_under_test_deps: Optional[List[List[str]]] = None,
    ) -> None:
        """Install dependencies from a string into the testbed"""

        adtlog.debug("%s: satisfying %s" % (what, deps))

        # check if we can use apt-get
        can_apt_get = False
        if "root-on-testbed" in self.caps:
            can_apt_get = True
        adtlog.debug("can use apt-get on testbed: %s" % can_apt_get)

        if can_apt_get:
            self.install_apt(deps, shell_on_failure, package_under_test_deps)
        else:
            # This whole `else` branch is meant to check if we can get
            # away without being root on the testbed because all the
            # dependencies are already installed. Hopefully one day
            # we'll replace it with something better, e.g. checking
            # the output of `apt-get --simulate`.

            has_dpkg_checkbuilddeps = (
                self.execute(
                    ["sh", "-ec", "command -v dpkg-checkbuilddeps"],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                )[0]
                == 0
            )
            if has_dpkg_checkbuilddeps:
                rc, _, err = self.execute(
                    ["dpkg-checkbuilddeps", "-d", deps, "/dev/null"],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                )
                if rc != 0:
                    missing = re.sub(
                        "dpkg-checkbuilddeps: error: Unmet build dependencies: ",
                        "",
                        err,
                    )
                    self.badpkg("test dependencies missing: %s" % missing)
            else:
                adtlog.warning(
                    "test dependencies (%s) are not fully satisfied, but continuing anyway since dpkg-checkbuilddeps it not available to determine which ones are missing."
                    % deps
                )

    def _add_apt_preference(self, preference: str, filename: str) -> None:
        """Add APT preference under /etc/apt/preferences.d"""

        preference = preference.strip()
        if not preference:
            return
        filename = self._sanitize_filename_for_apt(filename)
        adtlog.debug(f"adding APT preference to {filename}:\n{preference}")
        self.check_exec(
            [
                "sh",
                "-ec",
                f'"$@" > "/etc/apt/preferences.d/{filename}"',
                "add_apt_preference",
                "printf",
                r"%s\n",
                preference,
            ]
        )

    def run_shell(self, cwd=None, extra_env=[]):
        """Run shell in testbed for debugging tests"""

        adtlog.info(" - - - - - - - - - - running shell - - - - - - - - - -")
        self.command("shell", [cwd or "/"] + extra_env)

    def run_test(
        self,
        tree,
        test,
        extra_env=[],
        shell_on_failure=False,
        shell=False,
        build_parallel=None,
    ):
        """Run given test in testbed

        tree (a TestbedPath) is the source tree root.
        """

        def _info(m):
            adtlog.info("test %s: %s" % (test.name, m))

        self.last_test_name = test.name

        if self.skip_remaining_tests:
            test.set_skipped("global timeout exceeded")
            return

        if test.path and not os.path.exists(os.path.join(tree.host, test.path)):
            self.badpkg("%s does not exist" % test.path)

        # record installed package versions
        if self.output_dir:
            pkglist = TempPath(self, test.name + "-packages.all", autoclean=False)
            self.check_exec(
                [
                    "sh",
                    "-ec",
                    "dpkg-query --show -f '${Package}\\t${Version}\\n' > %s"
                    % pkglist.tb,
                ]
            )
            pkglist.copyup()

            # filter out packages from the base system
            with open(pkglist.host[:-4], "w") as out:
                join = subprocess.Popen(
                    [
                        "join",
                        "-v2",
                        "-t\t",
                        os.path.join(self.output_dir, "testbed-packages"),
                        pkglist.host,
                    ],
                    stdout=out,
                    env={},
                )
                join.communicate()
                if join.returncode != 0:
                    self.badpkg(
                        "failed to call join for test specific package list, code %d"
                        % join.returncode
                    )
            os.unlink(pkglist.host)

        # ensure our tests are in the testbed
        tree.copydown(check_existing=True)

        # stdout/err files in testbed
        so = TempPath(self, test.name + "-stdout", autoclean=False)
        se = TempPath(self, test.name + "-stderr", autoclean=False)

        # create script to run test
        test_artifacts = "%s/%s-artifacts" % (self.scratch, test.name)
        autopkgtest_tmp = "%s/autopkgtest_tmp" % (self.scratch)
        assert self.nproc is not None

        argv = [self.wrapper_sh]

        if adtlog.verbosity >= 2:
            # This should come first so that it affects all CLI options
            argv.append("--debug")

        if any(
            r in ["isolation-machine", "isolation-container"] for r in test.restrictions
        ):
            # Must come early, because it influences behaviour of other options eg --artifacts
            argv.append("--isolation")

        argv.extend(
            [
                "--artifacts={}".format(test_artifacts),
                "--chdir={}".format(tree.tb),
                "--env=AUTOPKGTEST_TESTBED_ARCH={}".format(self.dpkg_arch),
                "--env=AUTOPKGTEST_TEST_ARCH={}".format(self.test_arch),
                "--env=DEB_BUILD_OPTIONS=parallel={}".format(
                    build_parallel or self.nproc
                ),
                "--env=DEBIAN_FRONTEND=noninteractive",
                "--env=LANG=C.UTF-8",
                "--unset-env=LANGUAGE",
                "--unset-env=LC_ADDRESS",
                "--unset-env=LC_ALL",
                "--unset-env=LC_COLLATE",
                "--unset-env=LC_CTYPE",
                "--unset-env=LC_IDENTIFICATION",
                "--unset-env=LC_MEASUREMENT",
                "--unset-env=LC_MESSAGES",
                "--unset-env=LC_MONETARY",
                "--unset-env=LC_NAME",
                "--unset-env=LC_NUMERIC",
                "--unset-env=LC_PAPER",
                "--unset-env=LC_TELEPHONE",
                "--unset-env=LC_TIME",
                "--script-pid-file=/tmp/autopkgtest_script_pid",
                "--source-profile",
                "--stderr={}".format(se.tb),
                "--stdout={}".format(so.tb),
                "--tmp={}".format(autopkgtest_tmp),
            ]
        )

        if self.test_arch_is_foreign:
            rc, dpkg_arch_vars, _ = self.execute(
                ["dpkg-architecture", "-a" + self.test_arch],
                stdout=subprocess.PIPE,
                stderr=subprocess.DEVNULL,
            )
            testbed_has_dpkg_architecture = rc == 0
            if testbed_has_dpkg_architecture:
                for var in dpkg_arch_vars.splitlines():
                    # set up environment for cross-architecture building.
                    if var.startswith("DEB_HOST_") or var.startswith("DEB_BUILD_"):
                        argv.append(f"--env={var}")
                adtlog.info("test environment configured for cross building")

        if "needs-root" in test.restrictions and self.user is not None:
            argv.append("--env=AUTOPKGTEST_NORMAL_USER={}".format(self.user))
            argv.append("--env=ADT_NORMAL_USER={}".format(self.user))

        for e in extra_env:
            argv.append("--env={}".format(e))

        if test.path:
            test_cmd = os.path.join(tree.tb, test.path)
            argv.extend(
                [
                    "--make-executable=" + test_cmd,
                    "--",
                    test_cmd,
                ]
            )
        else:
            argv.extend(["--", "bash", "-ec", test.command])

        # Convert the argv into a shell one-line so we can wrap it in su
        script = "set -e; exec " + " ".join(map(shlex.quote, argv))

        if "needs-root" not in test.restrictions and self.user is not None:
            if "root-on-testbed" not in self.caps:
                self.bomb(
                    "cannot change to user %s without root-on-testbed" % self.user,
                    adtlog.AutopkgtestError,
                )
            # we don't want -l here which resets the environment from
            # self.execute(); so emulate the parts that we want
            # FIXME: move "run as user" as an argument of execute()/check_exec() and run with -l
            test_argv = ["su", "-s", "/bin/bash", self.su_user, "-c"]

            if "rw-build-tree" in test.restrictions:
                self.check_exec(["chown", "-R", self.user, tree.tb])
        else:
            # this ensures that we have a PAM/logind session for root tests as
            # well; with some interfaces like ttyS1 or lxc_attach we don't log
            # in to the testbed
            if "root-on-testbed" in self.caps:
                test_argv = ["su", "-s", "/bin/bash", "root", "-c"]
            else:
                test_argv = ["bash", "-c"]

        # run test script
        if test.command:
            _info(test.command)
        _info("[-----------------------")

        # tests may reboot, so we might need to run several times
        self.last_reboot_marker = ""
        timeout = False
        while True:
            if self.last_reboot_marker:
                script_prefix = (
                    'export AUTOPKGTEST_REBOOT_MARK="%s"; export ADT_REBOOT_MARK="$AUTOPKGTEST_REBOOT_MARK"; '
                    % self.last_reboot_marker
                )
            else:
                script_prefix = ""
            try:
                rc = self.execute(test_argv + [script_prefix + script], kind="test")[0]
            except VirtSubproc.Timeout as exc:
                rc = 1
                timeout = True
                self.skip_remaining_tests = exc.abort
                break

            # If the shell or su command that runs wrapper.sh exits with
            # status 128+SIGKILL, that probably means either wrapper.sh or
            # the test itself was terminated with SIGKILL, which might be
            # because the test asked for the testbed to be rebooted.
            if rc in (-signal.SIGKILL, 128 + signal.SIGKILL) and "reboot" in self.caps:
                adtlog.debug("test process SIGKILLed, checking for reboot marker")
                (code, reboot_marker, err) = self.execute(
                    ["cat", "/run/autopkgtest-reboot-mark"],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                )
                if code == 0:
                    self.last_reboot_marker = reboot_marker.strip()
                    adtlog.info(
                        "test process requested reboot with marker %s"
                        % self.last_reboot_marker
                    )
                    self.reboot()
                    continue

                adtlog.debug(
                    "test process SIGKILLed, checking for prepare-reboot marker"
                )
                (code, reboot_marker, err) = self.execute(
                    ["cat", "/run/autopkgtest-reboot-prepare-mark"],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                )
                if code == 0:
                    self.last_reboot_marker = reboot_marker.strip()
                    adtlog.info(
                        "test process requested preparation for reboot with marker %s"
                        % self.last_reboot_marker
                    )
                    self.reboot(prepare_only=True)
                    continue

                adtlog.debug("no reboot marker, considering a failure")
            break

        # give the setup_trace() cats some time to catch up
        sys.stdout.flush()
        sys.stderr.flush()
        time.sleep(0.3)
        _info("-----------------------]")
        adtlog.debug("testbed executing test finished with exit status %i" % rc)

        # copy stdout/err files to host
        try:
            so.copyup()
            se.copyup()
            se_size = os.path.getsize(se.host)
        except adtlog.TestbedFailure:
            if timeout:
                # if the test timed out, it's likely that the test destroyed
                # the testbed, so ignore this and call it a failure
                adtlog.warning("Copying up test output timed out, ignoring")
                se_size = 0
                so.host = None
                se.host = None
            else:
                # smells like a tmpfail
                raise

        # avoid mixing up stdout (from report) and stderr (from logging) in output
        sys.stdout.flush()
        sys.stderr.flush()
        time.sleep(0.1)

        _info(" - - - - - - - - - - results - - - - - - - - - -")

        if timeout:
            test.failed("timed out")
        elif rc == 77 and "skippable" in test.restrictions:
            test.set_skipped("exit status 77 and marked as skippable")
        elif rc != 0:
            if "needs-internet" in test.restrictions and self.needs_internet == "try":
                test.set_skipped(
                    "Failed, but test has needs-internet and that's not guaranteed"
                )
            else:
                test.failed("non-zero exit status %d" % rc)
        elif se_size != 0 and "allow-stderr" not in test.restrictions:
            with open(se.host, encoding="UTF-8", errors="replace") as f:
                stderr_top = f.readline().rstrip("\n \t\r")
            test.failed("stderr: %s" % stderr_top)
        else:
            test.passed()

        sys.stdout.flush()
        sys.stderr.flush()

        # skip the remaining processing if the testbed got broken
        if se.host is None:
            adtlog.debug(
                "Skipping remaining log processing and testbed restore after timeout"
            )
            return

        if os.path.getsize(so.host) == 0:
            # don't produce empty -stdout files in --output-dir
            so.autoclean = True

        if se_size != 0 and "allow-stderr" not in test.restrictions:
            # give tee processes some time to catch up, to avoid mis-ordered logs
            time.sleep(0.2)
            _info(" - - - - - - - - - - stderr - - - - - - - - - -")
            with open(se.host, "rb") as f:
                while True:
                    block = f.read1(1000000)
                    if not block:
                        break
                    sys.stderr.buffer.write(block)
            sys.stderr.buffer.flush()
        else:
            # don't produce empty -stderr files in --output-dir
            if se_size == 0:
                se.autoclean = True

        # copy artifacts to host, if we have --output-dir
        if self.output_dir:
            ap = TestbedPath(
                self,
                os.path.join(self.output_dir, "artifacts", test.name),
                test_artifacts,
                is_dir=True,
            )
            ap.copyup()
            # don't keep an empty artifacts dir around
            if not os.listdir(ap.host):
                os.rmdir(ap.host)
            if not os.listdir(os.path.dirname(ap.host)):
                os.rmdir(os.path.dirname(ap.host))

        if shell or (shell_on_failure and not test.result):
            self.run_shell(
                tree.tb,
                [
                    'AUTOPKGTEST_ARTIFACTS="%s"' % test_artifacts,
                    'AUTOPKGTEST_TMP="%s"' % autopkgtest_tmp,
                ],
            )

        # clean up artifacts and AUTOPKGTEST_TMP dirs
        self.check_exec(["rm", "-rf", test_artifacts, autopkgtest_tmp])

    #
    # helper methods
    #

    def _set_baseline_pin_priority_for_release(self, release):
        """Give Pin-Priority: 500 to any package that did not match an earlier
        preference. This prevents "NotAutomatic: yes", "ButAutomaticUpgrades:
        yes" to have any effect on the default priority of release.
        See APT's Default Priority Assignments in apt_preferences(5)."""
        # This pref file should sort last, hence the "zz" in the file name.
        pref_file = f"autopkgtest-zz-{release}-baseline.pref"
        preference = textwrap.dedent(
            f"""\
            Package: *
            Pin: release {release}
            Pin-Priority: 500
            """
        )
        self._add_apt_preference(preference, pref_file)

    def _create_apt_pinning_for_packages(self, release, pkglist):
        """Create apt pinning for --apt-pocket=pocket=pkglist and
        --pin-packages=release=pkglist"""

        pref_template = Template(
            textwrap.dedent(
                """\
                Package: $pkgs
                Pin: release $release
                Pin-Priority: $prio

                """
            )
        )

        pkglist = pkglist.split(",")

        if self.apt_version < Version("2.0"):
            # sort pkglist into source and binary packages
            binpkgs = []
            srcpkgs = []
            for i in pkglist:
                i = i.strip()
                if i.startswith("src:"):
                    srcpkgs.append(i[4:])
                else:
                    binpkgs.append(i)

            # translate src:name entries into binaries of that source
            #
            # apt 1.9.11 implemented support for pinning by source package
            # using the src:pkg syntax, as documented in apt_preferences(5),
            # however that's too new for us to use for now.
            if srcpkgs:
                apt_cache = self.check_exec(["apt-cache", "dumpavail"], True)
                for paragr in Deb822.iter_paragraphs(apt_cache):
                    pkg = paragr.get("Package")
                    src = paragr.get("Source", pkg).split()[0]
                    if src in srcpkgs and pkg not in binpkgs:
                        binpkgs.append(pkg)

            pkglist = binpkgs

        # We want pins to be valid on any architecture, so that they work
        # consistently across architectures in multi-arch systems.
        for i, pkg in enumerate(pkglist):
            if ":" not in (pkg[4:] if pkg.startswith("src:") else pkg):
                pkglist[i] += ":any"

        # Pin the release with priority 100, which is the same as if it had
        # "NotAutomatic: yes", "ButAutomaticUpgrades: yes", which is what we
        # want as only pinned packages should be taken from it.
        pref_content = pref_template.substitute(
            pkgs="*",
            release=release,
            prio="100",
        )

        # prefer given packages from the specified release
        if pkglist:
            pref_content += pref_template.substitute(
                pkgs=" ".join(pkglist),
                release=release,
                prio="995",
            )

        pref_file = f"autopkgtest-{release}.pref"
        self._add_apt_preference(pref_content, pref_file)

        self.apt_pin_for_releases.append(release)

    def _set_default_release(self):
        """Set apt's default release."""
        # The "default release" preference file should sort first so to behave
        # like APT::Default-Release. See APT's Default Priority Assignments in
        # apt_preferences(5), "Note that this has precedence...".
        pref_file = "autopkgtest-00-default-release.pref"
        preference = textwrap.dedent(
            f"""\
            Package: *
            Pin: release {self.default_release}
            Pin-Priority: 990
            """
        )
        self._add_apt_preference(preference, pref_file)

    def _add_apt_source(self, source: str, filename: str) -> None:
        source = source.strip()
        if not source:
            return
        adtlog.debug("adding APT source: %s" % source)
        filename = self._sanitize_filename_for_apt(filename)
        if filename.endswith(".sources"):
            # add empty line to ensure deb822 stanzas are correctly separated
            source += "\n"
        new_source_path = f"/etc/apt/sources.list.d/{filename}"
        self.check_exec(
            [
                "sh",
                "-ec",
                f'"$@" >> "{new_source_path}"',
                "add_apt_source",
                "printf",
                r"%s\n",
                source,
            ]
        )

        adtlog.info(f"updating testbed package index for new source {filename}")
        try:
            self._run_apt_command(
                action="update",
                # Options to only update the added source, without discarding existing data.
                extra_opts=[
                    f"-o=Dir::Etc::SourceList={new_source_path}",
                    "-o=Dir::Etc::SourceParts=/dev/null",
                    "-oAPT::Get::List-Cleanup=0",
                ],
            )
        except adtlog.AptError:
            self.bomb(f"failed testbed apt update for {filename}")

    def _get_apt_sources_one_line_style(self) -> List[str]:
        """Return list of all the configured one-line-style sources."""

        script = load_shell_script("setup-commands/get-apt-sources-one-line-style")
        sources = self.check_exec(["sh", "-ec", script], stdout=True)
        sources = sources.splitlines()
        return sources

    def _get_apt_sources_deb822(self) -> List[Deb822]:
        """Return a list of all the deb822 configured sources."""

        script = load_shell_script("setup-commands/get-apt-sources-deb822")
        sources = self.check_exec(["sh", "-ec", script], stdout=True)
        sources = Deb822.iter_paragraphs(sources)
        sources = list(sources)
        return sources

    def _add_apt_release_one_line_style(self, release: str) -> None:
        """
        For each existing "deb" one-line-style source entry where "suite" is
        the testbed's base release, make new deb and deb-src source entries
        based on the existing one, but with the suite now being <release>.

        For example, if an existing source entry looks like:

        deb [ trusted=yes ] http://archive.ubuntu.com/ubuntu/ mantic main universe

        and we are adding the release 'noble', the resulting source entries
        would be:

        deb [ trusted=yes ] http://archive.ubuntu.com/ubuntu/ noble main universe
        deb-src [ trusted=yes ] http://archive.ubuntu.com/ubuntu/ noble main universe

        The new sources are written to a new file,
        /etc/apt/sources.list.d/autopkgtest-add-apt-release-<release>.list.
        """
        sources = self._get_apt_sources_one_line_style()
        if not sources:
            return

        sources_for_release = []
        for line in sources:
            match = self.SOURCELIST_PATTERN.match(line)
            if not match:
                adtlog.info(f"Failed to parse APT list entry: {line}")
                continue
            source = match.groupdict()
            if source["type"] != "deb" or source["suite"] != self.release:
                continue

            new_source = ["deb"]
            if source["options"]:
                new_source.append(source["options"])
            new_source.append(source["uri"])
            new_source.append(release)
            new_source.append(source["components"])
            sources_for_release.append(" ".join(new_source))
            new_source[0] = "deb-src"
            sources_for_release.append(" ".join(new_source))

        self._add_apt_source(
            "\n".join(sources_for_release).strip(),
            filename=f"autopkgtest-add-apt-release-{release}.list",
        )

    def _add_apt_release_deb822(self, release: str) -> None:
        """
        For each existing deb822 source entry with "Types: deb" and Suites
        including the testbed's base release, make a new source entry that is
        a copy of the existing one, but replace the Suites with just <release>,
        and enable both deb and deb-src.

        For example, if an existing source entry looks like:

        Types: deb
        URIs: http://archive.ubuntu.com/ubuntu/
        Suites: mantic mantic-updates
        Components: main universe

        and we are adding the release 'noble', the resulting source entry
        would be:

        Types: deb deb-src
        URIs: http://archive.ubuntu.com/ubuntu/
        Suites: noble
        Components: main universe

        The new sources are written to a new file,
        /etc/apt/sources.list.d/autopkgtest-add-apt-release-<release>.sources.
        """
        sources = self._get_apt_sources_deb822()
        if not sources:
            return

        sources_for_release = []
        for source in sources:
            if "deb" not in source["Types"].split():
                continue
            if self.release not in source["Suites"].split():
                continue
            new_source = source.copy()
            new_source["Suites"] = release
            new_source["Types"] = "deb deb-src"
            sources_for_release.append(new_source)

        # Convert Deb822 objects into strings
        sources_for_release = [str(source) for source in sources_for_release]
        self._add_apt_source(
            "\n".join(sources_for_release),
            filename=f"autopkgtest-add-apt-release-{release}.sources",
        )

    def _add_apt_ppa(self, ppaspec: str) -> None:
        """Add PPA to the APT sources.list"""

        # parse the ppa spec
        ppaspec = ppaspec.strip()
        match = self.PPA_PATTERN.match(ppaspec)
        if not match:
            self.bomb(f"invalid PPA name: {ppaspec}", adtlog.AutopkgtestError)
            return
        adtlog.debug(f"adding PPA: {ppaspec}")
        ppa = match.groupdict()
        ppa_shorthand = f"{ppa['name']}/{ppa['owner']}"
        ppa_is_private = bool(ppa["user"])

        # with private PPAs we can't query Launchpad for the fingerprint
        if ppa_is_private and not ppa["fingerprint"]:
            self.bomb(
                f"fingerprint not specified for private PPA {ppa_shorthand}",
                adtlog.AutopkgtestError,
            )

        # construct PPA repository URI
        if ppa_is_private:
            lpcontent_endpoint = "private-ppa.launchpadcontent.net"
        else:
            lpcontent_endpoint = "ppa.launchpadcontent.net"
        ppa_repo_uri = (
            f"https://{lpcontent_endpoint}/{ppa['owner']}/{ppa['name']}/ubuntu/"
        )

        # create opener for PPA repository URIs
        if ppa_is_private:
            ppa_password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
            ppa_password_mgr.add_password(None, ppa_repo_uri, ppa["user"], ppa["token"])
            ppa_auth_handler = urllib.request.HTTPBasicAuthHandler(ppa_password_mgr)
            ppa_urlopener = urllib.request.build_opener(ppa_auth_handler)
        else:
            ppa_urlopener = urllib.request.build_opener()

        # check that ppa exists and has contents for release; we actively probe
        # the PPA to fail early with an AutopkgtestError if case of issues, and
        # not later e.g. when updating the index, possibly causing a misleading
        # testbed failure.
        adtlog.debug(f"checking that {ppa_shorthand} has contents for {self.release}")
        http_timeout = 20
        try:
            url = urllib.parse.urljoin(ppa_repo_uri, f"dists/{self.release}/Release")
            adtlog.debug(f"checking that {ppa_shorthand} Release file exists: {url}")
            req = urllib.request.Request(url, method="HEAD")
            # we are only interested in the http status code, so we open().close()
            ppa_urlopener.open(req, timeout=http_timeout).close()

            url = urllib.parse.urljoin(
                ppa_repo_uri, f"dists/{self.release}/main/source/Release"
            )
            adtlog.debug(
                f"checking that {ppa_shorthand} source/Release file exists: {url}"
            )
            req = urllib.request.Request(url, method="HEAD")
            ppa_urlopener.open(url, timeout=http_timeout).close()
        except (urllib.error.URLError, urllib.error.HTTPError) as e:
            self.bomb(
                f"no packages found for {self.release} in PPA {ppa_shorthand}: {e}",
                adtlog.AutopkgtestError,
            )

        # get ppa signing key fingerprint
        if not ppa["fingerprint"]:
            url = f"https://api.launchpad.net/1.0/~{ppa['owner']}/+archive/ubuntu/{ppa['name']}"
            adtlog.debug(f"querying Launchpad about ppa {ppa_shorthand}: {url}")
            try:
                with urllib.request.urlopen(url, timeout=http_timeout) as response:
                    content = response.read().decode("utf-8")
            except (urllib.error.URLError, urllib.error.HTTPError) as e:
                self.bomb(
                    f"failed to get signing key fingerprint for {ppa_shorthand}: {e}",
                    adtlog.AutopkgtestError,
                )
            try:
                ppadata = json.loads(content)
            except json.JSONDecodeError as e:
                self.bomb(
                    f"bad json data for PPA {ppa_shorthand} from {url}: {e}",
                    adtlog.AutopkgtestError,
                )
            ppa["fingerprint"] = ppadata["signing_key_fingerprint"]
            adtlog.debug(
                f"PPA {ppa_shorthand} signing key fingerprint: {ppa['fingerprint']}"
            )

        # get ppa signing pubkey
        url = f"https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x{ppa['fingerprint']}"
        adtlog.debug(f"retrieving pubkey for ppa {ppa_shorthand}: {url}")
        try:
            with urllib.request.urlopen(url, timeout=http_timeout) as response:
                signed_by = response.read().decode("utf-8")
        except (urllib.error.URLError, urllib.error.HTTPError) as e:
            self.bomb(
                f"failed to get signing key for {ppa_shorthand}: {e}",
                adtlog.AutopkgtestError,
            )

        # if private PPA, embed user:token in uri for inclusion in source list
        if ppa_is_private:
            parsed_uri = urllib.parse.urlparse(ppa_repo_uri)
            netloc_with_userinfo = f"{ppa['user']}:{ppa['token']}@{parsed_uri.netloc}"
            parsed_uri = parsed_uri._replace(netloc=netloc_with_userinfo)
            ppa_repo_uri = urllib.parse.urlunparse(parsed_uri)

        # construct source list for ppa
        if self.apt_version < Version("2.3.10"):
            sourcelist = f"autopkgtest-ppa-{ppa['owner']}-{ppa['name']}.list"

            # add key to APT keyring
            ppa_keyring_basename = f"autopkgtest-ppa-{ppa['owner']}-{ppa['name']}.gpg"
            ppa_keyring_basename = self._sanitize_filename_for_apt(ppa_keyring_basename)
            ppa_keyring = f"/etc/apt/trusted.gpg.d/{ppa_keyring_basename}"
            self.check_exec(
                [
                    "sh",
                    "-ec",
                    'touch "$2";printf %s "$1" | apt-key --keyring "$2" add -;',
                    "add_apt_ppa",
                    signed_by,
                    ppa_keyring,
                ],
                xenv=["APT_KEY_DONT_WARN_ON_DANGEROUS_USAGE=1"],
            )

            # construct one-line-style source list
            ppa_source = textwrap.dedent(
                f"""\
                deb [ signed-by={ppa_keyring} ] {ppa_repo_uri} {self.release} main
                deb-src [ signed-by={ppa_keyring} ] {ppa_repo_uri} {self.release} main
                """
            )
        else:
            sourcelist = f"autopkgtest-ppa-{ppa['owner']}-{ppa['name']}.sources"

            # make pubkey suitable for inclusion in a deb822(5) multiline field
            signed_by = re.sub(r"^\s*$(?=\n)", ".", signed_by, flags=re.MULTILINE)
            signed_by = re.sub(r"^(?=.)", " ", signed_by, flags=re.MULTILINE)

            # construct deb822 source list
            ppa_source = textwrap.dedent(
                f"""\
                Types: deb deb-src
                URIs: {ppa_repo_uri}
                Suites: {self.release}
                Components: main
                Signed-By:
                """
            )
            ppa_source += signed_by

        # add apt source
        self._add_apt_source(ppa_source, filename=sourcelist)

    def _sanitize_filename_for_apt(self, filename: str) -> str:
        """APT only allows a-zA-Z0-9_.- in file names of preferences and
        data sources lists, see apt_preferences(5) and sources.list(5)."""

        bad_chars = re.escape(r'~+=,()[]{}!@#$%&\'":;|\/')
        return re.sub(f"[{bad_chars}]", "_", filename)

    def get_policy_for_package(self, package: str) -> Optional[str]:
        """Return apt policy for package"""

        (rc, stdout, _) = self.execute(
            [
                "apt-cache",
                "policy",
                # . and + are valid package name characters but are
                # interpreted by $(apt-cache policy) as regex
                "^" + package.replace(".", "\\.").replace("+", "\\+") + "$",
            ],
            stdout=subprocess.PIPE,
        )

        if rc != 0 or not stdout:
            return None

        return stdout

    def get_candidate(
        self, package: str
    ) -> Tuple[Optional[Version], Optional[int], Optional[str]]:
        """Return candidate version and priority for package"""

        policy = self.get_policy_for_package(package)
        if not policy:
            return None, None, None

        # Get the candidate version of this binary
        version = None
        for line in policy.splitlines():
            if not line.strip().startswith("Candidate:"):
                continue
            if "none" in line.lower():
                break
            version = line.strip().replace("Candidate: ", "")
            break

        if not version:
            return None, None, None

        priority, suite = self._version_info_from_policy(policy, version)

        return Version(version), priority, suite

    def _version_info_from_policy(
        self, policy: str, version: str
    ) -> Tuple[int, Optional[str]]:
        """Extract the priority for a package version from its policy.

        :param policy: `apt-cache policy` output for a single package
        :type policy: str
        :param version: the version for which to extract the prioriy
        :type version: str
        :return: priority and suite for the given version
        :rtype: tuple[int, str]
        """

        # The apt-policy output for a package looks like this:
        #
        # apt:
        #   Installed: 2.7.14build2
        #   Candidate: 2.7.14build2
        #   Version table:
        #      2.9.2 100
        #         100 http://archive.ubuntu.com/ubuntu oracular-proposed/main amd64 Packages
        #      2.9.1 50
        #         100 http://archive.ubuntu.com/ubuntu oracular-baz/main amd64 Packages
        #  *** 2.7.14build2 500 (phased 20%)
        #         500 http://archive.ubuntu.com/ubuntu oracular/main amd64 Packages
        #         100 /path/to/dpkg/status
        #      0.0.1 -30000
        #        -30000 http://archive.ubuntu.com/ubuntu oracular-foo/main amd64 Packages
        #      0.0.2 1
        #           1 http://archive.ubuntu.com/ubuntu oracular-bar/main amd64 Packages
        #
        # We'll proceed in stages:
        # 1. extract the "Version table"
        # 2. from that, extract the entry for the desired version
        # 3. from that, extract the version priority from the "version line" (first line)
        # If priority != 0 (always true with apt >= 1.1~exp9) we are done, otherwise:
        # 4. extract the max priority among "file" priorities.
        #
        # Notes:
        # - The '***' indicates the currently installed version, and may or may
        #   not be present.
        # - For apt (<< 1.1~exp9) compatibility, we can't just extract the priority
        #   from the line containing the package version, as apt (<< 1.1~exp9) just
        #   prints 0 there if there is no "version pin" pin set.

        # 1. Extract the version table. Note that apt prints "file priorities"
        # with this format string: "       %4i %s\n", see private-show.cc.
        needle = re.search(
            r"(?<=^  Version table:\n)(^( {5}| \*{3} | {7,})[^ ].*\n)+",
            policy,
            flags=re.M,
        )
        assert needle is not None
        version_table = needle.group(0)

        # 2. Extract the version table entry for the specified version. Refer to
        # the apt format string mentioned above to see why we look for "^ {7,}".
        needle = re.search(
            r"^( {5}| \*{3} )" + re.escape(version) + " .*\n(^ {7,}[^ ].*\n)+",
            version_table,
            flags=re.M,
        )
        assert needle is not None, (
            "version_table doesn't contain version '%s':\n%s\npolicy:\n%s"
            % (version, version_table, policy)
        )
        version_table_for_candidate = needle.group(0).splitlines()

        # 3. Extract the "version line" and from that the priority.
        version_line = version_table_for_candidate[0]
        priority = int(version_line[5:].split()[1])

        # This will always be false with apt >= 1.1~exp9.
        if priority == 0:
            # 4. Fallback to parsing "file priorities".
            file_priorities = version_table_for_candidate[1:]
            priorities = [int(fp.split(maxsplit=1)[0]) for fp in file_priorities]
            priority = max(priorities)

        assert priority != 0, "invalid priority, bug in policy parsing?"

        # Get suite from first highest-priority file entry providing the candidate.
        suite = None
        max_file_priority = float("-inf")
        for line in version_table_for_candidate[1:]:
            file_parts = line.split()
            if len(file_parts) < 5:
                # Skip /var/lib/dpkg/status and other possibly malformed entries.
                continue
            # Use a strict > and not a >= so that we respect source order.
            this_file_priority = int(file_parts[0])
            if this_file_priority > max_file_priority:
                max_file_priority = this_file_priority
                suite = file_parts[2].split("/")[0].split("-")[0]
                assert suite, "invalid suite, bug in policy parsing?"

        return priority, suite


class TestbedPath:
    """Represent a file/dir with a host and a testbed path"""

    def __init__(
        self,
        testbed: Testbed,
        host: str,
        tb: str,
        is_dir: Optional[bool] = None,
    ) -> None:
        """Create a TestbedPath object.

        The object itself is just a pair of file names, nothing more. They do
        not need to exist until you call copyup() or copydown() on them.

        testbed: the Testbed object which this refers to
        host: path of the file on the host
        tb: path of the file in testbed
        is_dir: whether path is a directory; None for "unspecified" if you only
                need copydown()
        """
        self.testbed = testbed
        self.host = host
        self.tb = tb
        self.is_dir = is_dir

    def copydown(
        self,
        check_existing=False,
        mode="",
    ) -> None:
        """Copy file from the host to the testbed

        If check_existing is True, don't copy if the testbed path already
        exists.
        """
        if check_existing and self.testbed.execute(["test", "-e", self.tb])[0] == 0:
            adtlog.debug("copydown: tb path %s already exists" % self.tb)
            return

        # create directory on testbed
        self.testbed.check_exec(["mkdir", "-p", os.path.dirname(self.tb)])

        if os.path.isdir(self.host):
            # directories need explicit '/' appended for VirtSubproc
            self.testbed.command("copydown", (self.host + "/", self.tb + "/"))
        else:
            self.testbed.command("copydown", (self.host, self.tb))

        # we usually want our files be readable for the non-root user
        if mode:
            self.testbed.check_exec(["chmod", "-R", mode, "--", self.tb])
        elif self.testbed.user:
            rc = self.testbed.execute(
                ["chown", "-R", self.testbed.user, "--", self.tb],
                stderr=subprocess.PIPE,
            )[0]
            if rc != 0:
                # chowning doesn't work on all shared downtmps, try to chmod
                # instead

                # I believe this is only needed for autopkgtest-virt-unshare, which
                # no longer advertises a downtmp at all.
                # But we allow it still, with --insecure, in case someone needs it.
                assert self.testbed.insecure
                self.testbed.check_exec(["chmod", "-R", "go+rwX", "--", self.tb])

    def copyup(self, check_existing=False):
        """Copy file from the testbed to the host

        If check_existing is True, don't copy if the host path already
        exists.
        """
        if check_existing and os.path.exists(self.host):
            adtlog.debug("copyup: host path %s already exists" % self.host)
            return

        os.makedirs(os.path.dirname(self.host), exist_ok=True, mode=0o2755)
        assert self.is_dir is not None
        if self.is_dir:
            self.testbed.command("copyup", (self.tb + "/", self.host + "/"))
        else:
            self.testbed.command("copyup", (self.tb, self.host))


class TempPath(TestbedPath):
    """Represent a file in the hosts'/testbed's temporary directories

    These are only guaranteed to exit within one testbed run.
    """

    # private; used to make sure the names of autocleaned files don't collide
    _filename_prefix = 1

    def __init__(
        self,
        testbed: Testbed,
        name: str,
        is_dir: bool = False,
        autoclean: bool = True,
    ) -> None:
        """Create a temporary TestbedPath object.

        The object itself is just a pair of file names, nothing more. They do
        not need to exist until you call copyup() or copydown() on them.

        testbed: the Testbed object which this refers to
        name: name of the temporary file (without path); host and tb
              will then be derived from that
        is_dir: whether path is a directory; None for "unspecified" if you only
                need copydown()
        autoclean: If True (default), remove file when test finishes. Should
                be set to False for files which you want to keep in the
                --output-dir which are useful for reporting results, like test
                stdout/err, log files, and binaries.
        """
        # if the testbed supports a shared downtmp, use that to avoid
        # unnecessary copying, unless we want to permanently keep the file
        if testbed.shared_downtmp and (not testbed.output_dir or autoclean):
            host = testbed.shared_downtmp
        else:
            host = testbed.output_dir
        self.autoclean = autoclean
        if autoclean:
            name = str(self._filename_prefix) + "-" + name
            TempPath._filename_prefix += 1

        assert testbed.scratch, "undefined scratch directory"
        TestbedPath.__init__(
            self,
            testbed,
            os.path.join(host, name),
            os.path.join(testbed.scratch, name),
            is_dir,
        )

    def __del__(self):
        if self.autoclean:
            if os.path.exists(self.host):
                try:
                    os.unlink(self.host)
                except OSError as e:
                    if e.errno == errno.EPERM:
                        self.testbed.check_exec(["rm", "-rf", self.tb])
                    else:
                        raise


#
# Helper functions
#


def child_ps(pid):
    """Get all child processes of pid"""

    try:
        out = subprocess.check_output(["ps", "-o", "pid=", "--ppid", str(pid)])
        return [int(p) for p in out.split()]
    except subprocess.CalledProcessError:
        return []


def killtree(pid, kill_signal=signal.SIGTERM):
    """Recursively kill pid and all of its children"""

    for child in child_ps(pid):
        killtree(child, kill_signal)
    try:
        os.kill(pid, kill_signal)
    except OSError:
        pass


def locate_setup_command(name: str) -> str:
    # Shortcut for shipped scripts
    if "/" not in name:
        shipped = os.path.join(PKGDATADIR, "setup-commands", name)

        if os.path.exists(shipped):
            return shipped

    return name