From 084741f6956f4a076ce8da3b32902642f77eda9a Mon Sep 17 00:00:00 2001 From: jsboige Date: Sun, 30 Aug 2026 06:11:31 +0200 Subject: [PATCH] feat(guards,#13562): source-output ratchet - changed code must not ride identical outputs Mirror of check_papermill_ratchet: catches a changed code SOURCE riding byte-identical outputs (rule C.2's missing gate - founding case PR #13550). Exemptions imported from validate_pr_notebooks (Lean/QC-Cloud), per-cell body escape door, merge-base resolution like the papermill ratchet. Tests 19/19 incl. git-backed positive control (#13550 reconstructed: FAIL unpatched / PASS patched). Co-Authored-By: Claude-Code --- .../check_source_output_ratchet.py | 342 ++++++++++++++++++ .../tests/test_check_source_output_ratchet.py | 291 +++++++++++++++ 2 files changed, 633 insertions(+) create mode 100644 scripts/notebook_tools/check_source_output_ratchet.py create mode 100644 scripts/tests/test_check_source_output_ratchet.py diff --git a/scripts/notebook_tools/check_source_output_ratchet.py b/scripts/notebook_tools/check_source_output_ratchet.py new file mode 100644 index 0000000000..d5b0406a28 --- /dev/null +++ b/scripts/notebook_tools/check_source_output_ratchet.py @@ -0,0 +1,342 @@ +"""Source-output ratchet: changed code must not ride unchanged outputs. + +Issue #13562. Mirror image of check_papermill_ratchet.py: that gate catches +new outputs riding an old ``metadata.papermill`` block; this one catches a +changed code SOURCE riding byte-identical outputs. It is literally rule C.2 +of CLAUDE.md - "modify a code cell = re-execute before commit" - which had +no organ: check_exec_sequence and check_exec_ratchet only look at +execution_count (untouched here), and check_papermill_ratchet's +precondition (outputs changed) is false in exactly this case. + +Founding measurement (issue #13562, PR #13550): three cells with modified +source and byte-identical outputs on a PR that was green end to end. The +committed outputs had been produced by a source that no longer existed. + +Verdict per CODE cell of every .ipynb changed between BASE and HEAD +(cells are indexed over ALL cells, the convention the issue's evidence +uses): + + source changed AND outputs byte-identical AND outputs non-empty + -> STALE_OUTPUT (regression: the output describes a source + that no longer exists) + everything else -> clean + +Deliberate design choices from the issue body: +- outputs empty (``[]``) never fails: that is H.3's turf (never-executed), + there is nothing to go stale. +- a comment-only edit FAILS all the same: statically indistinguishable + from a semantic change, and a re-execution is cheap; the escape door + below is the relief valve, not a heuristic. +- notebooks added by the PR (no base blob) are skipped: their execution + evidence is notebook-execution-required.yml's job (H.3). +- cells not executable on any worker machine reuse validate_pr_notebooks' + exemption predicates - Lean kernels and QuantBook/QC-Cloud notebooks - + never a second parallel list. .NET Interactive is NOT exempt: it runs + on every worker machine (same rationale as ALLOW_NULL_EXEC_COUNT_KERNELS + excluding .NET in validate_pr_notebooks.py). + +Escape door (case "cannot change the output" and GPU notebooks whose +re-exec is routed to another machine): a written sentence in the PR body, +per cell, same mechanics as the repo's other lifts - a sentence, an +author, an hour; not a label, not a global skip: + + Source-output ratchet: [12] exempte -- comment-only edit, unchanged + output expected. + +An optional notebook qualifier disambiguates multi-notebook PRs: + + Source-output ratchet: MyIA.AI.Notebooks/Foo.ipynb: [12] exempte -- ... + +A bare ``[12]`` matches any notebook of the PR. + +Index pairing caveat: cells are paired by position over the full cell +list. An insertion shifts later indices; a shifted code cell then +compares against a different base cell, which can fabricate a +source-changed/outputs-identical pair. Rare, visible in the diff, and +liftable through the body door - a re-execution also dissolves it. + +Usage: + python check_source_output_ratchet.py [--json] [--body-file F] + + base-ref base branch ref (CI: origin/). Resolved + internally to merge-base(base-ref, HEAD) like the + papermill ratchet, so a branch behind its base is judged + on its own diff only. + --body-file PR body text carrying per-cell exemptions (optional). + +Head state reads the working tree (in CI the checkout IS the head), base +state reads the blob via ``git show``. Exit code 1 iff at least one +stale-output regression survives the exemptions. +""" + +import argparse +import json +import re +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from validate_pr_notebooks import ( # noqa: E402 + ALLOW_NULL_EXEC_COUNT_KERNELS, + QC_CLOUD_PATHS, + QUANTBOOK_PATTERN, +) + +# Same exclusions as check_papermill_ratchet.py / check_exec_ratchet.py: +# archived, papermill-output and research copies are not deliverable +# notebooks. +EXCLUDE_MARKERS = ("/.ipynb_checkpoints/", "/archive/", "/_output/", + "/research/") + +# Kernels no worker machine can execute (Lean via lean4_jupyter/alectryon +# is advisory everywhere else in the repo). .NET Interactive is +# deliberately absent - see module docstring. +NON_EXECUTABLE_KERNELS = set(ALLOW_NULL_EXEC_COUNT_KERNELS) | {"lean"} + +# "Source-output ratchet: [12] exempte" / "Source-output ratchet: +# Path/To.ipynb: [12] exempte" +_EXEMPT_RE = re.compile( + r"Source-output ratchet:\s*(?:(?P[^:\[\n]+?):\s*)?" + r"\[(?P\d+)\]\s*exempte", + re.IGNORECASE, +) + + +def git(*args, cwd=None): + """Run a git command, returning stdout (utf-8) or None on failure.""" + try: + out = subprocess.run(["git", *args], cwd=cwd, capture_output=True, + encoding="utf-8", errors="replace", check=False) + except OSError: + return None + return out.stdout if out.returncode == 0 else None + + +def resolve_base(base, cwd=None): + """Resolve `base` to merge-base(base, HEAD). + + The gate must judge what the branch CHANGED, not how far behind it is + (measured on #11528: 15 phantom changed notebooks against origin/main + vs 1 against the merge base). Same rationale as the papermill ratchet. + """ + out = git("merge-base", base, "HEAD", cwd=cwd) + return out.strip() if out and out.strip() else base + + +def changed_notebooks(base, cwd=None): + """Notebooks added/copied/modified/renamed between base and HEAD.""" + out = git("diff", "--name-only", "--diff-filter=ACMR", + base, "HEAD", "--", "*.ipynb", cwd=cwd) + if out is None: + return [] + paths = [] + for line in out.splitlines(): + posix = line.strip().replace("\\", "/") + if posix and not any(m in f"/{posix}" for m in EXCLUDE_MARKERS): + paths.append(posix) + return sorted(paths) + + +def canonical_source(cell): + """Canonical serialization of a cell's source lines.""" + return "".join(cell.get("source") or []) + + +def canonical_outputs(cell): + """Canonical serialization of a cell's outputs.""" + return json.dumps(cell.get("outputs") or [], sort_keys=True, + ensure_ascii=False) + + +def notebook_kernel(nb): + """kernelspec.name of a notebook, or None.""" + return ((nb.get("metadata") or {}).get("kernelspec") or {}).get("name") + + +def notebook_exempt_reason(path, nb): + """Why a whole notebook is out of the ratchet's reach, or None. + + Predicates are validate_pr_notebooks' own (no second list): Lean + kernels cannot run on a worker, QuantBook notebooks only execute in + QC Cloud. The QuantBook check tests the SOURCE, not the path - the + path fast-path alone left 16 notebooks out (see #8056). + """ + kernel = notebook_kernel(nb) + if kernel in NON_EXECUTABLE_KERNELS: + return "kernel" + posix = path.replace("\\", "/") + if any(posix.startswith(p) or f"/{p}" in f"/{posix}" + for p in QC_CLOUD_PATHS): + return "qc_path" + if QUANTBOOK_PATTERN.search(json.dumps(nb, ensure_ascii=False)): + return "quantbook" + return None + + +def parse_body_exemptions(body_text): + """Set of (path-or-None, cell-index) lifted by the PR body.""" + lifted = set() + if not body_text: + return lifted + for m in _EXEMPT_RE.finditer(body_text): + path = m.group("path") + if path: + path = path.strip().replace("\\", "/").rstrip(":").strip() + lifted.add((path, int(m.group("idx")))) + return lifted + + +def classify_cells(base_nb, head_nb): + """Per-code-cell records of one notebook pair, indexed over ALL cells. + + Pure function over parsed notebooks: the tests build fixtures, the CLI + feeds git blobs. No exemption logic here - the caller applies + notebook-level and body-level lifts so the classification stays + reusable for the tier-1 sweep. + """ + base_cells = base_nb.get("cells", []) + head_cells = head_nb.get("cells", []) + records = [] + for i, hcell in enumerate(head_cells): + if hcell.get("cell_type") != "code": + continue + bcell = base_cells[i] if i < len(base_cells) else None + if bcell is None or bcell.get("cell_type") != "code": + records.append({"index": i, "verdict": "UNPAIRED", + "regression": False}) + continue + if canonical_source(bcell) == canonical_source(hcell): + records.append({"index": i, "verdict": "UNCHANGED", + "regression": False}) + continue + outs_nonempty = bool(hcell.get("outputs")) + if not outs_nonempty: + # Nothing to go stale; H.3 owns the never-executed case. + records.append({"index": i, "verdict": "NO_OUTPUTS", + "regression": False}) + continue + if canonical_outputs(bcell) == canonical_outputs(hcell): + records.append({"index": i, "verdict": "STALE_OUTPUT", + "regression": True}) + else: + records.append({"index": i, "verdict": "EXECUTED", + "regression": False}) + return records + + +def _parse_notebook(text): + try: + return json.loads(text), "OK" + except Exception: + return None, "PARSE_ERROR" + + +def classify_notebook(path, base_nb, head_nb, body_exemptions): + """Apply notebook-level and body-level lifts to classify_cells.""" + exempt = notebook_exempt_reason(path, head_nb) + if exempt: + return [{"verdict": f"EXEMPT_{exempt.upper()}", "regression": False}] + lifted = {idx for (p, idx) in body_exemptions + if p is None or p == path.replace("\\", "/")} + records = [] + for rec in classify_cells(base_nb, head_nb): + if rec["regression"] and rec["index"] in lifted: + rec = {"index": rec["index"], "verdict": "EXEMPT_BODY", + "regression": False} + records.append(rec) + return records + + +def ratchet(base, cwd=None, body_text=""): + """One record per changed notebook; regression = stale output ridden.""" + base = resolve_base(base, cwd=cwd) + body_exemptions = parse_body_exemptions(body_text) + records = [] + for path in changed_notebooks(base, cwd=cwd): + content = git("show", f"{base}:{path}", cwd=cwd) + if content is None: + records.append({"notebook": path, "verdict": "ADDED", + "cells": [], "regressions": 0}) + continue + base_nb, b_status = _parse_notebook(content) + try: + head_nb = json.loads((Path(cwd or ".") / path).read_text( + encoding="utf-8")) + h_status = "OK" + except Exception: + head_nb, h_status = None, "PARSE_ERROR" + if b_status != "OK" or h_status != "OK": + records.append({"notebook": path, "verdict": "PARSE_ERROR", + "cells": [], "regressions": 0}) + continue + cells = classify_notebook(path, base_nb, head_nb, body_exemptions) + records.append({ + "notebook": path, + "verdict": "CHANGED", + "cells": [{"index": c.get("index"), "verdict": c["verdict"], + "regression": c["regression"]} for c in cells], + "regressions": sum(1 for c in cells if c["regression"]), + }) + return records + + +def main(): + ap = argparse.ArgumentParser( + description="Source-output ratchet: changed code must not ride " + "unchanged outputs (issue #13562)") + ap.add_argument("base", help="Base git ref (CI: origin/)") + ap.add_argument("--json", action="store_true", dest="as_json", + help="Machine-readable output") + ap.add_argument("--body-file", dest="body_file", default=None, + help="PR body text carrying per-cell exemptions") + args = ap.parse_args() + + body_text = "" + if args.body_file: + try: + body_text = Path(args.body_file).read_text(encoding="utf-8") + except OSError as exc: + print(f"cannot read body file: {exc}", file=sys.stderr) + sys.exit(2) + + records = ratchet(args.base, body_text=body_text) + total_reg = sum(r["regressions"] for r in records) + failing = [r for r in records if r["regressions"]] + + if args.as_json: + print(json.dumps({ + "base": args.base, + "changed": len(records), + "regressions": total_reg, + "records": records}, ensure_ascii=False, indent=1)) + else: + print(f"source-output ratchet -- base {args.base}") + print(f"changed notebooks : {len(records)}") + print(f"stale cells : {total_reg}") + for r in records: + mark = (f" REGRESSION x{r['regressions']}" + if r["regressions"] else "") + print(f" {r['verdict']:12s} {r['notebook']}{mark}") + for c in r["cells"]: + if c["regression"]: + print(f" [{c['index']}] STALE_OUTPUT") + + if failing: + for r in failing: + for c in r["cells"]: + if c["regression"]: + print(f"::error file={r['notebook']}::cell [{c['index']}] " + f"source changed but outputs are byte-identical to " + f"{args.base} - the committed output was produced " + f"by a source that no longer exists. Re-execute, " + f"or lift with 'Source-output ratchet: " + f"[{c['index']}] exempte -- ' in the PR " + f"body", file=sys.stderr) + sys.exit(1) + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/scripts/tests/test_check_source_output_ratchet.py b/scripts/tests/test_check_source_output_ratchet.py new file mode 100644 index 0000000000..fca678b746 --- /dev/null +++ b/scripts/tests/test_check_source_output_ratchet.py @@ -0,0 +1,291 @@ +"""Tests for check_source_output_ratchet.py (issue #13562). + +The centerpiece is the positive control the issue body demands: reconstruct +the #13550 case (source of cell [26] modified, base outputs kept) and prove +the guard FAILS on it, then PASSES once the outputs are refreshed. The +reconstruction is git-backed - a throwaway repository with a base commit +and a head commit - so the test exercises the real CLI path (resolve_base, +changed_notebooks, git show, exit codes), not just the pure functions. + +The module is loaded from its file path because scripts/tests sits outside +the scripts package root and the scripts/ tree is a namespace-package +minefield on Windows: importlib on the direct path sidesteps the ambiguity. +""" + +import importlib.util +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +TOOL = Path(__file__).resolve().parents[1] / "notebook_tools" / \ + "check_source_output_ratchet.py" + + +def _load(): + spec = importlib.util.spec_from_file_location( + "check_source_output_ratchet", TOOL) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +CSR = _load() + + +def md(*lines): + return {"cell_type": "markdown", "metadata": {}, "source": list(lines)} + + +def code(src, outputs, execution_count=1): + return {"cell_type": "code", "execution_count": execution_count, + "metadata": {}, "outputs": outputs, "source": [src]} + + +def nb(cells, kernel="python3"): + return {"cells": cells, + "metadata": {"kernelspec": {"display_name": kernel, + "name": kernel}}} + + +# Two DIFFERENT non-empty outputs: the founding case kept the base output +# byte-identical while editing the source; the patch refreshes it. +OUT_BASE = [{"output_type": "stream", "name": "stdout", "text": ["42\n"]}] +OUT_REFRESHED = [{"output_type": "execute_result", + "data": {"text/plain": ["42"]}, + "execution_count": 7, "metadata": {}}] + + +def fixture_13550_base(): + """27 cells, index 26 = the code cell the founding PR edited.""" + cells = [md("# Titre"), md("## Section")] + cells += [code(f"x{i} = {i}", [{"output_type": "stream", + "name": "stdout", + "text": [f"{i}\n"]}]) for i in range(24)] + cells.append(code("resultat = 40 + 2\nprint(resultat)", OUT_BASE)) + return nb(cells) + + +def fixture_13550_head(outputs): + """Same notebook after the source edit, outputs per `outputs`.""" + cells = [md("# Titre"), md("## Section modifiee")] + cells += [code(f"x{i} = {i}", [{"output_type": "stream", + "name": "stdout", + "text": [f"{i}\n"]}]) for i in range(24)] + cells.append(code("resultat = 41 + 1 # reformule\nprint(resultat)", + outputs)) + return nb(cells) + + +class GitRepo: + """Throwaway repository: one commit per notebook state.""" + + def __init__(self, states): + self.dir = tempfile.TemporaryDirectory() + self.path = Path(self.dir.name) + self._git("init", "-q") + path = "MyIA.AI.Notebooks/Fake/Fake.ipynb" + (self.path / "MyIA.AI.Notebooks/Fake").mkdir(parents=True) + for i, state in enumerate(states): + (self.path / path).write_text( + json.dumps(state, ensure_ascii=False, indent=1) + "\n", + encoding="utf-8") + self._git("add", "-A") + self._git("-c", "user.email=t@t", "-c", "user.name=t", + "commit", "-q", "-m", f"state {i}") + + def _git(self, *args): + out = subprocess.run(["git", *args], cwd=self.path, + capture_output=True, check=True) + return out + + def run_guard(self, base="HEAD~1", body=None): + cmd = [sys.executable, str(TOOL), base, "--json"] + if body is not None: + body_file = self.path / "body.md" + body_file.write_text(body, encoding="utf-8") + cmd += ["--body-file", str(body_file)] + return subprocess.run(cmd, cwd=self.path, capture_output=True, + text=True, encoding="utf-8") + + def close(self): + self.dir.cleanup() + + +class TestPositiveControl13550(unittest.TestCase): + """The guard the issue demands: FAIL unpatched, PASS patched.""" + + def setUp(self): + # Cell [26] source edited, base outputs kept: the founding defect. + self.repo = GitRepo([fixture_13550_base(), + fixture_13550_head(OUT_BASE)]) + + def tearDown(self): + self.repo.close() + + def test_fails_on_reconstructed_defect(self): + proc = self.repo.run_guard() + self.assertEqual(proc.returncode, 1, proc.stderr) + payload = json.loads(proc.stdout) + self.assertEqual(payload["regressions"], 1) + # The target cell is the LAST code cell (index 26 over all cells); + # the 24 filler code cells before it are UNCHANGED and come first. + cell = payload["records"][0]["cells"][-1] + self.assertEqual(cell["index"], 26) + self.assertEqual(cell["verdict"], "STALE_OUTPUT") + + def test_passes_once_outputs_refreshed(self): + repo = GitRepo([fixture_13550_base(), + fixture_13550_head(OUT_REFRESHED)]) + try: + proc = repo.run_guard() + self.assertEqual(proc.returncode, 0, proc.stderr) + payload = json.loads(proc.stdout) + self.assertEqual(payload["regressions"], 0) + cell = payload["records"][0]["cells"][-1] + self.assertEqual(cell["verdict"], "EXECUTED") + finally: + repo.close() + + def test_body_exemption_lifts_the_defect(self): + body = ("Source-output ratchet: [26] exempte -- comment-only edit, " + "unchanged output expected.") + proc = self.repo.run_guard(body=body) + self.assertEqual(proc.returncode, 0, proc.stderr) + payload = json.loads(proc.stdout) + cell = payload["records"][0]["cells"][-1] + self.assertEqual(cell["verdict"], "EXEMPT_BODY") + + +class TestClassifyCells(unittest.TestCase): + """Pure classification, indexed over ALL cells.""" + + def test_source_changed_outputs_identical_is_stale(self): + base = nb([md("intro"), code("print(1)", OUT_BASE)]) + head = nb([md("intro"), code("print(2)", OUT_BASE)]) + recs = CSR.classify_cells(base, head) + self.assertEqual(recs[0]["index"], 1) + self.assertEqual(recs[0]["verdict"], "STALE_OUTPUT") + self.assertTrue(recs[0]["regression"]) + + def test_comment_only_edit_fails_all_the_same(self): + base = nb([code("print(1) # note", OUT_BASE)]) + head = nb([code("print(1) # note corrigee", OUT_BASE)]) + recs = CSR.classify_cells(base, head) + self.assertEqual(recs[0]["verdict"], "STALE_OUTPUT") + self.assertTrue(recs[0]["regression"]) + + def test_empty_outputs_never_fail(self): + base = nb([code("print(1)", [])]) + head = nb([code("print(2)", [])]) + recs = CSR.classify_cells(base, head) + self.assertEqual(recs[0]["verdict"], "NO_OUTPUTS") + self.assertFalse(recs[0]["regression"]) + + def test_unchanged_source_is_clean(self): + base = nb([code("print(1)", OUT_BASE)]) + head = nb([code("print(1)", OUT_BASE)]) + self.assertEqual(CSR.classify_cells(base, head)[0]["verdict"], + "UNCHANGED") + + def test_markdown_cells_are_skipped(self): + base = nb([md("a"), code("print(1)", OUT_BASE), md("b")]) + head = nb([md("A"), code("print(1)", OUT_BASE), md("B")]) + recs = CSR.classify_cells(base, head) + self.assertEqual(len(recs), 1) + self.assertEqual(recs[0]["index"], 1) + self.assertFalse(recs[0]["regression"]) + + def test_all_cell_indexing_survives_markdown_edits(self): + # Markdown edits around a code cell must not shift its index: the + # founding evidence quotes cell [26] of the ALL-cell list. + base = fixture_13550_base() + head = fixture_13550_head(OUT_REFRESHED) + recs = CSR.classify_cells(base, head) + self.assertEqual(recs[-1]["index"], 26) + + def test_inserted_cell_is_unpaired_not_stale(self): + # A cell inserted before a code cell shifts it: no base partner at + # that index -> UNPAIRED (clean), never a fabricated stale pair. + base = nb([code("print(1)", OUT_BASE)]) + head = nb([md("nouveau"), code("print(1)", OUT_BASE)]) + recs = CSR.classify_cells(base, head) + self.assertEqual([r["verdict"] for r in recs], ["UNPAIRED"]) + self.assertFalse(any(r["regression"] for r in recs)) + + +class TestNotebookExemptions(unittest.TestCase): + """validate_pr_notebooks' predicates, reused not duplicated.""" + + def test_lean_kernel_exempt(self): + base = nb([code("print(1)", OUT_BASE)], kernel="lean4-wsl") + head = nb([code("print(2)", OUT_BASE)], kernel="lean4-wsl") + recs = CSR.classify_notebook("Foo/Bar.lean.ipynb", base, head, set()) + self.assertEqual(recs[0]["verdict"], "EXEMPT_KERNEL") + self.assertFalse(recs[0]["regression"]) + + def test_qc_cloud_path_exempt(self): + path = "MyIA.AI.Notebooks/QuantConnect/Python/Research.ipynb" + base = nb([code("print(1)", OUT_BASE)]) + head = nb([code("print(2)", OUT_BASE)]) + recs = CSR.classify_notebook(path, base, head, set()) + self.assertEqual(recs[0]["verdict"], "EXEMPT_QC_PATH") + + def test_quantbook_source_exempt(self): + path = "MyIA.AI.Notebooks/Elsewhere/Research.ipynb" + base = nb([code("qb = QuantBook()", OUT_BASE)]) + head = nb([code("qb = QuantBook() # edit", OUT_BASE)]) + recs = CSR.classify_notebook(path, base, head, set()) + self.assertEqual(recs[0]["verdict"], "EXEMPT_QUANTBOOK") + + def test_dotnet_not_exempt(self): + base = nb([code("Console.WriteLine(1)", OUT_BASE)], + kernel=".net-csharp") + head = nb([code("Console.WriteLine(2)", OUT_BASE)], + kernel=".net-csharp") + recs = CSR.classify_notebook("Foo/Bar.ipynb", base, head, set()) + self.assertEqual(recs[0]["verdict"], "STALE_OUTPUT") + self.assertTrue(recs[0]["regression"]) + + +class TestBodyExemptions(unittest.TestCase): + def test_bare_index_matches_any_notebook(self): + lifted = CSR.parse_body_exemptions( + "avant\nSource-output ratchet: [12] exempte -- raison\napres") + self.assertEqual(lifted, {(None, 12)}) + + def test_qualified_index_matches_only_that_notebook(self): + lifted = CSR.parse_body_exemptions( + "Source-output ratchet: MyIA.AI.Notebooks/Foo.ipynb: [12] " + "exempte -- raison") + self.assertEqual(lifted, {("MyIA.AI.Notebooks/Foo.ipynb", 12)}) + + def test_case_insensitive_and_multiple(self): + lifted = CSR.parse_body_exemptions( + "source-output ratchet: [1] exempte -- a\n" + "Source-Output Ratchet: [2] exempte -- b") + self.assertEqual(lifted, {(None, 1), (None, 2)}) + + def test_no_sentence_no_lift(self): + self.assertEqual(CSR.parse_body_exemptions("sorties [12] stables"), + set()) + + def test_qualifier_scopes_the_lift(self): + body = ("Source-output ratchet: MyIA.AI.Notebooks/Foo.ipynb: [0] " + "exempte -- raison") + lifted = CSR.parse_body_exemptions(body) + base = nb([code("print(1)", OUT_BASE)]) + head = nb([code("print(2)", OUT_BASE)]) + foo = CSR.classify_notebook("MyIA.AI.Notebooks/Foo.ipynb", + base, head, lifted) + bar = CSR.classify_notebook("MyIA.AI.Notebooks/Bar.ipynb", + base, head, lifted) + self.assertEqual(foo[0]["verdict"], "EXEMPT_BODY") + self.assertEqual(bar[0]["verdict"], "STALE_OUTPUT") + self.assertTrue(bar[0]["regression"]) + + +if __name__ == "__main__": + unittest.main()