#!/usr/bin/env python3
"""Score every test in the run history, print a CLI table, write an HTML report.
Reads the JSONL history written by collect.py (one line per test attempt) and
answers one question: which tests changed their mind on unchanged code?
Usage:
python analyze.py # table only
python analyze.py --html flaky-report.html # table + HTML report
python analyze.py --json flaky.json # machine-readable
python analyze.py --detect 0.10 --quarantine 0.35 # tune thresholds
Stdlib only. No install step.
"""
import argparse, json, os, sys, html, datetime, collections
DETECT_DEFAULT = 0.10 # flakiness rate at/above this = FLAKY
QUARANTINE_DEFAULT = 0.35 # flakiness rate at/above this = quarantine candidate
# ---------- math ----------
def score_test(attempts):
"""attempts = chronological list of 'passed'/'failed'. Returns the stats dict.
flakiness_rate = failures / attempts (the headline number)
flip_rate = flips / (attempts - 1) (how erratic the order is)
A test that never passes is BROKEN, not flaky: it is a real failure.
"""
outcomes = [a for a in attempts if a in ("passed", "failed")]
total = len(outcomes)
if total == 0:
return None
passes = outcomes.count("passed")
failures = outcomes.count("failed")
flips = sum(1 for a, b in zip(outcomes, outcomes[1:]) if a != b)
flakiness_rate = failures / total
pass_rate = passes / total
flip_rate = flips / (total - 1) if total > 1 else 0.0
if failures == 0:
verdict = "STABLE"
elif passes == 0:
verdict = "BROKEN" # always fails: fix it, do not call it flake
else:
verdict = "FLAKY"
return {
"attempts": total, "passes": passes, "failures": failures,
"flips": flips, "flakiness_rate": flakiness_rate,
"pass_rate": pass_rate, "flip_rate": flip_rate,
"verdict": verdict, "outcomes": outcomes,
}
def flake_score(tests):
"""Suite reliability, 0-100. 100 = nothing flaky.
Only FLAKY tests drag the score down. BROKEN tests are failures, not flake,
so they are reported separately instead of being averaged into the score.
"""
if not tests:
return 100.0
penalty = sum(t["flakiness_rate"] for t in tests.values() if t["verdict"] == "FLAKY")
return round(100.0 * (1 - penalty / len(tests)), 1)
# ---------- io ----------
def load(history):
if not os.path.exists(history):
sys.exit(f"no history at {history}. Run collect.py first.")
by_test = collections.OrderedDict()
meta = {"runs": set(), "frameworks": set()}
with open(history) as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
r = json.loads(line)
except json.JSONDecodeError:
continue
if r.get("status") == "skipped":
continue
by_test.setdefault(r["id"], {"id": r["id"], "file": r.get("file", ""),
"framework": r.get("framework", ""),
"attempts": [], "messages": []})
by_test[r["id"]]["attempts"].append(r.get("status"))
if r.get("status") == "failed" and r.get("message"):
by_test[r["id"]]["messages"].append(r["message"])
meta["runs"].add(r.get("run"))
if r.get("framework"):
meta["frameworks"].add(r["framework"])
return by_test, meta
def analyze(history, detect, quarantine):
raw, meta = load(history)
tests = {}
for tid, t in raw.items():
s = score_test(t["attempts"])
if not s:
continue
s.update({"id": tid, "file": t["file"], "framework": t["framework"],
"message": (t["messages"] or [""])[0][:300]})
if s["verdict"] == "FLAKY" and s["flakiness_rate"] < detect:
s["verdict"] = "STABLE" # below the detection floor: treat as noise-free
s["quarantine"] = s["verdict"] == "FLAKY" and s["flakiness_rate"] >= quarantine
tests[tid] = s
return tests, meta
# ---------- cli table ----------
def bar(t):
return "".join("✓" if o == "passed" else "✗" for o in t["outcomes"])
def cli_table(tests, meta, score, detect, quarantine, color=True):
C = {"g": "\033[32m", "r": "\033[31m", "y": "\033[33m", "b": "\033[1m", "d": "\033[2m", "x": "\033[0m"}
if not color:
C = {k: "" for k in C}
order = {"FLAKY": 0, "BROKEN": 1, "STABLE": 2}
rows = sorted(tests.values(), key=lambda t: (order[t["verdict"]], -t["flakiness_rate"], t["id"]))
runs = len([r for r in meta["runs"] if r is not None])
fw = ", ".join(sorted(meta["frameworks"])) or "unknown"
name_w = min(max([len(shorten(t["id"])) for t in rows] + [16]), 54)
hdr = f"{'TEST':<{name_w}} {'RESULTS':<10} {'PASS':>5} {'FAIL':>5} {'RATE':>6} {'FLIP':>6} VERDICT"
line = "-" * len(hdr)
out = []
out.append("")
out.append(f"{C['b']}Flaky Test Report{C['x']} {C['d']}{runs} runs - {fw} - {len(tests)} tests{C['x']}")
out.append(line)
out.append(f"{C['b']}{hdr}{C['x']}")
out.append(line)
for t in rows:
col = C["y"] if t["verdict"] == "FLAKY" else (C["r"] if t["verdict"] == "BROKEN" else C["g"])
v = t["verdict"] + (" +QUARANTINE" if t.get("quarantine") else "")
out.append(f"{shorten(t['id']):<{name_w}} {bar(t):<10} {t['passes']:>5} {t['failures']:>5} "
f"{t['flakiness_rate']*100:>5.0f}% {t['flip_rate']*100:>5.0f}% {col}{v}{C['x']}")
out.append(line)
n_flaky = sum(1 for t in tests.values() if t["verdict"] == "FLAKY")
n_broken = sum(1 for t in tests.values() if t["verdict"] == "BROKEN")
n_stable = sum(1 for t in tests.values() if t["verdict"] == "STABLE")
n_q = sum(1 for t in tests.values() if t.get("quarantine"))
sc = C["g"] if score >= 95 else (C["y"] if score >= 80 else C["r"])
out.append(f"{C['b']}FlakeScore: {sc}{score} / 100{C['x']} "
f"{C['g']}{n_stable} stable{C['x']} {C['y']}{n_flaky} flaky{C['x']} "
f"{C['r']}{n_broken} broken{C['x']} {n_q} to quarantine")
out.append(f"{C['d']}flakiness rate = failures / attempts "
f"detect >= {detect*100:.0f}% quarantine >= {quarantine*100:.0f}%{C['x']}")
out.append("")
return "\n".join(out)
def shorten(tid, n=54):
return tid if len(tid) <= n else "..." + tid[-(n - 3):]
# ---------- html report ----------
def html_report(tests, meta, score, detect, quarantine):
e = lambda s: html.escape(str(s), quote=True)
rows = sorted(tests.values(), key=lambda t: ({"FLAKY": 0, "BROKEN": 1, "STABLE": 2}[t["verdict"]],
-t["flakiness_rate"], t["id"]))
runs = len([r for r in meta["runs"] if r is not None])
fw = ", ".join(sorted(meta["frameworks"])) or "unknown"
n = {v: sum(1 for t in tests.values() if t["verdict"] == v) for v in ("STABLE", "FLAKY", "BROKEN")}
n_q = sum(1 for t in tests.values() if t.get("quarantine"))
ring = "#16a34a" if score >= 95 else ("#ea580c" if score >= 80 else "#dc2626")
ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
trs = []
for t in rows:
cls = t["verdict"].lower()
beads = "".join(
f'' for o in t["outcomes"])
q = 'QUARANTINE' if t.get("quarantine") else ""
msg = f'
flakiness rate = failures / attempts -
flip rate = outcome changes / (attempts - 1)
Flagged FLAKY at >= {detect*100:.0f}% - quarantine candidate at >= {quarantine*100:.0f}%.
A test that never passes is BROKEN, not flaky: that is a real failure, fix the test or the app.
FlakeScore = 100 x (1 - sum(flakiness of flaky tests) / total tests)
'''
def main():
ap = argparse.ArgumentParser(description="Score flaky tests from collected run history.")
ap.add_argument("--history", default=".flaky/history.jsonl")
ap.add_argument("--html", metavar="FILE", help="also write an HTML report")
ap.add_argument("--json", metavar="FILE", help="also write machine-readable JSON")
ap.add_argument("--detect", type=float, default=DETECT_DEFAULT)
ap.add_argument("--quarantine", type=float, default=QUARANTINE_DEFAULT)
ap.add_argument("--no-color", action="store_true")
ap.add_argument("--fail-on-flaky", action="store_true", help="exit 1 if any test is flaky (CI gate)")
a = ap.parse_args()
tests, meta = analyze(a.history, a.detect, a.quarantine)
score = flake_score(tests)
print(cli_table(tests, meta, score, a.detect, a.quarantine,
color=not a.no_color and sys.stdout.isatty()))
if a.html:
with open(a.html, "w") as fh:
fh.write(html_report(tests, meta, score, a.detect, a.quarantine))
print(f"HTML report: {a.html}")
if a.json:
payload = {
"timestamp": datetime.datetime.now().isoformat(),
"totalRuns": len([r for r in meta["runs"] if r is not None]),
"totalTests": len(tests), "flakeScore": score,
"flakyTests": [
{"testName": t["id"], "file": t["file"], "flakinessRate": round(t["flakiness_rate"], 3),
"flipRate": round(t["flip_rate"], 3), "passes": t["passes"], "failures": t["failures"],
"quarantine": t["quarantine"], "failureMessages": [t["message"]] if t["message"] else []}
for t in tests.values() if t["verdict"] == "FLAKY"],
"brokenTests": [t["id"] for t in tests.values() if t["verdict"] == "BROKEN"],
}
with open(a.json, "w") as fh:
json.dump(payload, fh, indent=2)
print(f"JSON: {a.json}")
if a.fail_on_flaky and any(t["verdict"] == "FLAKY" for t in tests.values()):
sys.exit(1)
if __name__ == "__main__":
main()