tests/test_compose_release.py

"""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()