merge-queues-for-arch.py

# coding: utf-8
import argparse
import apt_pkg
import urllib.request
import urllib.parse
import itertools
import glob
import json
import os
import sys
import apt.progress.base

REQUEST_URL = "https://autopkgtest.ubuntu.com/request.cgi"


def get_proposed_packages():
    """Get all valid triggers in proposed, returns set of package/version strings."""
    tmpdir = "merge-requests.tmp"
    os.makedirs(os.path.join(tmpdir, "etc/apt"), exist_ok=True)
    os.makedirs(os.path.join(tmpdir, "var/lib/apt"), exist_ok=True)
    with open(os.path.join(tmpdir, "etc/apt/sources.list"), "w") as sources:
        print(
            "deb-src [signed-by=/usr/share/keyrings/ubuntu-archive-keyring.gpg] https://snapshot.ubuntu.com/ubuntu noble-proposed main universe restricted multiverse",
            file=sources,
        )

    apt_pkg.config["Dir"] = tmpdir
    apt_pkg.init()
    sl = apt_pkg.SourceList()
    sl.read_main_list()
    cache = apt_pkg.Cache(None)
    cache.update(apt.progress.base.AcquireProgress(), sl)

    srcrecords = apt_pkg.SourceRecords()

    proposed = set()
    while srcrecords.step():
        proposed.add(srcrecords.package + "/" + srcrecords.version)
    return proposed


def get_queue(release, arch):
    """Build a single queue of package, args tuples."""
    with urllib.request.urlopen("https://autopkgtest.ubuntu.com/queued.json") as url:
        queued = json.load(url)
    queue = [
        (i.split()[0], json.loads(i.split(None, 1)[1]))
        for i in queued["queues"]["ubuntu"][release][arch]["requests"]
        + queued["queues"]["huge"][release][arch]["requests"]
    ]
    print("input:", len(queue), file=sys.stderr)
    return queue


def merge_queue(queue, proposed):
    """Calculate the merged queue"""
    out = []
    for package, runs in itertools.groupby(
        sorted(queue, key=lambda i: i[0]), lambda i: i[0]
    ):
        triggers = set()
        for package, args in runs:
            for trigger in args["triggers"]:
                if trigger in proposed or trigger == "migration-reference/0":
                    triggers.add(trigger)
                else:
                    print(
                        "Dropping obsolete trigger",
                        trigger,
                        "for",
                        package,
                        file=sys.stderr,
                    )

        if "migration-reference/0" in triggers:
            out.append((package, {"triggers": ["migration-reference/0"]}))
            triggers.remove("migration-reference/0")
        if triggers:
            out.append((package, {"all-proposed": "1", "triggers": sorted(triggers)}))

    print("output:", len(out), file=sys.stderr)
    return out


def main():
    """Entry point."""
    parser = argparse.ArgumentParser()
    parser.add_argument("release")
    parser.add_argument("arch")
    parser.add_argument("--run-autopkgtest", action="store_true")
    args = parser.parse_args()
    proposed = get_proposed_packages()
    queue = get_queue(args.release, args.arch)
    merged = merge_queue(queue, proposed)

    # Print the commands or URLs to requeue them.
    for pkg, run_args in merged:
        if args.run_autopkgtest:
            print(
                "run-autopkgtest {allproposed}-s {release} -a {arch} {triggers} {pkg}".format(
                    allproposed="--all-proposed " if "all-proposed" in run_args else "",
                    arch=args.arch,
                    pkg=pkg,
                    release=args.release,
                    triggers=" ".join(
                        ["--trigger=%s" % trigger for trigger in run_args["triggers"]]
                    ),
                )
            )
        else:
            params = [("release", args.release), ("arch", args.arch), ("package", pkg)]
            params += [("trigger", t) for t in run_args["triggers"]]
            if "all-proposed" in run_args:
                params.append(("all-proposed", "1"))
            url = REQUEST_URL + "?" + urllib.parse.urlencode(params)
            print(url)


if __name__ == "__main__":
    main()