"""tools/compose_release.py's refusal paths and its curated swap, offline -- the code audit of 2026-09-21, missing item 3 (compose_release.main was 73/90 statements missed and compose_curated 20/21). main() is driven with its subprocess runner (cr.run), the worktree removal and the git helpers mocked: no worktree is created, no test suite or embed runs, no commit is made and no branch moves. python -m unittest tests.test_compose_release """ import contextlib import io import json import sys import tempfile import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "tools")) import compose_release as cr # noqa: E402 import predeploy as pd # noqa: E402 CODE, APPROVED, RELEASE = "c" * 40, "a" * 40, "e" * 40 STATE = {"approved": {"commit": APPROVED, "date": "2026-09-16"}, "metadata_files": ["review_state.json"]} def dumps(obj): return json.dumps(obj).encode("utf-8") class ComposeCuratedTests(unittest.TestCase): """The worktree's data/curated/*.json <- the approved commit's.""" def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.wt = Path(tmp.name) self.curated = self.wt / pd.CURATED_DIR self.curated.mkdir(parents=True) self.code_files = {"us_equivalency.json": {"pairings": ["held"]}, # differs "authorities.json": {"a": 1}, # equal "new_held.json": {"x": 1}, # not approved "review_state.json": {"approved": "main's"}} # metadata for name, obj in self.code_files.items(): (self.curated / name).write_text(json.dumps(obj, indent=1), encoding="utf-8") self.approved_files = {"us_equivalency.json": {"pairings": ["approved"]}, "authorities.json": {"a": 1}, "retired.json": {"r": 1}, # gone on main "review_state.json": {"approved": "older"}} def compose(self): names = {CODE: sorted(self.code_files), APPROVED: sorted(self.approved_files)} blobs = {f"{pd.CURATED_DIR}/{n}": dumps(o) for n, o in self.approved_files.items()} with mock.patch.object(pd, "curated_names", side_effect=lambda c: names[c]), \ mock.patch.object(pd, "show_bytes", side_effect=lambda c, p: blobs[p]): return cr.compose_curated(self.wt, CODE, APPROVED, STATE) def read(self, name): return json.loads((self.curated / name).read_text(encoding="utf-8")) def test_the_approved_json_replaces_mains_and_held_files_leave(self): replaced, held, restored = self.compose() self.assertEqual((replaced, held, restored), (["us_equivalency.json"], ["new_held.json"], ["retired.json"])) self.assertEqual(self.read("us_equivalency.json"), {"pairings": ["approved"]}) self.assertEqual(self.read("retired.json"), {"r": 1}) self.assertFalse((self.curated / "new_held.json").exists()) def test_an_equal_file_is_left_as_is_and_metadata_ships_from_main(self): before = (self.curated / "authorities.json").read_bytes() # indented: equal JSON self.compose() self.assertEqual((self.curated / "authorities.json").read_bytes(), before) self.assertEqual(self.read("review_state.json"), {"approved": "main's"}) class _Runner: """Stands in for cr.run: records every command and answers by its shape.""" def __init__(self, wt, tests_rc=0, predeploy_rc=0, staged=True): self.wt, self.tests_rc, self.predeploy_rc, self.staged = wt, tests_rc, predeploy_rc, staged self.cmds = [] def __call__(self, cmd, cwd, env=None, check=True): cmd = [str(c) for c in cmd] self.cmds.append(cmd) rc, text = 0, "" if cmd[1:3] == ["-c", "import canlex; print(canlex.__file__)"]: text = str(self.wt / "canlex" / "__init__.py") elif cmd[1:4] == ["-m", "unittest", "discover"]: rc = self.tests_rc text = ("FAIL: test_x (tests.test_x.T.test_x)\nRan 2001 tests\nFAILED (failures=1)" if rc else "Ran 2001 tests\nOK") elif cmd[1:3] == ["tools/predeploy.py", "--skip-tests"]: rc, text = self.predeploy_rc, "predeploy: " + ("FAIL" if self.predeploy_rc else "PASS") elif cmd[:4] == ["git", "diff", "--cached", "--quiet"]: rc = 1 if self.staged else 0 if check and rc != 0: raise RuntimeError(f"{cmd[0]} exited {rc}") return rc, text, 0.1 def ran(self, *prefix): return [c for c in self.cmds if c[:len(prefix)] == list(prefix) or c[1:1 + len(prefix)] == list(prefix)] class MainRefusalTests(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.wt = Path(tmp.name).resolve() (self.wt / "data" / "processed").mkdir(parents=True) self.after = [{"id": "commentary-x", "text": "approved"}] (self.wt / pd.COMMENTARY).write_text(json.dumps(self.after), encoding="utf-8") def main(self, argv, runner, before=None, gap=()): before = self.after if before is None else before manifest = {"dispositions": [], "pairings": [], "aliases": {}} git = mock.Mock(return_value=b"") with mock.patch.object(cr, "run", side_effect=runner), \ mock.patch.object(cr, "remove_worktree"), \ mock.patch.object(cr, "compose_curated", return_value=(["a.json"], [], [])), \ mock.patch.object(cr, "write_held_manifest", return_value=manifest), \ mock.patch.object(cr, "stale_report", return_value=["stale: none"]), \ mock.patch.object(cr, "compose_drafts", return_value=False), \ mock.patch.object(pd, "review_state", return_value=STATE), \ mock.patch.object(pd, "rev", side_effect=lambda ref, cwd=None: RELEASE if ref == "HEAD" else CODE if ref == "main" else ref), \ mock.patch.object(pd, "show_json", return_value=before), \ mock.patch.object(pd, "embeddings_gap", return_value=list(gap)), \ mock.patch.object(pd, "git", git), \ contextlib.redirect_stdout(io.StringIO()) as out: rc = cr.main(["--workdir", str(self.wt)] + argv) return rc, out.getvalue(), git def test_failing_composed_tests_produce_no_release(self): runner = _Runner(self.wt, tests_rc=1) rc, out, git = self.main([], runner) self.assertEqual(rc, 1) self.assertIn("THE COMPOSED TREE FAILS ITS TESTS -- not committed", out) self.assertIn("FAIL: test_x", out) self.assertTrue(out.rstrip().endswith("compose_release: no release produced"), out[-300:]) self.assertEqual(runner.ran("git", "commit"), []) self.assertEqual(runner.ran("tools/predeploy.py"), []) git.assert_not_called() # no branch moved def test_allowing_test_failures_never_passes_a_dry_run(self): rc, out, _ = self.main(["--no-commit", "--allow-test-failures"], _Runner(self.wt, tests_rc=1)) self.assertEqual(rc, 1) self.assertIn("(--allow-test-failures)", out) self.assertIn("compose_release: no release produced", out) def test_a_passing_dry_run_exits_0_without_a_commit(self): runner = _Runner(self.wt) rc, out, git = self.main(["--no-commit"], runner) self.assertEqual(rc, 0, out) self.assertNotIn("no release produced", out) self.assertEqual(runner.ran("git", "commit"), []) git.assert_not_called() def test_stale_vectors_block_the_commit(self): runner = _Runner(self.wt) rc, out, _ = self.main(["--no-embed"], runner, before=[{"id": "commentary-x", "text": "main's"}], gap=["commentary-x"]) self.assertEqual(rc, 1) self.assertIn("--no-embed: 1 commentary chunk(s) lack a current vector -- the release " "will not be committed", out) self.assertEqual(runner.ran("-m", "canlex.embed"), []) self.assertEqual(runner.ran("git", "commit"), []) self.assertIn("compose_release: no release produced", out) def test_a_release_failing_its_predeploy_checks_moves_no_branch(self): runner = _Runner(self.wt, predeploy_rc=1) rc, out, git = self.main([], runner) self.assertEqual(rc, 1) self.assertEqual(len(runner.ran("git", "commit")), 1) # committed in the worktree self.assertIn(f"release commit {RELEASE} FAILED its checks; release-composed not moved", out) git.assert_not_called() self.assertIn("compose_release: no release produced", out) def test_a_release_that_passes_moves_the_branch_and_prints_the_recipe(self): runner = _Runner(self.wt) rc, out, git = self.main(["--allow-held-code"], runner) self.assertEqual(rc, 0, out) check = runner.ran("tools/predeploy.py")[0] self.assertEqual(check[-5:], ["--tree", RELEASE, "--approved", APPROVED, "--allow-held-code"]) self.assertEqual(git.call_args_list[0].args, ("branch", "-f", "release-composed", RELEASE)) self.assertIn("git push space deploy:main", out) self.assertNotIn("no release produced", out) def test_canlex_imported_from_outside_the_worktree_is_refused(self): runner = _Runner(Path(tempfile.gettempdir()) / "elsewhere") with self.assertRaises(RuntimeError) as cm: self.main([], runner) self.assertIn("not the worktree", str(cm.exception)) if __name__ == "__main__": unittest.main()