mmdebstrap/coverage.py

#!/usr/bin/env python3

from debian.deb822 import Deb822
import os
import sys
import shutil
import subprocess
import argparse

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"

default_dist = os.getenv("DEFAULT_DIST", "unstable")
all_dists = ["oldstable", "stable", "testing", "unstable"]
default_mode = "unshare" 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"]

only_dists = []

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


def skip(condition, dist, mode, variant, fmt):
    if not condition:
        return False
    toskip = False
    for line in condition.splitlines():
        if not line:
            continue
        if eval(line):
            toskip = True
            break
    return toskip


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.",
    )
    args = parser.parse_args()

    onlyrun = None
    if len(sys.argv) > 1:
        onlyrun = sys.argv[1]

    tests = []
    with open("coverage.txt") as f:
        for test in Deb822.iter_paragraphs(f):
            name = test["Test"]
            if args.test and name not in args.text:
                continue
            tt = None
            if have_qemu:
                tt = "qemu"
            elif test.get("Needs-QEMU", "false") == "true":
                tt = "skip"
            elif test.get("Needs-Root", "false") == "true":
                tt = "sudo"
            elif have_unshare:
                tt = "null"
            else:
                tt = "sudo"
            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:
                if only_dists and dist not in only_dists:
                    continue
                for mode in modes:
                    for variant in variants:
                        for fmt in formats:
                            if skip(test.get("Skip-If"), dist, mode, variant, fmt):
                                tt = "skip"
                            tests.append((tt, name, dist, mode, variant, fmt))

    skipped = []
    failed = []
    for i, (test, name, dist, mode, variant, fmt) in enumerate(tests):
        print(
            "------------------------------------------------------------------------------"
        )
        print("(%d/%d) %s" % (i + 1, len(tests), name))
        print("dist: %s" % dist)
        print("mode: %s" % mode)
        print("variant: %s" % variant)
        print("format: %s" % fmt)
        print(
            "------------------------------------------------------------------------------"
        )
        with open("tests/" + name) as fin, open("shared/test.sh", "w") as fout:
            for line in fin:
                for e in ["CMD", "SOURCE_DATE_EPOCH"]:
                    line = line.replace("{{ " + e + " }}", os.getenv(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":
                skipped.append((name, dist, mode, variant, fmt))
                continue
        proc = subprocess.Popen(argv)
        try:
            proc.wait()
        except KeyboardInterrupt:
            proc.kill()
            break
        if proc.returncode != 0:
            failed.append((name, dist, mode, variant, fmt))
        if args.maxfail and len(failed) >= args.maxfail:
            break
    if skipped:
        print("skipped:")
        for t in skipped:
            print(t)
    if failed:
        print("failed:")
        for t in failed:
            print(t)
        exit(1)


if __name__ == "__main__":
    main()