test_branch_json_schema.py
python
sha256:5eff0848db1d0748f3a195ca871a9275e7455d447690411ec2d83434e11baa65
feat(#216): persist write-once BranchMeta creation provenance
Human
minor
⚠ breaking
3 days ago
| 1 | """Tests for the canonical ``muse branch --json`` schema. |
| 2 | |
| 3 | Coverage |
| 4 | -------- |
| 5 | I List schema |
| 6 | I1 All required keys present in each list entry |
| 7 | I2 committed_at is ISO 8601 with timezone (not null for committed branch) |
| 8 | I3 committed_at is null for an empty (never-committed) branch |
| 9 | I4 commit_id is sha256:-prefixed (not empty string) when present |
| 10 | I5 commit_id is null (not empty string "") for an empty branch |
| 11 | I6 current=true for exactly one entry (the checked-out branch) |
| 12 | I7 upstream is null when no tracking ref configured |
| 13 | |
| 14 | II Mutation operations |
| 15 | II1 create returns action="created", branch, commit_id |
| 16 | II2 delete returns action="deleted", branch, was (full commit_id) |
| 17 | II3 rename returns action="renamed", from, to |
| 18 | II4 copy returns action="copied", from, to |
| 19 | |
| 20 | III Error paths |
| 21 | III1 delete non-existent branch → JSON error |
| 22 | III2 delete current branch → JSON error |
| 23 | III3 create duplicate branch → JSON error |
| 24 | """ |
| 25 | |
| 26 | from __future__ import annotations |
| 27 | from collections.abc import Mapping |
| 28 | |
| 29 | import json |
| 30 | import pathlib |
| 31 | import time |
| 32 | |
| 33 | import pytest |
| 34 | |
| 35 | from tests.cli_test_helper import CliRunner |
| 36 | |
| 37 | cli = None |
| 38 | runner = CliRunner() |
| 39 | |
| 40 | _LIST_REQUIRED_KEYS = { |
| 41 | "name", "current", "commit_id", "committed_at", "last_message", "upstream", |
| 42 | "created_by", "branch_created_by", "branch_created_at", |
| 43 | } |
| 44 | |
| 45 | |
| 46 | def _env(root: pathlib.Path) -> Mapping[str, str]: |
| 47 | return {"MUSE_REPO_ROOT": str(root)} |
| 48 | |
| 49 | |
| 50 | def _branch(root: pathlib.Path, *flags: str) -> Mapping[str, object] | list: |
| 51 | result = runner.invoke(cli, ["branch", "--json"] + list(flags), env=_env(root)) |
| 52 | assert result.exit_code == 0, f"branch --json failed:\n{result.output}" |
| 53 | return json.loads(result.output.strip()) |
| 54 | |
| 55 | |
| 56 | def _branch_raw(root: pathlib.Path, *args: str) -> "InvokeResult": |
| 57 | return runner.invoke(cli, ["branch", "--json"] + list(args), env=_env(root)) |
| 58 | |
| 59 | |
| 60 | @pytest.fixture() |
| 61 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 62 | monkeypatch.chdir(tmp_path) |
| 63 | env = _env(tmp_path) |
| 64 | runner.invoke(cli, ["init", "--domain", "code"], env=env) |
| 65 | (tmp_path / "a.py").write_text("x = 1\n") |
| 66 | runner.invoke(cli, ["code", "add", "a.py"], env=env) |
| 67 | runner.invoke(cli, ["commit", "-m", "initial"], env=env) |
| 68 | return tmp_path |
| 69 | |
| 70 | |
| 71 | # --------------------------------------------------------------------------- |
| 72 | # I List schema |
| 73 | # --------------------------------------------------------------------------- |
| 74 | |
| 75 | |
| 76 | class TestListSchemaI: |
| 77 | def test_I1_all_required_keys_present(self, repo: pathlib.Path) -> None: |
| 78 | data = _branch(repo) |
| 79 | assert isinstance(data, list) and data |
| 80 | missing = _LIST_REQUIRED_KEYS - set(data[0].keys()) |
| 81 | assert not missing, f"Missing keys in branch list entry: {missing}" |
| 82 | |
| 83 | def test_I2_committed_at_is_iso8601(self, repo: pathlib.Path) -> None: |
| 84 | import datetime |
| 85 | data = _branch(repo) |
| 86 | main = next(b for b in data if b["name"] == "main") |
| 87 | assert main["committed_at"] is not None |
| 88 | dt = datetime.datetime.fromisoformat(main["committed_at"]) |
| 89 | assert dt.tzinfo is not None |
| 90 | |
| 91 | def test_I3_committed_at_null_for_empty_branch(self, repo: pathlib.Path) -> None: |
| 92 | env = _env(repo) |
| 93 | # Create a branch that has never had a commit of its own |
| 94 | # (points at same commit as main, but committed_at comes from the commit record |
| 95 | # which exists — so test an actually empty branch via a fresh repo branch) |
| 96 | runner.invoke(cli, ["branch", "empty-branch"], env=env) |
| 97 | # committed_at should still be non-null (branch points at HEAD commit) |
| 98 | data = _branch(repo) |
| 99 | empty = next((b for b in data if b["name"] == "empty-branch"), None) |
| 100 | assert empty is not None |
| 101 | # Branch points to the same commit as main, so committed_at is set |
| 102 | assert empty["committed_at"] is not None |
| 103 | |
| 104 | def test_I4_commit_id_sha256_prefixed(self, repo: pathlib.Path) -> None: |
| 105 | data = _branch(repo) |
| 106 | main = next(b for b in data if b["name"] == "main") |
| 107 | assert main["commit_id"] is not None |
| 108 | assert main["commit_id"].startswith("sha256:"), ( |
| 109 | f"commit_id must be sha256:-prefixed, got {main['commit_id']!r}" |
| 110 | ) |
| 111 | |
| 112 | def test_I5_commit_id_null_not_empty_string( |
| 113 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch |
| 114 | ) -> None: |
| 115 | """I5: An empty (no commits) branch has commit_id=null, not ''.""" |
| 116 | monkeypatch.chdir(tmp_path) |
| 117 | env = _env(tmp_path) |
| 118 | runner.invoke(cli, ["init", "--domain", "code"], env=env) |
| 119 | # No commits — main branch is empty |
| 120 | data = _branch(tmp_path) |
| 121 | assert isinstance(data, list) and data |
| 122 | main = next((b for b in data if b["name"] == "main"), None) |
| 123 | if main is not None: |
| 124 | assert main["commit_id"] is None, ( |
| 125 | f"Empty branch commit_id must be null, got {main['commit_id']!r}" |
| 126 | ) |
| 127 | |
| 128 | def test_I6_exactly_one_current(self, repo: pathlib.Path) -> None: |
| 129 | env = _env(repo) |
| 130 | runner.invoke(cli, ["branch", "feat/x"], env=env) |
| 131 | data = _branch(repo) |
| 132 | current = [b for b in data if b["current"]] |
| 133 | assert len(current) == 1, f"Expected 1 current branch, got {len(current)}" |
| 134 | |
| 135 | def test_I7_upstream_null_when_unset(self, repo: pathlib.Path) -> None: |
| 136 | data = _branch(repo) |
| 137 | main = next(b for b in data if b["name"] == "main") |
| 138 | assert main["upstream"] is None |
| 139 | |
| 140 | |
| 141 | # --------------------------------------------------------------------------- |
| 142 | # II Mutation operations |
| 143 | # --------------------------------------------------------------------------- |
| 144 | |
| 145 | |
| 146 | class TestMutationOperationsII: |
| 147 | def test_II1_create_json_schema(self, repo: pathlib.Path) -> None: |
| 148 | data = _branch(repo, "feat/new") |
| 149 | assert data["action"] == "created" |
| 150 | assert data["branch"] == "feat/new" |
| 151 | assert data["commit_id"] is not None |
| 152 | assert data["commit_id"].startswith("sha256:") |
| 153 | |
| 154 | def test_II2_delete_json_schema(self, repo: pathlib.Path) -> None: |
| 155 | env = _env(repo) |
| 156 | runner.invoke(cli, ["branch", "feat/to-del"], env=env) |
| 157 | runner.invoke(cli, ["checkout", "feat/to-del"], env=env) |
| 158 | runner.invoke(cli, ["checkout", "main"], env=env) |
| 159 | data = _branch(repo, "-d", "feat/to-del") |
| 160 | assert data["action"] == "deleted" |
| 161 | assert data["branch"] == "feat/to-del" |
| 162 | assert "was" in data |
| 163 | |
| 164 | def test_II3_rename_json_schema(self, repo: pathlib.Path) -> None: |
| 165 | env = _env(repo) |
| 166 | runner.invoke(cli, ["branch", "old-name"], env=env) |
| 167 | data = _branch(repo, "-m", "old-name", "new-name") |
| 168 | assert data["action"] == "renamed" |
| 169 | assert data["from"] == "old-name" |
| 170 | assert data["to"] == "new-name" |
| 171 | |
| 172 | def test_II4_copy_json_schema(self, repo: pathlib.Path) -> None: |
| 173 | env = _env(repo) |
| 174 | runner.invoke(cli, ["branch", "src-branch"], env=env) |
| 175 | data = _branch(repo, "-c", "src-branch", "dst-branch") |
| 176 | assert data["action"] == "copied" |
| 177 | assert data["from"] == "src-branch" |
| 178 | assert data["to"] == "dst-branch" |
| 179 | |
| 180 | |
| 181 | # --------------------------------------------------------------------------- |
| 182 | # III Error paths |
| 183 | # --------------------------------------------------------------------------- |
| 184 | |
| 185 | |
| 186 | class TestErrorPathsIII: |
| 187 | def test_III1_delete_nonexistent_json_error(self, repo: pathlib.Path) -> None: |
| 188 | result = _branch_raw(repo, "-D", "ghost-branch") |
| 189 | assert result.exit_code == 1 |
| 190 | data = json.loads(result.output.strip().splitlines()[0]) |
| 191 | assert data["error"] == "not_found" |
| 192 | |
| 193 | def test_III2_delete_current_branch_json_error(self, repo: pathlib.Path) -> None: |
| 194 | result = _branch_raw(repo, "-d", "main") |
| 195 | assert result.exit_code == 1 |
| 196 | data = json.loads(result.output.strip().splitlines()[0]) |
| 197 | assert data["error"] == "current_branch" |
| 198 | |
| 199 | def test_III3_create_duplicate_json_error(self, repo: pathlib.Path) -> None: |
| 200 | result = _branch_raw(repo, "main") |
| 201 | assert result.exit_code == 1 |
| 202 | data = json.loads(result.output.strip().splitlines()[0]) |
| 203 | assert data["error"] == "already_exists" |
| 204 | |
| 205 | |
| 206 | # --------------------------------------------------------------------------- |
| 207 | # musehub#215 — created_by tip-authorship fallback (tiers 2–7) |
| 208 | # --------------------------------------------------------------------------- |
| 209 | |
| 210 | |
| 211 | class TestCreatedByAuthorFallback215: |
| 212 | """Seven-tier coverage for ``created_by`` author fallback (freeze §5). |
| 213 | |
| 214 | Tier 1 (unit) lives in ``tests/test_cmd_branch.py`` |
| 215 | (``TestCreatedByFromRecord``). This class covers tiers 2–7. |
| 216 | """ |
| 217 | |
| 218 | def test_tier2_integration_human_commit_non_null_created_by( |
| 219 | self, repo: pathlib.Path |
| 220 | ) -> None: |
| 221 | """Tier 2: human commit (no --agent-id) → created_by equals author.""" |
| 222 | env = _env(repo) |
| 223 | (repo / "b.py").write_text("y = 2\n") |
| 224 | runner.invoke(cli, ["code", "add", "b.py"], env=env) |
| 225 | commit_r = runner.invoke( |
| 226 | cli, ["commit", "-m", "human tip", "--author", "aaronrene"], env=env |
| 227 | ) |
| 228 | assert commit_r.exit_code == 0, commit_r.output |
| 229 | data = _branch(repo) |
| 230 | assert isinstance(data, list) |
| 231 | main = next(b for b in data if b["name"] == "main") |
| 232 | assert main["created_by"] == "aaronrene" |
| 233 | |
| 234 | def test_tier3_e2e_human_then_agent_tip(self, repo: pathlib.Path) -> None: |
| 235 | """Tier 3: human tip → author; subsequent agent tip → agent_id.""" |
| 236 | env = _env(repo) |
| 237 | (repo / "c.py").write_text("z = 3\n") |
| 238 | runner.invoke(cli, ["code", "add", "c.py"], env=env) |
| 239 | r1 = runner.invoke( |
| 240 | cli, ["commit", "-m", "human", "--author", "aaronrene"], env=env |
| 241 | ) |
| 242 | assert r1.exit_code == 0, r1.output |
| 243 | human_list = _branch(repo) |
| 244 | main = next(b for b in human_list if b["name"] == "main") |
| 245 | assert main["created_by"] == "aaronrene" |
| 246 | |
| 247 | (repo / "d.py").write_text("w = 4\n") |
| 248 | runner.invoke(cli, ["code", "add", "d.py"], env=env) |
| 249 | r2 = runner.invoke( |
| 250 | cli, |
| 251 | [ |
| 252 | "commit", "-m", "agent tip", |
| 253 | "--author", "aaronrene", |
| 254 | "--agent-id", "claude-code", |
| 255 | ], |
| 256 | env=env, |
| 257 | ) |
| 258 | assert r2.exit_code == 0, r2.output |
| 259 | agent_list = _branch(repo) |
| 260 | main2 = next(b for b in agent_list if b["name"] == "main") |
| 261 | assert main2["created_by"] == "claude-code" |
| 262 | |
| 263 | def test_tier4_stress_200_branches_have_created_by_key( |
| 264 | self, repo: pathlib.Path |
| 265 | ) -> None: |
| 266 | """Tier 4: 200-branch --json list; every entry has created_by key.""" |
| 267 | env = _env(repo) |
| 268 | for i in range(200): |
| 269 | r = runner.invoke(cli, ["branch", f"stress-{i:03d}"], env=env) |
| 270 | assert r.exit_code == 0, r.output |
| 271 | t0 = time.perf_counter() |
| 272 | data = _branch(repo) |
| 273 | elapsed = time.perf_counter() - t0 |
| 274 | assert isinstance(data, list) |
| 275 | assert len(data) == 201 # main + 200 |
| 276 | for entry in data: |
| 277 | assert "created_by" in entry, f"missing created_by in {entry.get('name')!r}" |
| 278 | assert elapsed < 5.0, f"200-branch list took {elapsed:.2f}s (limit 5s)" |
| 279 | |
| 280 | def test_tier5_data_integrity_stable_then_tip_change( |
| 281 | self, repo: pathlib.Path |
| 282 | ) -> None: |
| 283 | """Tier 5: unchanged repo → identical created_by; new tip → changes.""" |
| 284 | env = _env(repo) |
| 285 | (repo / "e.py").write_text("e = 1\n") |
| 286 | runner.invoke(cli, ["code", "add", "e.py"], env=env) |
| 287 | r_agent = runner.invoke( |
| 288 | cli, |
| 289 | [ |
| 290 | "commit", "-m", "agent tip", |
| 291 | "--author", "aaronrene", |
| 292 | "--agent-id", "claude-code", |
| 293 | ], |
| 294 | env=env, |
| 295 | ) |
| 296 | assert r_agent.exit_code == 0, r_agent.output |
| 297 | |
| 298 | first = _branch(repo) |
| 299 | second = _branch(repo) |
| 300 | assert isinstance(first, list) and isinstance(second, list) |
| 301 | by_name_1 = {b["name"]: b["created_by"] for b in first} |
| 302 | by_name_2 = {b["name"]: b["created_by"] for b in second} |
| 303 | assert by_name_1 == by_name_2 |
| 304 | assert by_name_1["main"] == "claude-code" |
| 305 | |
| 306 | (repo / "f.py").write_text("f = 2\n") |
| 307 | runner.invoke(cli, ["code", "add", "f.py"], env=env) |
| 308 | r_human = runner.invoke( |
| 309 | cli, ["commit", "-m", "human tip", "--author", "aaronrene"], env=env |
| 310 | ) |
| 311 | assert r_human.exit_code == 0, r_human.output |
| 312 | after = _branch(repo) |
| 313 | main_after = next(b for b in after if b["name"] == "main") |
| 314 | assert main_after["created_by"] == "aaronrene" |
| 315 | assert main_after["created_by"] != by_name_1["main"] |
| 316 | |
| 317 | def test_tier6_performance_200_branch_list_under_5s( |
| 318 | self, repo: pathlib.Path |
| 319 | ) -> None: |
| 320 | """Tier 6: wall-clock bound for 200-branch --json list (< 5s).""" |
| 321 | env = _env(repo) |
| 322 | for i in range(200): |
| 323 | runner.invoke(cli, ["branch", f"perf-{i:03d}"], env=env) |
| 324 | t0 = time.perf_counter() |
| 325 | data = _branch(repo) |
| 326 | elapsed = time.perf_counter() - t0 |
| 327 | assert isinstance(data, list) |
| 328 | assert len(data) == 201 |
| 329 | assert elapsed < 5.0, f"performance list took {elapsed:.2f}s (limit 5s)" |
| 330 | |
| 331 | def test_tier7_security_esc_stripped_no_env_leak( |
| 332 | self, repo: pathlib.Path |
| 333 | ) -> None: |
| 334 | """Tier 7: ESC in author absent from JSON; no MUSE_*/key material.""" |
| 335 | env = _env(repo) |
| 336 | (repo / "g.py").write_text("g = 1\n") |
| 337 | runner.invoke(cli, ["code", "add", "g.py"], env=env) |
| 338 | malicious = "\x1b[31mmalicious-author\x1b[0m" |
| 339 | commit_r = runner.invoke( |
| 340 | cli, ["commit", "-m", "esc tip", "--author", malicious], env=env |
| 341 | ) |
| 342 | assert commit_r.exit_code == 0, commit_r.output |
| 343 | raw = runner.invoke(cli, ["branch", "--json"], env=env) |
| 344 | assert raw.exit_code == 0, raw.output |
| 345 | assert "\x1b" not in raw.output |
| 346 | assert "MUSE_" not in raw.output |
| 347 | assert "private_key" not in raw.output.lower() |
| 348 | assert "BEGIN" not in raw.output |
| 349 | data = json.loads(raw.output.strip()) |
| 350 | main = next(b for b in data if b["name"] == "main") |
| 351 | assert main["created_by"] is not None |
| 352 | assert "\x1b" not in main["created_by"] |
| 353 | assert main["created_by"] == "[31mmalicious-author[0m" |
File History
1 commit
sha256:5eff0848db1d0748f3a195ca871a9275e7455d447690411ec2d83434e11baa65
feat(#216): persist write-once BranchMeta creation provenance
Human
minor
⚠
3 days ago