A Regression Manager

JOBS = [
    ("which runs exist", "suite config", "a file in the repo — 34.5"),
    ("where each run happens", "isolated dirs", "35.3 already does this"),
    ("how many at once", "--jobs", "bounded, and not from cpu_count"),
    ("what to skip", "--resume", "the hard one — §7"),
    ("what the answer is", "aggregation", "sorted, or it is not a report"),
    ("why it failed", "failure buckets", "35.1 already does this"),
    ("what the flow sees", "one exit code", "0 only if every job passed"),
]

print(f"{'responsibility':<24} {'mechanism':<18} note")
for what, how, note in JOBS:
    print(f"{what:<24} {how:<18} {note}")

print()
print("responsibilities         :", len(JOBS))
print("already solved by 35.1/35.3:", 2)
print("new because of concurrency :", 3)
print("the one with a wrong answer")
print("  that looks right         :", "--resume")

responsibility           mechanism          note
which runs exist         suite config       a file in the repo — 34.5
where each run happens   isolated dirs      35.3 already does this
how many at once         --jobs             bounded, and not from cpu_count
what to skip             --resume           the hard one — §7
what the answer is       aggregation        sorted, or it is not a report
why it failed            failure buckets    35.1 already does this
what the flow sees       one exit code      0 only if every job passed

responsibilities         : 7
already solved by 35.1/35.3: 2
new because of concurrency : 3
the one with a wrong answer
  that looks right         : --resume

FLAGS = [
    ("--suite nightly", "names a list of (test, seed) pairs in config"),
    ("--jobs 32", "an explicit bound — never os.cpu_count()"),
    ("--resume", "skip runs that already produced a result"),
]

print(f"{'flag':<20} meaning")
for flag, meaning in FLAGS:
    print(f"{flag:<20} {meaning}")

print()
WHY_NOT_CPU = [
    ("os.cpu_count()", "the report depends on which machine ran it"),
    ("cpu_count() - 1", "same, plus an off-by-one nobody can justify"),
    ("a fixed 32", "reproducible, and wrong on a laptop"),
    ("--jobs, required", "the operator decides and it is recorded"),
]
print(f"{'how --jobs could be set':<22} consequence")
for how, why in WHY_NOT_CPU:
    print(f"{how:<22} {why}")

print()
print("the machine's cpu count is published here:", False)
print("because it is                            :", "machine-specific — 33.14")
print("and --jobs ends up in                    :", "cmd.sh, so the run is replayable")

flag                 meaning
--suite nightly      names a list of (test, seed) pairs in config
--jobs 32            an explicit bound — never os.cpu_count()
--resume             skip runs that already produced a result

how --jobs could be set consequence
os.cpu_count()         the report depends on which machine ran it
cpu_count() - 1        same, plus an off-by-one nobody can justify
a fixed 32             reproducible, and wrong on a laptop
--jobs, required       the operator decides and it is recorded

the machine's cpu count is published here: False
because it is                            : machine-specific — 33.14
and --jobs ends up in                    : cmd.sh, so the run is replayable

from pathlib import Path

SUITE = [("axi_burst", 48210), ("axi_burst", 48211), ("axi_lite", 7734),
         ("axi_wrap", 9001)]


def fake_run(test, seed):
    """Stands in for 35.3's run_sim so this section is about the LOOP."""
    status = "FAIL" if seed % 2 == 1 else "PASS"
    return {"test": test, "seed": seed, "status": status}


# v1: a serial loop that prints
print("v1 — for each test, for each seed, run it and print")
passed = failed = 0
for test, seed in SUITE:
    r = fake_run(test, seed)
    print(f"  {r['test']} {r['seed']} -> {r['status']}")
    if r["status"] == "PASS":
        passed += 1
    else:
        failed += 1

print()
print("  passed:", passed, "failed:", failed)
print()
print("what it gets right:", ["every run happens", "a verdict per run",
                             "a total"])
print("what it cannot do :", ["run two at once", "resume", "group failures",
                              "produce a file"])
print()
print("runs in a real nightly  :", 400)
print("this loop's runtime for  ")
print("  400 serial runs        :", "400 x one run — the whole night")

v1 — for each test, for each seed, run it and print
  axi_burst 48210 -> PASS
  axi_burst 48211 -> FAIL
  axi_lite 7734 -> PASS
  axi_wrap 9001 -> FAIL

  passed: 2 failed: 2

what it gets right: ['every run happens', 'a verdict per run', 'a total']
what it cannot do : ['run two at once', 'resume', 'group failures', 'produce a file']

runs in a real nightly  : 400
this loop's runtime for  
  400 serial runs        : 400 x one run — the whole night

SUBMITTED = [("axi_burst", 48210), ("axi_burst", 48211), ("axi_lite", 7734),
             ("axi_wrap", 9001), ("axi_err", 2024)]

# An explicit completion order. This page does NOT race to produce one,
# because a race would make its own output nondeterministic.
COMPLETION = [("axi_wrap", 9001), ("axi_burst", 48210), ("axi_err", 2024),
              ("axi_lite", 7734), ("axi_burst", 48211)]


def report(rows):
    return "\n".join(f"{t},{s}" for t, s in rows)


print("submitted order :", [f"{t}:{s}" for t, s in SUBMITTED])
print("completion order:", [f"{t}:{s}" for t, s in COMPLETION])
print()
print("a report written as runs finish == a report in submitted order:",
      report(COMPLETION) == report(SUBMITTED))
print("after sorting both                                           :",
      report(sorted(COMPLETION)) == report(sorted(SUBMITTED)))
print()
print("so two runs of the same suite produce:", "different files")
print("and 30.3's new-versus-fixed diff sees :", "every row as changed")
print()
BROKEN = [
    ("report in completion order", "a diff shows every row changed", "sort"),
    ("resume, marker written early", "a run is lost silently", "§7"),
    ("a shared scratch path", "two runs overwrite each other", "33.7"),
]
print(f"{'what concurrency breaks':<30} {'symptom':<32} fix")
for what, sym, fix in BROKEN:
    print(f"{what:<30} {sym:<32} {fix}")

submitted order : ['axi_burst:48210', 'axi_burst:48211', 'axi_lite:7734', 'axi_wrap:9001', 'axi_err:2024']
completion order: ['axi_wrap:9001', 'axi_burst:48210', 'axi_err:2024', 'axi_lite:7734', 'axi_burst:48211']

a report written as runs finish == a report in submitted order: False
after sorting both                                           : True

so two runs of the same suite produce: different files
and 30.3's new-versus-fixed diff sees : every row as changed

what concurrency breaks        symptom                          fix
report in completion order     a diff shows every row changed   sort
resume, marker written early   a run is lost silently           §7
a shared scratch path          two runs overwrite each other    33.7

from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Job:
    test: str
    seed: int

    @property
    def name(self):
        return f"{self.test}_{self.seed}"

    def run_dir(self, root):
        return Path(root) / self.name


SUITES = {
    "smoke": {"tests": ["axi_burst", "axi_lite"], "seeds": [48210]},
    "nightly": {"tests": ["axi_burst", "axi_lite", "axi_wrap"],
                "seeds": [48210, 48211, 7734]},
}


def plan(suite_name, suites=SUITES):
    """Pure: config in, an ordered list of Jobs out."""
    cfg = suites[suite_name]
    return [Job(t, s) for t in cfg["tests"] for s in cfg["seeds"]]


for name in sorted(SUITES):
    jobs = plan(name)
    print(f"{name:<10} {len(jobs):>3} jobs  "
          f"{len(SUITES[name]['tests'])} tests x "
          f"{len(SUITES[name]['seeds'])} seeds")

print()
nightly = plan("nightly")
print("the first four jobs:")
for j in nightly[:4]:
    print(f"  {j.name:<22} dir={j.run_dir('runs')}")

print()
print("the plan is deterministic      :", plan("nightly") == plan("nightly"))
print("job names are unique           :",
      len({j.name for j in nightly}) == len(nightly))
print("two seeds of one test collide  :", False)
print("because the dir name includes  :", "the seed — 35.3 §9")

nightly      9 jobs  3 tests x 3 seeds
smoke        2 jobs  2 tests x 1 seeds

the first four jobs:
  axi_burst_48210        dir=runs/axi_burst_48210
  axi_burst_48211        dir=runs/axi_burst_48211
  axi_burst_7734         dir=runs/axi_burst_7734
  axi_lite_48210         dir=runs/axi_lite_48210

the plan is deterministic      : True
job names are unique           : True
two seeds of one test collide  : False
because the dir name includes  : the seed — 35.3 §9

import json
from pathlib import Path


def run_job(name, root, marker_first, interrupt_on=None):
    """Simulates one run, with the two possible write orders."""
    d = Path(root) / name
    d.mkdir(parents=True, exist_ok=True)
    if marker_first:
        (d / "done").write_text("", encoding="utf-8")
    if interrupt_on == name:
        raise KeyboardInterrupt(f"the sweep was killed during {name}")
    (d / "result.json").write_text(
        json.dumps({"job": name, "status": "PASS"}), encoding="utf-8")
    if not marker_first:
        (d / "done").write_text("", encoding="utf-8")


def completed_by(root, filename):
    return {d.name for d in Path(root).iterdir()
            if (d / filename).exists()}


JOBS = [f"t{i}" for i in range(6)]

for marker_first in (True, False):
    root = Path("marker_first" if marker_first else "result_first")
    for name in JOBS:
        try:
            run_job(name, root, marker_first, interrupt_on="t3")
        except KeyboardInterrupt:
            break
    by_marker = completed_by(root, "done")
    by_result = completed_by(root, "result.json")
    label = "marker BEFORE result" if marker_first else "marker AFTER result"
    print(f"--- {label}")
    print(f"    jobs started            : 4  (killed during t3)")
    print(f"    dirs with a 'done' file : {sorted(by_marker)}")
    print(f"    dirs with a result      : {sorted(by_result)}")
    print(f"    a resume keyed on 'done' skips  : {len(by_marker)}")
    print(f"    results that actually exist     : {len(by_result)}")
    print(f"    skipped with NO result          : {sorted(by_marker - by_result)}")

--- marker BEFORE result
    jobs started            : 4  (killed during t3)
    dirs with a 'done' file : ['t0', 't1', 't2', 't3']
    dirs with a result      : ['t0', 't1', 't2']
    a resume keyed on 'done' skips  : 4
    results that actually exist     : 3
    skipped with NO result          : ['t3']
--- marker AFTER result
    jobs started            : 4  (killed during t3)
    dirs with a 'done' file : ['t0', 't1', 't2']
    dirs with a result      : ['t0', 't1', 't2']
    a resume keyed on 'done' skips  : 3
    results that actually exist     : 3
    skipped with NO result          : []

import json
from pathlib import Path
from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    test: str
    seed: int

    @property
    def name(self):
        return f"{self.test}_{self.seed}"


def result_path(root, job):
    return Path(root) / job.name / "result.json"


def is_done(root, job):
    """A job is done iff a COMPLETE result can be read back."""
    p = result_path(root, job)
    if not p.is_file():
        return False
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return False                     # a half-written file is not done
    return "status" in data


def to_run(root, jobs, resume):
    return [j for j in jobs if not (resume and is_done(root, j))]


JOBS = [Job("axi_burst", s) for s in (48210, 48211, 7734)]
ROOT = Path("runs")

# two jobs completed, one left a TRUNCATED result file
for j in JOBS[:2]:
    d = ROOT / j.name
    d.mkdir(parents=True, exist_ok=True)
    (d / "result.json").write_text(
        json.dumps({"job": j.name, "status": "PASS"}), encoding="utf-8")
d = ROOT / JOBS[2].name
d.mkdir(parents=True, exist_ok=True)
(d / "result.json").write_text('{"job": "axi_burst_7734", "stat',
                               encoding="utf-8")

print("jobs in the suite        :", len(JOBS))
print("result files on disk     :",
      sum(1 for j in JOBS if result_path(ROOT, j).is_file()))
print("jobs is_done() considers :", sum(1 for j in JOBS if is_done(ROOT, j)))
print()
print("without --resume, jobs to run:", len(to_run(ROOT, JOBS, resume=False)))
print("with    --resume, jobs to run:",
      [j.name for j in to_run(ROOT, JOBS, resume=True)])
print()
print("the truncated result is re-run:", True)
print("because is_done() requires    :", "a parseable result with a status")
print("and a half-written file is    :", "not a completed job")

jobs in the suite        : 3
result files on disk     : 3
jobs is_done() considers : 2

without --resume, jobs to run: 3
with    --resume, jobs to run: ['axi_burst_7734']

the truncated result is re-run: True
because is_done() requires    : a parseable result with a status
and a half-written file is    : not a completed job

from concurrent.futures import ThreadPoolExecutor, as_completed

JOBS = [f"job_{i}" for i in range(8)]


def work(name):
    return {"job": name, "status": "PASS" if int(name[-1]) % 3 else "FAIL"}


# ex.map preserves submission order, by documented contract
with ThreadPoolExecutor(max_workers=4) as pool:
    mapped = list(pool.map(work, JOBS))

print("ex.map returns results in submission order:",
      [r["job"] for r in mapped] == JOBS)

# as_completed does not promise any order
with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(work, n): n for n in JOBS}
    completed = [futures[f] for f in as_completed(futures)]

print("as_completed returns the same set         :",
      sorted(completed) == sorted(JOBS))
print("as_completed promises submission order    :", False)
print("sorting recovers it                       :",
      sorted(completed) == JOBS)
print()
print("jobs                 :", len(JOBS))
print("workers              :", 4)
print("max in flight at once:", 4)
print()
print("so the rule is:")
print("  use ex.map when you only need the results")
print("  use as_completed when you want to PRINT progress as they land")
print("  and in either case, SORT before writing the report")

ex.map returns results in submission order: True
as_completed returns the same set         : True
as_completed promises submission order    : False
sorting recovers it                       : True

jobs                 : 8
workers              : 4
max in flight at once: 4

so the rule is:
  use ex.map when you only need the results
  use as_completed when you want to PRINT progress as they land
  and in either case, SORT before writing the report

import re
from collections import defaultdict
from dataclasses import dataclass

LINE_NO, HEX, INT = (re.compile(r"\(\d+\)"), re.compile(r"0x[0-9a-fA-F]+"),
                     re.compile(r"\b\d+\b"))


@dataclass(frozen=True)
class Result:
    test: str
    seed: int
    status: str
    first_failure: str = ""

    @property
    def name(self):
        return f"{self.test}_{self.seed}"


def bucket_key(message):
    """35.1's normaliser, reused unchanged."""
    return INT.sub("N", HEX.sub("0xH", message))


RESULTS = [
    Result("axi_burst", 48210, "PASS"),
    Result("axi_burst", 48211, "FAIL", "drv.sv(88) data mismatch exp=0xa5"),
    Result("axi_burst", 7734, "FAIL", "drv.sv(91) data mismatch exp=0xb7"),
    Result("axi_lite", 48210, "PASS"),
    Result("axi_lite", 48211, "TIMEOUT", ""),
    Result("axi_lite", 7734, "FAIL", "mon.sv(12) response timeout after 500"),
    Result("axi_wrap", 48210, "BUILD", ""),
    Result("axi_wrap", 48211, "PASS"),
    Result("axi_wrap", 7734, "FATAL", "env.sv(3) virtual interface not set"),
]

by_status = defaultdict(list)
for r in sorted(RESULTS, key=lambda r: (r.test, r.seed)):
    by_status[r.status].append(r.name)

ORDER = ("BUILD", "FATAL", "TIMEOUT", "FAIL", "PASS")
print("by status, worst first:")
for status in ORDER:
    if status in by_status:
        print(f"  {status:<8} {len(by_status[status]):>2}  "
              f"{by_status[status]}")

buckets = defaultdict(list)
for r in sorted(RESULTS, key=lambda r: (r.test, r.seed)):
    if r.first_failure:
        buckets[bucket_key(r.first_failure)].append(r.name)

print()
print("failure buckets, largest first:")
for key, runs in sorted(buckets.items(), key=lambda kv: (-len(kv[1]), kv[0])):
    print(f"  {len(runs):>2}  {key}")
    print(f"      {runs}")

print()
total = len(RESULTS)
passed = len(by_status.get("PASS", []))
print("jobs            :", total)
print("passed          :", passed)
print("not passed      :", total - passed)
print("distinct causes :", len(buckets))
print("runs with no message to bucket:",
      sum(1 for r in RESULTS if r.status not in ("PASS",) and not r.first_failure))

by status, worst first:
  BUILD     1  ['axi_wrap_48210']
  FATAL     1  ['axi_wrap_7734']
  TIMEOUT   1  ['axi_lite_48211']
  FAIL      3  ['axi_burst_7734', 'axi_burst_48211', 'axi_lite_7734']
  PASS      3  ['axi_burst_48210', 'axi_lite_48210', 'axi_wrap_48211']

failure buckets, largest first:
   2  drv.sv(N) data mismatch exp=0xH
      ['axi_burst_7734', 'axi_burst_48211']
   1  env.sv(N) virtual interface not set
      ['axi_wrap_7734']
   1  mon.sv(N) response timeout after N
      ['axi_lite_7734']

jobs            : 9
passed          : 3
not passed      : 6
distinct causes : 3
runs with no message to bucket: 2

import json
import re
import subprocess
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, asdict
from pathlib import Path

EXIT = {"PASS": 0, "FAIL": 1, "FATAL": 1, "TIMEOUT": 1, "BUILD": 3}
DONE_MARKER = "TEST DONE"
STATUS_ORDER = ("BUILD", "FATAL", "TIMEOUT", "FAIL", "PASS")

TB = """module tb;
  reg clk = 0;
  initial begin
    $display("UVM_INFO @0: tb.sv(1) start seed=%0d", `SEED);
`ifdef INJECT_ERROR
    $display("UVM_ERROR @10: drv.sv(88) data mismatch exp=0xa5 got=0x5a");
`endif
    repeat (4) begin #5 clk = ~clk; end
`ifndef INJECT_TIMEOUT
    $display("TEST DONE");
`endif
    $finish;
  end
endmodule
"""


@dataclass(frozen=True)
class Job:
    test: str
    seed: int
    defines: tuple = ()

    @property
    def name(self):
        return f"{self.test}_{self.seed}"


def classify(text, rc, built):
    if not built:
        return "BUILD"
    lines = text.splitlines()
    errs = sum(1 for l in lines
               if l.startswith("UVM_ERROR") and "UVM_ERROR :" not in l)
    fat = sum(1 for l in lines
              if l.startswith("UVM_FATAL") and "UVM_FATAL :" not in l)
    done = any(l.strip() == DONE_MARKER for l in lines)
    if fat or rc != 0:
        return "FATAL"
    if not done:
        return "TIMEOUT"
    return "FAIL" if errs else "PASS"


def first_failure(text):
    for line in text.splitlines():
        if line.startswith(("UVM_ERROR", "UVM_FATAL")) \
           and " :" not in line[:12]:
            parts = line.split(": ", 1)
            return parts[1] if len(parts) == 2 else line
    return ""


def run_one(job, root):
    """35.3's runner, called once per job. Writes result.json LAST."""
    d = Path(root) / job.name
    d.mkdir(parents=True, exist_ok=True)
    (d / "tb.v").write_text(TB, encoding="utf-8")
    argv = (["iverilog", f"-DSEED={job.seed}"]
            + [f"-D{x}" for x in job.defines] + ["-o", "sim.vvp", "tb.v"])
    (d / "cmd.sh").write_text("#!/bin/sh\n" + " ".join(argv)
                              + "\nvvp sim.vvp\n", encoding="utf-8")
    b = subprocess.run(argv, cwd=d, stdout=subprocess.PIPE,
                       stderr=subprocess.STDOUT, text=True)
    if b.returncode != 0:
        status, text = "BUILD", b.stdout
    else:
        r = subprocess.run(["vvp", "sim.vvp"], cwd=d, stdout=subprocess.PIPE,
                           stderr=subprocess.STDOUT, text=True)
        text = r.stdout
        (d / "sim.log").write_text(text, encoding="utf-8")
        status = classify(text, r.returncode, built=True)
    record = {"test": job.test, "seed": job.seed, "status": status,
              "first_failure": first_failure(text) if status != "PASS" else ""}
    tmp = d / "result.json.tmp"
    tmp.write_text(json.dumps(record, sort_keys=True), encoding="utf-8")
    tmp.replace(d / "result.json")          # atomic — §7
    return record


def is_done(root, job):
    p = Path(root) / job.name / "result.json"
    if not p.is_file():
        return False
    try:
        return "status" in json.loads(p.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return False


def load_result(root, job):
    return json.loads((Path(root) / job.name / "result.json")
                      .read_text(encoding="utf-8"))


def run_suite(jobs, root, jobs_at_once, resume=False):
    pending = [j for j in jobs if not (resume and is_done(root, j))]
    skipped = [j for j in jobs if resume and is_done(root, j)]
    with ThreadPoolExecutor(max_workers=jobs_at_once) as pool:
        list(pool.map(lambda j: run_one(j, root), pending))
    # collect EVERYTHING from disk, in submission order — §8
    results = [load_result(root, j) for j in jobs]
    return results, len(pending), len(skipped)


SUITE = [
    Job("axi_burst", 48210),
    Job("axi_burst", 48211, ("INJECT_ERROR",)),
    Job("axi_lite", 7734, ("INJECT_TIMEOUT",)),
    Job("axi_wrap", 9001),
]
ROOT = Path("runs")

results, ran, skipped = run_suite(SUITE, ROOT, jobs_at_once=4, resume=False)

by_status = defaultdict(list)
for r in results:
    by_status[r["status"]].append(f"{r['test']}_{r['seed']}")

print("--- first pass")
print("  jobs run     :", ran, "| skipped:", skipped)
for s in STATUS_ORDER:
    if s in by_status:
        print(f"  {s:<8} {len(by_status[s])}  {by_status[s]}")

buckets = defaultdict(list)
for r in results:
    if r["first_failure"]:
        key = re.sub(r"\b\d+\b", "N",
                     re.sub(r"0x[0-9a-fA-F]+", "0xH",
                            re.sub(r"\(\d+\)", "(N)", r["first_failure"])))
        buckets[key].append(f"{r['test']}_{r['seed']}")

print("  buckets      :", len(buckets))
for k, v in sorted(buckets.items()):
    print(f"    {k}  {v}")

worst = min((STATUS_ORDER.index(r["status"]) for r in results), default=4)
code = EXIT[STATUS_ORDER[worst]]
print("  exit code    :", code)

# --- now resume: nothing should re-run
results2, ran2, skipped2 = run_suite(SUITE, ROOT, jobs_at_once=4, resume=True)
print()
print("--- resumed pass")
print("  jobs run     :", ran2, "| skipped:", skipped2)
print("  same results :", results == results2)
print()
print("the report is identical across the two passes:", results == results2)
print("because results are collected                :", "in submission order")
print("and the exit code is chosen                  :", "once, from the worst")

--- first pass
  jobs run     : 4 | skipped: 0
  TIMEOUT  1  ['axi_lite_7734']
  FAIL     1  ['axi_burst_48211']
  PASS     2  ['axi_burst_48210', 'axi_wrap_9001']
  buckets      : 1
    drv.sv(N) data mismatch exp=0xH got=0xH  ['axi_burst_48211']
  exit code    : 1

--- resumed pass
  jobs run     : 0 | skipped: 4
  same results : True

the report is identical across the two passes: True
because results are collected                : in submission order
and the exit code is chosen                  : once, from the worst

import io
import json
import unittest
from dataclasses import dataclass
from pathlib import Path

STATUS_ORDER = ("BUILD", "FATAL", "TIMEOUT", "FAIL", "PASS")
EXIT = {"PASS": 0, "FAIL": 1, "FATAL": 1, "TIMEOUT": 1, "BUILD": 3}


@dataclass(frozen=True)
class Job:
    test: str
    seed: int

    @property
    def name(self):
        return f"{self.test}_{self.seed}"


SUITES = {"nightly": {"tests": ["a", "b"], "seeds": [1, 2]}}


def plan(suite, suites=SUITES):
    cfg = suites[suite]
    return [Job(t, s) for t in cfg["tests"] for s in cfg["seeds"]]


def is_done(root, job):
    p = Path(root) / job.name / "result.json"
    if not p.is_file():
        return False
    try:
        return "status" in json.loads(p.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return False


def to_run(root, jobs, resume):
    return [j for j in jobs if not (resume and is_done(root, j))]


def overall_exit(statuses):
    if not statuses:
        return 3                      # an empty suite is not a pass
    worst = min(STATUS_ORDER.index(s) for s in statuses)
    return EXIT[STATUS_ORDER[worst]]


def write_result(root, job, status, partial=False):
    d = Path(root) / job.name
    d.mkdir(parents=True, exist_ok=True)
    text = ('{"status": "' + status + '"}') if not partial else '{"stat'
    (d / "result.json").write_text(text, encoding="utf-8")


class Manager(unittest.TestCase):

    def setUp(self):
        import tempfile, shutil
        self.root = Path(tempfile.mkdtemp())
        self.addCleanup(shutil.rmtree, self.root, True)

    def test_the_plan_is_deterministic(self):
        self.assertEqual(plan("nightly"), plan("nightly"))

    def test_job_names_are_unique(self):
        jobs = plan("nightly")
        self.assertEqual(len({j.name for j in jobs}), len(jobs))

    def test_without_resume_everything_runs(self):
        jobs = plan("nightly")
        write_result(self.root, jobs[0], "PASS")
        self.assertEqual(len(to_run(self.root, jobs, resume=False)),
                         len(jobs))

    def test_with_resume_completed_jobs_are_skipped(self):
        jobs = plan("nightly")
        write_result(self.root, jobs[0], "PASS")
        self.assertEqual([j.name for j in to_run(self.root, jobs, True)],
                         [j.name for j in jobs[1:]])

    def test_a_partial_result_is_rerun(self):
        """§7: a half-written JSON is not a completed job."""
        jobs = plan("nightly")
        write_result(self.root, jobs[0], "PASS", partial=True)
        self.assertIn(jobs[0], to_run(self.root, jobs, resume=True))

    def test_an_empty_dir_is_not_done(self):
        jobs = plan("nightly")
        (self.root / jobs[0].name).mkdir(parents=True)
        self.assertFalse(is_done(self.root, jobs[0]))

    def test_all_pass_exits_zero(self):
        self.assertEqual(overall_exit(["PASS", "PASS"]), 0)

    def test_one_failure_exits_one(self):
        self.assertEqual(overall_exit(["PASS", "FAIL", "PASS"]), 1)

    def test_a_build_failure_dominates(self):
        self.assertEqual(overall_exit(["PASS", "FAIL", "BUILD"]), 3)

    def test_an_empty_suite_is_not_a_pass(self):
        self.assertEqual(overall_exit([]), 3)

    def test_only_all_pass_gives_zero(self):
        for statuses in (["FAIL"], ["TIMEOUT"], ["FATAL"], ["BUILD"],
                         ["PASS", "FAIL"], []):
            with self.subTest(statuses=statuses):
                self.assertNotEqual(overall_exit(statuses), 0)

    def test_report_rows_are_in_submission_order(self):
        jobs = plan("nightly")
        for j in jobs:
            write_result(self.root, j, "PASS")
        completion = [jobs[3], jobs[0], jobs[2], jobs[1]]
        rows = [j.name for j in sorted(completion,
                                       key=lambda j: (j.test, j.seed))]
        self.assertEqual(rows, [j.name for j in jobs])


def run(case):
    suite = unittest.TestLoader().loadTestsFromTestCase(case)
    res = unittest.TextTestRunner(stream=io.StringIO(), verbosity=0).run(suite)
    return {"run": res.testsRun, "failures": len(res.failures),
            "ok": res.wasSuccessful()}


print("manager suite:", run(Manager))
print()
print("simulators launched in this suite:", 0)
print("behaviours pinned                :", 12)
print()
print("the tests that exist because of a measured bug:")
print("  test_a_partial_result_is_rerun        — §7")
print("  test_an_empty_suite_is_not_a_pass     — a vacuous PASS")
print("  test_report_rows_are_in_submission_order — §8")

manager suite: {'run': 12, 'failures': 0, 'ok': True}

simulators launched in this suite: 0
behaviours pinned                : 12

the tests that exist because of a measured bug:
  test_a_partial_result_is_rerun        — §7
  test_an_empty_suite_is_not_a_pass     — a vacuous PASS
  test_report_rows_are_in_submission_order — §8

import json
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

STATUS_ORDER = ("BUILD", "FATAL", "TIMEOUT", "FAIL", "PASS")
EXIT = {"PASS": 0, "FAIL": 1, "FATAL": 1, "TIMEOUT": 1, "BUILD": 3}


def overall_exit(statuses):
    if not statuses:
        return 3
    return EXIT[STATUS_ORDER[min(STATUS_ORDER.index(s) for s in statuses)]]


# A
print("A:", overall_exit(["PASS", "PASS"]), overall_exit(["PASS", "FAIL"]))

# B
print("B:", overall_exit([]), all([]))

# C
with ThreadPoolExecutor(max_workers=3) as pool:
    got = list(pool.map(lambda n: n * 2, range(6)))
print("C:", got == [0, 2, 4, 6, 8, 10])

# D
Path("d").mkdir(exist_ok=True)
(Path("d") / "result.json").write_text('{"stat', encoding="utf-8")
try:
    json.loads((Path("d") / "result.json").read_text(encoding="utf-8"))
    print("D:", "parsed")
except json.JSONDecodeError:
    print("D:", (Path("d") / "result.json").is_file(), "but not parseable")

A: 0 1
B: 3 True
C: True
D: True but not parseable