File size: 10,039 Bytes
901bf8c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | """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()
|