dev/rust-excuses.py

   1 #!/usr/bin/pypy3
   2 # pypy3 runs faster than python3
   3 #
   4 # the excuses file can be found here:
   5 # https://release.debian.org/britney/excuses.yaml
   6
   7 import re
   8 import subprocess
   9 import sys
  10 import yaml
  11
  12 if len(sys.argv) != 4:
  13     print(
  14         """Generates dot files to show the migration deps
  15 Usage: %s excuses.dot excuses_arch.dot regressions.list
  16
  17 Expects excuses.yaml in the current dir
  18     """
  19         % sys.argv[0]
  20     )
  21     sys.exit(0)
  22
  23 print("parsing excuses.yaml...", file=sys.stderr)
  24 with open("excuses.yaml") as fp:
  25     y = yaml.load(fp, Loader=yaml.FullLoader)
  26
  27 excuses = {}
  28 for e in y["sources"]:
  29     excuses[e["source"]] = e
  30
  31 print("generating dot files...", file=sys.stderr)
  32 rust_excuses = open(sys.argv[1], "w")
  33 rust_excuses_arch = open(sys.argv[2], "w")
  34 rust_regressions = open(sys.argv[3], "w")
  35
  36 already_seen = set()
  37
  38 def edge_dep(name, dep):
  39     return " ".join(['"%s"' % name, "->", '"%s"' % dep])
  40
  41 def edge_dep_label(name, dep, label):
  42     return " ".join(['"%s"' % name, "->", '"%s"' % dep, '[label="%s"]' % label])
  43
  44 def print_all(*args, **kwargs):
  45     print(*args, **kwargs, file=rust_excuses)
  46     print(*args, **kwargs, file=rust_excuses_arch)
  47
  48 is_in_debian_cache = {}
  49 def is_in_debian(src):
  50     global is_in_debian_cache
  51     if src not in is_in_debian_cache:
  52         n = subprocess.check_output(
  53             "apt-cache showsrc %s 2>/dev/null | grep ^Package: | wc -l" % src,
  54             shell=True,
  55         )
  56         is_in_debian_cache[src] = int(n.strip())
  57     x = is_in_debian_cache[src]
  58     return x
  59
  60
  61 BG_NOT_IN_DEBIAN = "#cc0000"
  62 BG_OLD_IN_DEBIAN = "#ffcc66"
  63 BG_TOO_NEW = "#66ff99"
  64 BG_MISC_WAIT = "#ffcc99"
  65 BG_MISC_FAIL = "#ff6666"
  66 BG_SOURCEONLY = "#9999ff"
  67 VERDICT_FAIL = ("REJECTED_CANNOT_DETERMINE_IF_PERMANENT", "REJECTED_NEEDS_APPROVAL", "REJECTED_PERMANENTLY")
  68
  69 def traverse(name, arch="", d=0):
  70     if name in already_seen:
  71         return
  72     already_seen.add(name)
  73     dependencies = excuses.get(name, {}).get("dependencies", {})
  74
  75     edges = set()
  76     for arch, deps in dependencies.get("unsatisfiable-dependencies", {}).items():
  77         for dep in deps:
  78             vers = ""
  79             if dep.startswith("librust-"):
  80                 try:
  81                     results = re.match(r"librust-(\S+?)(-[.0-9]+)?(\+\S*)?-dev", dep)
  82                     dep = "rust-" + results.group(1)
  83                     if results.group(2):
  84                         vers = results.group(2)
  85                 except Exception:
  86                     print(dep, file=sys.stderr)
  87                     raise
  88             edges.add(edge_dep_label(name, dep, vers))
  89             print(
  90                 edge_dep_label(name, dep, "%s/%s" % (vers, arch) if vers else arch),
  91                 file=rust_excuses_arch,
  92             )
  93             if dep not in already_seen:
  94                 if is_in_debian(dep):
  95                     print_all('"%s" [fillcolor="%s",style=filled]' % (dep, BG_OLD_IN_DEBIAN))
  96                 else:
  97                     print_all('"%s" [fillcolor="%s",style=filled]' % (dep, BG_NOT_IN_DEBIAN))
  98     for edge in edges:
  99         print(edge, file=rust_excuses)
 100
 101     for dep in dependencies.get("migrate-after", []) + dependencies.get(
 102         "blocked-by", []
 103     ):
 104         if "/" in dep:
 105             dep, arch = dep.split("/")
 106             print(edge_dep_label(name, dep, arch), file=rust_excuses_arch)
 107         else:
 108             print_all(edge_dep(name, dep))
 109         traverse(dep, arch, d + 1)
 110
 111     policy = excuses.get(name, {}).get("policy_info", {})
 112     failed = {k: v["verdict"] for (k, v) in policy.items() if v.get("verdict", "") != "PASS"}
 113     attrs = {"shape": "box"}
 114     if "age" in failed:
 115         del failed["age"]
 116         attrs.update({ "fillcolor": BG_TOO_NEW, "style": "filled" })
 117     if "autopkgtest" in failed:
 118         for k, v in policy["autopkgtest"].items():
 119             if k == "verdict": continue
 120             for uu in v.values():
 121                 if "REGRESSION" in uu:
 122                     for u in uu:
 123                         if u and u.startswith("https://ci.debian.net/data/autopkgtest/testing"):
 124                             print(u, file=rust_regressions)
 125     if failed:
 126         if list(failed.keys()) == ["builtonbuildd"]:
 127             bg = BG_SOURCEONLY
 128             attrs.update({ "label": "\\N\\nneeds source-only upload" })
 129         elif any(v in VERDICT_FAIL for v in failed.values()):
 130             bg = BG_MISC_FAIL
 131             attrs.update({ "label": "\\N\\nfailed: %s" % ",".join(failed.keys()) })
 132         else:
 133             bg = BG_MISC_WAIT
 134             attrs.update({ "label": "\\N\\nwaiting: %s" % ",".join(failed.keys()) })
 135         attrs.update({ "fillcolor": bg, "style": "filled" })
 136     print_all('"%s" [%s]' % (name, ",".join("%s=\"%s\"" % p for p in attrs.items())))
 137
 138
 139 # import code
 140 # code.interact(local=locals())
 141
 142
 143 print_all("digraph {")
 144 for s in excuses.keys():
 145     if s.startswith("rust-"):
 146         traverse(s)
 147 print_all("}")