mmdebstrap/coverage.py

#!/usr/bin/env python3

from debian.deb822 import Deb822, Release
import email.utils
import os
import sys
import shutil
import subprocess
import argparse
import time
from datetime import timedelta
from collections import defaultdict

have_qemu = os.getenv("HAVE_QEMU", "yes") == "yes"
have_unshare = os.getenv("HAVE_UNSHARE", "yes") == "yes"
have_binfmt = os.getenv("HAVE_BINFMT", "yes") == "yes"
run_ma_same_tests = os.getenv("RUN_MA_SAME_TESTS", "yes") == "yes"
cmd = os.getenv("CMD", "./mmdebstrap")

default_dist = os.getenv("DEFAULT_DIST", "unstable")
all_dists = ["oldstable", "stable", "testing", "unstable"]
default_mode = "auto" if have_unshare else "root"
all_modes = ["auto", "root", "unshare", "fakechroot", "chrootless"]
default_variant = "apt"
all_variants = [
    "extract",
    "custom",
    "essential",
    "apt",
    "minbase",
    "buildd",
    "-",
    "standard",
]
default_format = "auto"
all_formats = ["auto", "directory", "tar", "squashfs", "ext2", "null"]

mirror = os.getenv("mirror", "http://127.0.0.1/debian")
hostarch = subprocess.check_output(["dpkg", "--print-architecture"]).decode().strip()

release_path = f"./shared/cache/debian/dists/{default_dist}/Release"
if not os.path.exists(release_path):
    print("path doesn't exist:", release_path, file=sys.stderr)
    print("run ./make_mirror.sh first", file=sys.stderr)
    exit(1)
if os.getenv("SOURCE_DATE_EPOCH") is not None:
    s_d_e = os.getenv("SOURCE_DATE_EPOCH")
else:
    with open(release_path) as f:
        rel = Release(f)
    s_d_e = str(email.utils.mktime_tz(email.utils.parsedate_tz(rel["Date"])))

separator = (
    "------------------------------------------------------------------------------"
)


def skip(condition, dist, mode, variant, fmt):
    if not condition:
        return ""
    for line in condition.splitlines():
        if not line:
            continue
        if eval(line):
            return line.strip()
    return ""


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("test", nargs="*", help="only run these tests")
    parser.add_argument(
        "-x",
        "--exitfirst",
        action="store_const",
        dest="maxfail",
        const=1,
        help="exit instantly on first error or failed test.",
    )
    parser.add_argument(
        "--maxfail",
        metavar="num",
        action="store",
        type=int,
        dest="maxfail",
        default=0,
        help="exit after first num failures or errors.",
    )
    parser.add_argument("--mode", metavar="mode", help="only run tests with this mode")
    parser.add_argument("--dist", metavar="dist", help="only run tests with this dist")
    args = parser.parse_args()

    # copy over files from git or as distributed
    for (git, dist, target) in [
        ("./mmdebstrap", "/usr/bin/mmdebstrap", "mmdebstrap"),
        ("./taridshift", "/usr/bin/mmtaridshift", "taridshift"),
        ("./tarfilter", "/usr/bin/mmtarfilter", "tarfilter"),
        (
            "./proxysolver",
            "/usr/lib/apt/solvers/mmdebstrap-dump-solution",
            "proxysolver",
        ),
        (
            "./ldconfig.fakechroot",
            "/usr/libexec/mmdebstrap/ldconfig.fakechroot",
            "ldconfig.fakechroot",
        ),
    ]:
        if os.path.exists(git):
            shutil.copy(git, f"shared/{target}")
        else:
            shutil.copy(dist, f"shared/{target}")
    # copy over hooks from git or as distributed
    if os.path.exists("hooks"):
        shutil.copytree("hooks", "shared/hooks", dirs_exist_ok=True)
    else:
        shutil.copytree(
            "/usr/share/mmdebstrap/hooks", "shared/hooks", dirs_exist_ok=True
        )

    tests = []
    with open("coverage.txt") as f:
        for test in Deb822.iter_paragraphs(f):
            name = test["Test"]
            dists = test.get("Dists", default_dist)
            if dists == "any":
                dists = all_dists
            elif dists == "default":
                dists = [default_dist]
            else:
                dists = dists.split()
            modes = test.get("Modes", default_mode)
            if modes == "any":
                modes = all_modes
            elif modes == "default":
                modes = [default_mode]
            else:
                modes = modes.split()
            variants = test.get("Variants", default_variant)
            if variants == "any":
                variants = all_variants
            elif variants == "default":
                variants = [default_variant]
            else:
                variants = variants.split()
            formats = test.get("Formats", default_format)
            if formats == "any":
                formats = all_formats
            elif formats == "default":
                formats = [default_format]
            else:
                formats = formats.split()
            for dist in dists:
                for mode in modes:
                    for variant in variants:
                        for fmt in formats:
                            skipreason = skip(
                                test.get("Skip-If"), dist, mode, variant, fmt
                            )
                            if skipreason:
                                tt = ("skip", skipreason)
                            elif have_qemu:
                                tt = "qemu"
                            elif test.get("Needs-QEMU", "false") == "true":
                                tt = ("skip", "test needs QEMU")
                            elif test.get("Needs-Root", "false") == "true":
                                tt = "sudo"
                            elif mode == "auto" and not have_unshare:
                                tt = "sudo"
                            elif mode == "root":
                                tt = "sudo"
                            elif mode == "unshare" and not have_unshare:
                                tt = ("skip", "test needs unshare")
                            else:
                                tt = "null"
                            tests.append((tt, name, dist, mode, variant, fmt))

    torun = []
    num_tests = len(tests)
    if args.test:
        # check if all given tests are either a valid name or a valid number
        for test in args.test:
            if test in [name for (_, name, _, _, _, _) in tests]:
                continue
            if not test.isdigit():
                print(f"cannot find test named {test}", file=sys.stderr)
                exit(1)
            if int(test) >= len(tests) or int(test) <= 0 or str(int(test)) != test:
                print(f"test number {test} doesn't exist", file=sys.stderr)
                exit(1)

        for i, (_, name, _, _, _, _) in enumerate(tests):
            # if either the number or test name matches, then we use this test,
            # otherwise we skip it
            if name in args.test:
                torun.append(i)
            if str(i + 1) in args.test:
                torun.append(i)
        num_tests = len(torun)

    starttime = time.time()
    skipped = defaultdict(list)
    failed = []
    num_success = 0
    num_finished = 0
    for i, (test, name, dist, mode, variant, fmt) in enumerate(tests):
        if torun and i not in torun:
            continue
        print(separator, file=sys.stderr)
        print("(%d/%d) %s" % (i + 1, len(tests), name), file=sys.stderr)
        print("dist: %s" % dist, file=sys.stderr)
        print("mode: %s" % mode, file=sys.stderr)
        print("variant: %s" % variant, file=sys.stderr)
        print("format: %s" % fmt, file=sys.stderr)
        if num_finished > 0:
            currenttime = time.time()
            timeleft = timedelta(
                seconds=int(
                    (num_tests - num_finished)
                    * (currenttime - starttime)
                    / num_finished
                )
            )
            print("time left: %s" % timeleft, file=sys.stderr)
        if failed:
            print("failed: %d" % len(failed))
        num_finished += 1
        with open("tests/" + name) as fin, open("shared/test.sh", "w") as fout:
            for line in fin:
                line = line.replace("{{ CMD }}", cmd)
                line = line.replace("{{ SOURCE_DATE_EPOCH }}", s_d_e)
                line = line.replace("{{ DIST }}", dist)
                line = line.replace("{{ MIRROR }}", mirror)
                line = line.replace("{{ MODE }}", mode)
                line = line.replace("{{ VARIANT }}", variant)
                line = line.replace("{{ FORMAT }}", fmt)
                line = line.replace("{{ HOSTARCH }}", hostarch)
                fout.write(line)
        argv = None
        match test:
            case "qemu":
                argv = ["./run_qemu.sh"]
            case "sudo":
                argv = ["./run_null.sh", "SUDO"]
            case "null":
                argv = ["./run_null.sh"]
            case ("skip", reason):
                skipped[reason].append(
                    ("(%d/%d) %s" % (i + 1, len(tests), name), dist, mode, variant, fmt)
                )
                print(f"skipped because of {reason}", file=sys.stderr)
                continue
        print(separator, file=sys.stderr)
        if args.dist and args.dist != dist:
            print(f"skipping because of --dist={args.dist}", file=sys.stderr)
            continue
        if args.mode and args.mode != mode:
            print(f"skipping because of --mode={args.mode}", file=sys.stderr)
            continue
        proc = subprocess.Popen(argv)
        try:
            proc.wait()
        except KeyboardInterrupt:
            proc.terminate()
            proc.wait()
            break
        print(separator, file=sys.stderr)
        if proc.returncode != 0:
            failed.append(
                ("(%d/%d) %s" % (i + 1, len(tests), name), dist, mode, variant, fmt)
            )
            print("result: FAILURE", file=sys.stderr)
        else:
            print("result: SUCCESS", file=sys.stderr)
            num_success += 1
        if args.maxfail and len(failed) >= args.maxfail:
            break
    print(
        "successfully ran %d tests" % num_success,
        file=sys.stderr,
    )
    if skipped:
        print("skipped %d:" % sum([len(v) for v in skipped.values()]), file=sys.stderr)
        for reason, l in skipped.items():
            print(f"skipped because of {reason}:", file=sys.stderr)
            for t in l:
                print(f"    {t}", file=sys.stderr)
    if failed:
        print("failed %d:" % len(failed), file=sys.stderr)
        for t in failed:
            print(f"    {t}", file=sys.stderr)
        exit(1)


if __name__ == "__main__":
    main()