test_print_next_cli.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Integration — ``ok next`` + ``governance-sync --print-next`` (§ONS.12 / §NXP.8).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from pathlib import Path |
| 7 | |
| 8 | from cli.commands.next import EXIT_NEXT_MALFORMED |
| 9 | from cli.kit_root import kit_root |
| 10 | from tests.support import FIXTURES, git_status_runner, run_cli, write_config |
| 11 | from tools.print_next.extract import CURRENT_NEXT_HEADING, set_read_at_clock |
| 12 | |
| 13 | FIXED_READ_AT = "2026-09-04T12:00:00Z" |
| 14 | |
| 15 | |
| 16 | def _seed(tmp_path: Path) -> Path: |
| 17 | write_config(tmp_path, "config-git-only.yaml") |
| 18 | docs = tmp_path / "docs" |
| 19 | docs.mkdir(parents=True, exist_ok=True) |
| 20 | handover = docs / "OVERSEER-HANDOVER.md" |
| 21 | handover.write_text( |
| 22 | (FIXTURES / "print-next-handover.md").read_text(encoding="utf-8"), |
| 23 | encoding="utf-8", |
| 24 | ) |
| 25 | (docs / "ROADMAP.md").write_text("# Roadmap\n", encoding="utf-8") |
| 26 | return handover |
| 27 | |
| 28 | |
| 29 | def _pin_clock(): |
| 30 | set_read_at_clock(lambda: FIXED_READ_AT) |
| 31 | |
| 32 | |
| 33 | def _unpin_clock(): |
| 34 | set_read_at_clock(None) |
| 35 | |
| 36 | |
| 37 | def test_ok_next_exit_zero_heading_model(tmp_path: Path, capsys) -> None: |
| 38 | _seed(tmp_path) |
| 39 | _pin_clock() |
| 40 | try: |
| 41 | runner = git_status_runner() |
| 42 | code = run_cli(["next"], cwd=tmp_path, runner=runner, kit=kit_root()) |
| 43 | out = capsys.readouterr().out |
| 44 | assert code == 0 |
| 45 | assert CURRENT_NEXT_HEADING in out |
| 46 | assert "**Source:**" in out |
| 47 | assert "Model:" in out |
| 48 | joined = " ".join(cmd for cmd, _cwd in runner.calls) |
| 49 | assert "muse" not in joined |
| 50 | assert "git" not in joined |
| 51 | finally: |
| 52 | _unpin_clock() |
| 53 | |
| 54 | |
| 55 | def test_print_next_synonym_same_stdout(tmp_path: Path, capsys) -> None: |
| 56 | _seed(tmp_path) |
| 57 | _pin_clock() |
| 58 | try: |
| 59 | runner = git_status_runner() |
| 60 | code_a = run_cli(["next"], cwd=tmp_path, runner=runner, kit=kit_root()) |
| 61 | out_a = capsys.readouterr().out |
| 62 | code_b = run_cli( |
| 63 | ["governance-sync", "--print-next"], |
| 64 | cwd=tmp_path, |
| 65 | runner=runner, |
| 66 | kit=kit_root(), |
| 67 | ) |
| 68 | out_b = capsys.readouterr().out |
| 69 | assert code_a == 0 |
| 70 | assert code_b == 0 |
| 71 | assert out_a == out_b |
| 72 | assert "**Source:**" in out_a |
| 73 | finally: |
| 74 | _unpin_clock() |
| 75 | |
| 76 | |
| 77 | def test_quiet_includes_provenance(tmp_path: Path, capsys) -> None: |
| 78 | _seed(tmp_path) |
| 79 | _pin_clock() |
| 80 | try: |
| 81 | code = run_cli( |
| 82 | ["--quiet", "next"], |
| 83 | cwd=tmp_path, |
| 84 | runner=git_status_runner(), |
| 85 | kit=kit_root(), |
| 86 | ) |
| 87 | out = capsys.readouterr().out |
| 88 | assert code == 0 |
| 89 | assert CURRENT_NEXT_HEADING in out |
| 90 | assert "**Source:**" in out |
| 91 | assert FIXED_READ_AT in out |
| 92 | finally: |
| 93 | _unpin_clock() |
| 94 | |
| 95 | |
| 96 | def test_json_carries_identity_keys(tmp_path: Path, capsys) -> None: |
| 97 | _seed(tmp_path) |
| 98 | _pin_clock() |
| 99 | try: |
| 100 | code = run_cli( |
| 101 | ["--json", "next"], |
| 102 | cwd=tmp_path, |
| 103 | runner=git_status_runner(), |
| 104 | kit=kit_root(), |
| 105 | json_mode=True, |
| 106 | ) |
| 107 | out = capsys.readouterr().out |
| 108 | assert code == 0 |
| 109 | payload = json.loads(out) |
| 110 | assert payload["ok"] is True |
| 111 | assert payload["path"] |
| 112 | assert "lane" in payload |
| 113 | assert payload["heading"] == CURRENT_NEXT_HEADING |
| 114 | assert payload["fence"] |
| 115 | assert payload["error"] is None |
| 116 | assert payload["repo_name"] == "test-git" |
| 117 | assert payload["repo_root"] == tmp_path.resolve().as_posix() |
| 118 | assert payload["read_at"] == FIXED_READ_AT |
| 119 | finally: |
| 120 | _unpin_clock() |
| 121 | |
| 122 | |
| 123 | def test_lane_flag_two_lane_fixture(tmp_path: Path, capsys) -> None: |
| 124 | write_config(tmp_path, "config-two-lane.yaml") |
| 125 | queue = tmp_path / "QUEUE_HANDOVER.md" |
| 126 | active_dir = tmp_path / "videos" / "_active" |
| 127 | active_dir.mkdir(parents=True, exist_ok=True) |
| 128 | body = ( |
| 129 | "# H\n\n### Paste-ready prompt\n\n```text\n" |
| 130 | "Lane body TOKEN-ACTIVE\nModel: Auto\n```\n" |
| 131 | ) |
| 132 | queue.write_text( |
| 133 | "# Q\n\n### Paste-ready prompt\n\n```text\nQueue body\nModel: Auto\n```\n", |
| 134 | encoding="utf-8", |
| 135 | ) |
| 136 | (tmp_path / "QUEUE_ROADMAP.md").write_text("# R\n", encoding="utf-8") |
| 137 | (active_dir / "HANDOVER.md").write_text(body, encoding="utf-8") |
| 138 | (active_dir / "ROADMAP.md").write_text("# R\n", encoding="utf-8") |
| 139 | _pin_clock() |
| 140 | try: |
| 141 | code = run_cli( |
| 142 | ["next", "--lane", "active"], |
| 143 | cwd=tmp_path, |
| 144 | runner=git_status_runner(), |
| 145 | kit=kit_root(), |
| 146 | ) |
| 147 | out = capsys.readouterr().out |
| 148 | assert code == 0 |
| 149 | assert "TOKEN-ACTIVE" in out |
| 150 | assert "lane `active`" in out |
| 151 | assert "Queue body" not in out |
| 152 | finally: |
| 153 | _unpin_clock() |
| 154 | |
| 155 | |
| 156 | def test_print_next_write_exclusive(tmp_path: Path, capsys) -> None: |
| 157 | _seed(tmp_path) |
| 158 | code = run_cli( |
| 159 | ["governance-sync", "--print-next", "--write"], |
| 160 | cwd=tmp_path, |
| 161 | runner=git_status_runner(), |
| 162 | kit=kit_root(), |
| 163 | ) |
| 164 | err = capsys.readouterr().err |
| 165 | assert code == 2 |
| 166 | assert "print-next mutually exclusive with --write" in err |
| 167 | |
| 168 | |
| 169 | def test_print_next_all_lanes_exclusive(tmp_path: Path, capsys) -> None: |
| 170 | _seed(tmp_path) |
| 171 | code = run_cli( |
| 172 | ["governance-sync", "--print-next", "--all-lanes"], |
| 173 | cwd=tmp_path, |
| 174 | runner=git_status_runner(), |
| 175 | kit=kit_root(), |
| 176 | ) |
| 177 | err = capsys.readouterr().err |
| 178 | assert code == 2 |
| 179 | assert "print-next mutually exclusive with --all-lanes" in err |
| 180 | |
| 181 | |
| 182 | def test_malformed_exit_37(tmp_path: Path) -> None: |
| 183 | write_config(tmp_path, "config-git-only.yaml") |
| 184 | (tmp_path / "docs").mkdir(parents=True, exist_ok=True) |
| 185 | (tmp_path / "docs" / "OVERSEER-HANDOVER.md").write_text("# empty\n", encoding="utf-8") |
| 186 | code = run_cli(["next"], cwd=tmp_path, runner=git_status_runner(), kit=kit_root()) |
| 187 | assert code == EXIT_NEXT_MALFORMED |
| 188 | |
| 189 | |
| 190 | def test_json_failure_shape_carries_identity_keys(tmp_path: Path, capsys) -> None: |
| 191 | """§NXP.4: on ``ok: false``, identity keys are still emitted.""" |
| 192 | write_config(tmp_path, "config-git-only.yaml") |
| 193 | (tmp_path / "docs").mkdir(parents=True, exist_ok=True) |
| 194 | (tmp_path / "docs" / "OVERSEER-HANDOVER.md").write_text("# empty\n", encoding="utf-8") |
| 195 | _pin_clock() |
| 196 | try: |
| 197 | code = run_cli( |
| 198 | ["--json", "next"], |
| 199 | cwd=tmp_path, |
| 200 | runner=git_status_runner(), |
| 201 | kit=kit_root(), |
| 202 | json_mode=True, |
| 203 | ) |
| 204 | payload = json.loads(capsys.readouterr().out) |
| 205 | finally: |
| 206 | _unpin_clock() |
| 207 | |
| 208 | assert code == EXIT_NEXT_MALFORMED |
| 209 | assert payload["ok"] is False |
| 210 | assert payload["error"] == "heading_missing" |
| 211 | # Resolvable → emitted (not null); read_at is always emitted. |
| 212 | assert payload["repo_name"] == "test-git" |
| 213 | assert payload["repo_root"] == tmp_path.resolve().as_posix() |
| 214 | assert payload["read_at"] == FIXED_READ_AT |
| 215 | |
| 216 | |
| 217 | def test_repo_root_unresolved_refuses_rather_than_printing( |
| 218 | tmp_path: Path, capsys, monkeypatch |
| 219 | ) -> None: |
| 220 | """§NXP.3.6: no anonymous block as a degrade path — refuse with exit 2.""" |
| 221 | import cli.commands.next as next_cmd |
| 222 | |
| 223 | _seed(tmp_path) |
| 224 | monkeypatch.setattr(next_cmd, "absolute_repo_root", lambda _root: None) |
| 225 | code = run_cli(["next"], cwd=tmp_path, runner=git_status_runner(), kit=kit_root()) |
| 226 | captured = capsys.readouterr() |
| 227 | |
| 228 | assert code == next_cmd.EXIT_CONFIG == 2 |
| 229 | assert "cannot resolve repo root for provenance" in captured.err |
| 230 | assert "repo_root_unresolved" in captured.err |
| 231 | # The whole point: a block must NOT be printed without provenance. |
| 232 | assert CURRENT_NEXT_HEADING not in captured.out |
| 233 | assert captured.out == "" |
| 234 | |
| 235 | |
| 236 | def test_absolute_repo_root_returns_none_when_unresolvable(monkeypatch) -> None: |
| 237 | """§NXP.3.6 predicate: resolution failure yields ``None``, not a guess.""" |
| 238 | from tools.print_next import extract as extract_mod |
| 239 | |
| 240 | def _boom(self, *args, **kwargs): |
| 241 | raise OSError("cannot resolve") |
| 242 | |
| 243 | monkeypatch.setattr(Path, "resolve", _boom) |
| 244 | assert extract_mod.absolute_repo_root(Path("/some/root")) is None |