test_bridge_phase1.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Phase 1 TDD tests for ``muse bridge`` — namespace structure and bridge state. |
| 2 | |
| 3 | Tests are organised into seven tiers: |
| 4 | |
| 5 | Tier 1 — Unit BridgeState TypedDict, read_bridge_state, write_bridge_state |
| 6 | Tier 2 — Contract CLI namespace: --help shows all subcommands and required flags |
| 7 | Tier 3 — Integration CliRunner invocation of each subcommand |
| 8 | Tier 4 — Property Bridge state round-trips for arbitrary valid state dicts |
| 9 | Tier 5 — Regression git-bridge.toml backward-compat: file with partial keys is readable |
| 10 | Tier 6 — Security Path traversal in bridge state location rejected / contained |
| 11 | Tier 7 — Stress Concurrent reads and writes to bridge state do not corrupt data |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import pathlib |
| 17 | import threading |
| 18 | import tomllib |
| 19 | from unittest.mock import patch |
| 20 | |
| 21 | import pytest |
| 22 | from hypothesis import HealthCheck, given, settings |
| 23 | from hypothesis import strategies as st |
| 24 | |
| 25 | from muse.core.bridge.state import BridgeState |
| 26 | from muse.core.paths import git_bridge_state_path, init_repo_dirs |
| 27 | from muse.core.types import fake_id |
| 28 | from tests.cli_test_helper import CliRunner |
| 29 | |
| 30 | runner = CliRunner() |
| 31 | |
| 32 | |
| 33 | # --------------------------------------------------------------------------- |
| 34 | # Helpers |
| 35 | # --------------------------------------------------------------------------- |
| 36 | |
| 37 | def _invoke(*args: str) -> "CliRunner": |
| 38 | return runner.invoke(None, list(args)) |
| 39 | |
| 40 | |
| 41 | def _read_bs(root: pathlib.Path) -> BridgeState: |
| 42 | """Import lazily so tests can be collected before bridge.py exists.""" |
| 43 | from muse.core.bridge.state import read_bridge_state |
| 44 | return read_bridge_state(root) |
| 45 | |
| 46 | |
| 47 | def _write_bs(root: pathlib.Path, state: BridgeState) -> None: |
| 48 | from muse.core.bridge.state import write_bridge_state |
| 49 | write_bridge_state(root, state) |
| 50 | |
| 51 | |
| 52 | def _fake_muse_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 53 | """Create a minimal .muse/ layout so find_repo_root() succeeds.""" |
| 54 | return init_repo_dirs(tmp_path) |
| 55 | |
| 56 | |
| 57 | # =========================================================================== |
| 58 | # Tier 1 — Unit: TypedDict, read_bridge_state, write_bridge_state |
| 59 | # =========================================================================== |
| 60 | |
| 61 | class TestBridgeStateTypedDict: |
| 62 | """BridgeState TypedDict has the right structure.""" |
| 63 | |
| 64 | def test_import_and_instantiate(self) -> None: |
| 65 | """BridgeState can be imported and used as a TypedDict.""" |
| 66 | from muse.core.bridge.state import BridgeState |
| 67 | state: BridgeState = { |
| 68 | "last_import": { |
| 69 | "git_sha": "a" * 40, |
| 70 | "git_ref": "main", |
| 71 | "git_remote": "origin", |
| 72 | "muse_branch": "main", |
| 73 | "muse_commit_id": fake_id("commit-1"), |
| 74 | "imported_at": "2026-01-01T00:00:00Z", |
| 75 | "commits_written": 3, |
| 76 | }, |
| 77 | "last_export": { |
| 78 | "muse_branch": "main", |
| 79 | "muse_commit_id": fake_id("commit-2"), |
| 80 | "git_remote": "origin", |
| 81 | "git_ref": "muse-mirror", |
| 82 | "git_sha": "b" * 40, |
| 83 | "exported_at": "2026-01-01T01:00:00Z", |
| 84 | }, |
| 85 | } |
| 86 | assert state["last_import"]["commits_written"] == 3 |
| 87 | assert state["last_export"]["git_remote"] == "origin" |
| 88 | |
| 89 | def test_muse_commit_id_is_prefixed(self) -> None: |
| 90 | """muse_commit_id values always carry sha256: prefix.""" |
| 91 | from muse.core.bridge.state import BridgeState |
| 92 | cid = fake_id("test-commit") |
| 93 | assert cid.startswith("sha256:") |
| 94 | state: BridgeState = { |
| 95 | "last_import": { |
| 96 | "git_sha": "c" * 40, |
| 97 | "git_ref": "main", |
| 98 | "git_remote": "origin", |
| 99 | "muse_branch": "main", |
| 100 | "muse_commit_id": cid, |
| 101 | "imported_at": "2026-01-01T00:00:00Z", |
| 102 | "commits_written": 1, |
| 103 | }, |
| 104 | "last_export": {}, |
| 105 | } |
| 106 | assert state["last_import"]["muse_commit_id"].startswith("sha256:") |
| 107 | |
| 108 | |
| 109 | class TestReadBridgeState: |
| 110 | """read_bridge_state returns a default empty state when file is missing.""" |
| 111 | |
| 112 | def test_missing_file_returns_empty_state(self, tmp_path: pathlib.Path) -> None: |
| 113 | root = _fake_muse_repo(tmp_path) |
| 114 | state = _read_bs(root) |
| 115 | assert "last_import" in state |
| 116 | assert "last_export" in state |
| 117 | assert state["last_import"] == {} |
| 118 | assert state["last_export"] == {} |
| 119 | |
| 120 | def test_existing_file_is_parsed(self, tmp_path: pathlib.Path) -> None: |
| 121 | root = _fake_muse_repo(tmp_path) |
| 122 | cid = fake_id("my-commit") |
| 123 | toml_text = f""" |
| 124 | [last_import] |
| 125 | git_sha = "{'a' * 40}" |
| 126 | git_ref = "main" |
| 127 | git_remote = "origin" |
| 128 | muse_branch = "main" |
| 129 | muse_commit_id = "{cid}" |
| 130 | imported_at = "2026-01-15T10:00:00Z" |
| 131 | commits_written = 7 |
| 132 | """ |
| 133 | (git_bridge_state_path(root)).write_text(toml_text) |
| 134 | state = _read_bs(root) |
| 135 | assert state["last_import"]["commits_written"] == 7 |
| 136 | assert state["last_import"]["muse_commit_id"] == cid |
| 137 | assert state["last_import"]["git_ref"] == "main" |
| 138 | |
| 139 | def test_partial_file_fills_missing_sections(self, tmp_path: pathlib.Path) -> None: |
| 140 | """File with only [last_import] — last_export defaults to empty dict.""" |
| 141 | root = _fake_muse_repo(tmp_path) |
| 142 | (git_bridge_state_path(root)).write_text("[last_import]\ngit_ref = \"dev\"\n") |
| 143 | state = _read_bs(root) |
| 144 | assert state["last_import"]["git_ref"] == "dev" |
| 145 | assert state["last_export"] == {} |
| 146 | |
| 147 | |
| 148 | class TestWriteBridgeState: |
| 149 | """write_bridge_state persists state to .muse/git-bridge.toml.""" |
| 150 | |
| 151 | def test_writes_toml_file(self, tmp_path: pathlib.Path) -> None: |
| 152 | root = _fake_muse_repo(tmp_path) |
| 153 | cid = fake_id("written-commit") |
| 154 | state = { |
| 155 | "last_import": { |
| 156 | "git_sha": "d" * 40, |
| 157 | "git_ref": "main", |
| 158 | "git_remote": "origin", |
| 159 | "muse_branch": "main", |
| 160 | "muse_commit_id": cid, |
| 161 | "imported_at": "2026-02-01T00:00:00Z", |
| 162 | "commits_written": 5, |
| 163 | }, |
| 164 | "last_export": {}, |
| 165 | } |
| 166 | _write_bs(root, state) |
| 167 | toml_path = git_bridge_state_path(root) |
| 168 | assert toml_path.exists() |
| 169 | parsed = tomllib.loads(toml_path.read_text()) |
| 170 | assert parsed["last_import"]["muse_commit_id"] == cid |
| 171 | assert parsed["last_import"]["commits_written"] == 5 |
| 172 | |
| 173 | def test_muse_commit_id_preserves_prefix(self, tmp_path: pathlib.Path) -> None: |
| 174 | """sha256: prefix is never stripped during write.""" |
| 175 | root = _fake_muse_repo(tmp_path) |
| 176 | cid = fake_id("prefix-check") |
| 177 | assert cid.startswith("sha256:") |
| 178 | _write_bs(root, {"last_import": {"muse_commit_id": cid}, "last_export": {}}) |
| 179 | toml_path = git_bridge_state_path(root) |
| 180 | parsed = tomllib.loads(toml_path.read_text()) |
| 181 | assert parsed["last_import"]["muse_commit_id"].startswith("sha256:") |
| 182 | |
| 183 | def test_round_trip(self, tmp_path: pathlib.Path) -> None: |
| 184 | """write then read gives back the same state.""" |
| 185 | root = _fake_muse_repo(tmp_path) |
| 186 | original: BridgeState = { |
| 187 | "last_import": {"git_sha": "e" * 40, "git_ref": "dev", "commits_written": 12}, |
| 188 | "last_export": {"muse_commit_id": fake_id("rt"), "git_sha": "f" * 40}, |
| 189 | } |
| 190 | _write_bs(root, original) |
| 191 | recovered = _read_bs(root) |
| 192 | assert recovered["last_import"]["git_sha"] == "e" * 40 |
| 193 | assert recovered["last_import"]["commits_written"] == 12 |
| 194 | assert recovered["last_export"]["git_sha"] == "f" * 40 |
| 195 | |
| 196 | |
| 197 | # =========================================================================== |
| 198 | # Tier 2 — Contract: CLI namespace shows all subcommands and required flags |
| 199 | # =========================================================================== |
| 200 | |
| 201 | class TestBridgeCliContract: |
| 202 | """muse bridge --help shows git-import, git-export, git-status.""" |
| 203 | |
| 204 | def test_bridge_help_shows_git_import(self) -> None: |
| 205 | result = _invoke("bridge", "--help") |
| 206 | assert "git-import" in result.output or "git-import" in result.stderr |
| 207 | |
| 208 | def test_bridge_help_shows_git_export(self) -> None: |
| 209 | result = _invoke("bridge", "--help") |
| 210 | assert "git-export" in result.output or "git-export" in result.stderr |
| 211 | |
| 212 | def test_bridge_help_shows_git_status(self) -> None: |
| 213 | result = _invoke("bridge", "--help") |
| 214 | assert "git-status" in result.output or "git-status" in result.stderr |
| 215 | |
| 216 | |
| 217 | class TestGitImportFlags: |
| 218 | """muse bridge git-import --help exposes all required flags.""" |
| 219 | |
| 220 | @pytest.mark.parametrize("flag", [ |
| 221 | "--target", "--branch", "--all", "--from-ref", "--incremental", |
| 222 | "--attribution-map", "--sign", "--dry-run", "--json", |
| 223 | ]) |
| 224 | def test_flag_present(self, flag: str) -> None: |
| 225 | result = _invoke("bridge", "git-import", "--help") |
| 226 | combined = result.output + result.stderr |
| 227 | assert flag in combined, f"Flag {flag!r} missing from git-import --help" |
| 228 | |
| 229 | |
| 230 | class TestGitExportFlags: |
| 231 | """muse bridge git-export --help exposes all required flags.""" |
| 232 | |
| 233 | @pytest.mark.parametrize("flag", [ |
| 234 | "--muse-ref", "--git-dir", "--git-branch", "--git-remote", |
| 235 | "--no-push", "--dry-run", "--json", |
| 236 | ]) |
| 237 | def test_flag_present(self, flag: str) -> None: |
| 238 | result = _invoke("bridge", "git-export", "--help") |
| 239 | combined = result.output + result.stderr |
| 240 | assert flag in combined, f"Flag {flag!r} missing from git-export --help" |
| 241 | |
| 242 | |
| 243 | class TestGitStatusFlags: |
| 244 | """muse bridge git-status --help exposes all required flags.""" |
| 245 | |
| 246 | @pytest.mark.parametrize("flag", ["--git-dir", "--json"]) |
| 247 | def test_flag_present(self, flag: str) -> None: |
| 248 | result = _invoke("bridge", "git-status", "--help") |
| 249 | combined = result.output + result.stderr |
| 250 | assert flag in combined, f"Flag {flag!r} missing from git-status --help" |
| 251 | |
| 252 | |
| 253 | # =========================================================================== |
| 254 | # Tier 3 — Integration: CliRunner invocation |
| 255 | # =========================================================================== |
| 256 | |
| 257 | class TestBridgeIntegration: |
| 258 | """Basic invocation through CliRunner exits cleanly or with known codes.""" |
| 259 | |
| 260 | def test_git_import_dry_run_no_source(self) -> None: |
| 261 | """git-import --dry-run with no SOURCE falls back to cwd gracefully.""" |
| 262 | result = _invoke("bridge", "git-import", "--dry-run") |
| 263 | # Should exit with user error (no valid git repo at cwd), not a crash |
| 264 | assert result.exit_code in (0, 1, 2) |
| 265 | |
| 266 | def test_git_export_no_git_dir_exits_user_error(self) -> None: |
| 267 | """git-export without --git-dir should exit with error, not crash.""" |
| 268 | with patch("muse.core.repo.find_repo_root", return_value=None): |
| 269 | result = _invoke("bridge", "git-export", "--git-dir", "/nonexistent/path") |
| 270 | assert result.exit_code in (1, 2) |
| 271 | |
| 272 | def test_git_status_no_repo_exits_cleanly(self) -> None: |
| 273 | """git-status with no Muse repo exits with user error, not traceback.""" |
| 274 | with patch("muse.core.repo.find_repo_root", return_value=None): |
| 275 | result = _invoke("bridge", "git-status") |
| 276 | assert result.exit_code in (0, 1, 2) |
| 277 | |
| 278 | |
| 279 | # =========================================================================== |
| 280 | # Tier 4 — Property: bridge state round-trips |
| 281 | # =========================================================================== |
| 282 | |
| 283 | _SAFE_TEXT = st.text( |
| 284 | alphabet=st.characters(whitelist_categories=("L", "N"), whitelist_characters="-_/:"), |
| 285 | min_size=0, |
| 286 | max_size=50, |
| 287 | ) |
| 288 | |
| 289 | class TestBridgeStateProperty: |
| 290 | """Property: bridge state survives a write/read round-trip.""" |
| 291 | |
| 292 | @given( |
| 293 | git_ref=_SAFE_TEXT, |
| 294 | commits_written=st.integers(min_value=0, max_value=10_000), |
| 295 | ) |
| 296 | @settings(max_examples=30, suppress_health_check=[HealthCheck.function_scoped_fixture]) |
| 297 | def test_import_state_round_trips( |
| 298 | self, tmp_path: pathlib.Path, git_ref: str, commits_written: int |
| 299 | ) -> None: |
| 300 | root = _fake_muse_repo(tmp_path) |
| 301 | cid = fake_id(f"prop-{git_ref[:10]}-{commits_written}") |
| 302 | state = { |
| 303 | "last_import": { |
| 304 | "git_sha": "a" * 40, |
| 305 | "git_ref": git_ref, |
| 306 | "commits_written": commits_written, |
| 307 | "muse_commit_id": cid, |
| 308 | }, |
| 309 | "last_export": {}, |
| 310 | } |
| 311 | _write_bs(root, state) |
| 312 | recovered = _read_bs(root) |
| 313 | assert recovered["last_import"]["commits_written"] == commits_written |
| 314 | assert recovered["last_import"]["muse_commit_id"] == cid |
| 315 | |
| 316 | |
| 317 | # =========================================================================== |
| 318 | # Tier 5 — Regression: partial / legacy state files are tolerated |
| 319 | # =========================================================================== |
| 320 | |
| 321 | class TestBridgeStateRegression: |
| 322 | """Previously, partial state files would crash. Now they must be tolerated.""" |
| 323 | |
| 324 | def test_empty_toml_file_returns_empty_state(self, tmp_path: pathlib.Path) -> None: |
| 325 | root = _fake_muse_repo(tmp_path) |
| 326 | (git_bridge_state_path(root)).write_text("") |
| 327 | state = _read_bs(root) |
| 328 | assert state["last_import"] == {} |
| 329 | assert state["last_export"] == {} |
| 330 | |
| 331 | def test_no_last_import_section(self, tmp_path: pathlib.Path) -> None: |
| 332 | root = _fake_muse_repo(tmp_path) |
| 333 | (git_bridge_state_path(root)).write_text("[last_export]\ngit_ref = \"muse-mirror\"\n") |
| 334 | state = _read_bs(root) |
| 335 | assert state["last_import"] == {} |
| 336 | assert state["last_export"]["git_ref"] == "muse-mirror" |
| 337 | |
| 338 | def test_extra_unknown_keys_are_preserved(self, tmp_path: pathlib.Path) -> None: |
| 339 | """Unknown TOML keys in bridge state are passed through, not rejected.""" |
| 340 | root = _fake_muse_repo(tmp_path) |
| 341 | (git_bridge_state_path(root)).write_text( |
| 342 | "[last_import]\nfuture_key = \"v2\"\n" |
| 343 | ) |
| 344 | state = _read_bs(root) |
| 345 | assert state["last_import"].get("future_key") == "v2" |
| 346 | |
| 347 | |
| 348 | # =========================================================================== |
| 349 | # Tier 6 — Security: path traversal in bridge state |
| 350 | # =========================================================================== |
| 351 | |
| 352 | class TestBridgeStateSecurity: |
| 353 | """Bridge state is always stored inside .muse/ — path traversal is rejected.""" |
| 354 | |
| 355 | def test_read_state_is_always_inside_muse(self, tmp_path: pathlib.Path) -> None: |
| 356 | """read_bridge_state always reads from <root>/.muse/git-bridge.toml.""" |
| 357 | root = _fake_muse_repo(tmp_path) |
| 358 | from muse.core.bridge.state import read_bridge_state |
| 359 | import inspect |
| 360 | src = inspect.getsource(read_bridge_state) |
| 361 | # Must reference .muse/ in the path, not accept arbitrary paths from env |
| 362 | assert ".muse" in src or "git-bridge" in src |
| 363 | |
| 364 | def test_write_state_stays_inside_muse(self, tmp_path: pathlib.Path) -> None: |
| 365 | """write_bridge_state only writes to <root>/.muse/git-bridge.toml.""" |
| 366 | root = _fake_muse_repo(tmp_path) |
| 367 | _write_bs(root, {"last_import": {}, "last_export": {}}) |
| 368 | written = list(tmp_path.rglob("git-bridge.toml")) |
| 369 | assert len(written) == 1 |
| 370 | assert ".muse" in str(written[0]) |
| 371 | |
| 372 | def test_muse_commit_id_without_prefix_raises(self, tmp_path: pathlib.Path) -> None: |
| 373 | """write_bridge_state rejects muse_commit_id without sha256: prefix.""" |
| 374 | root = _fake_muse_repo(tmp_path) |
| 375 | bare_hex = "a" * 64 # no sha256: prefix |
| 376 | with pytest.raises((ValueError, SystemExit)): |
| 377 | _write_bs(root, { |
| 378 | "last_import": {"muse_commit_id": bare_hex}, |
| 379 | "last_export": {}, |
| 380 | }) |
| 381 | |
| 382 | |
| 383 | # =========================================================================== |
| 384 | # Tier 7 — Stress: concurrent reads and writes |
| 385 | # =========================================================================== |
| 386 | |
| 387 | class TestBridgeStateStress: |
| 388 | """Concurrent reads and writes to the bridge state file do not corrupt data.""" |
| 389 | |
| 390 | def test_concurrent_writes_do_not_corrupt(self, tmp_path: pathlib.Path) -> None: |
| 391 | root = _fake_muse_repo(tmp_path) |
| 392 | errors: list[Exception] = [] |
| 393 | |
| 394 | def _worker(i: int) -> None: |
| 395 | try: |
| 396 | _write_bs(root, { |
| 397 | "last_import": { |
| 398 | "git_sha": hex(i)[2:].zfill(40)[:40], |
| 399 | "commits_written": i, |
| 400 | "muse_commit_id": fake_id(f"stress-{i}"), |
| 401 | }, |
| 402 | "last_export": {}, |
| 403 | }) |
| 404 | except Exception as exc: # noqa: BLE001 |
| 405 | errors.append(exc) |
| 406 | |
| 407 | threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] |
| 408 | for t in threads: |
| 409 | t.start() |
| 410 | for t in threads: |
| 411 | t.join() |
| 412 | |
| 413 | assert not errors, f"Errors during concurrent writes: {errors}" |
| 414 | # File must still be valid TOML after concurrent writes |
| 415 | state = _read_bs(root) |
| 416 | assert "last_import" in state |
| 417 | |
| 418 | def test_concurrent_reads_are_safe(self, tmp_path: pathlib.Path) -> None: |
| 419 | root = _fake_muse_repo(tmp_path) |
| 420 | _write_bs(root, { |
| 421 | "last_import": {"muse_commit_id": fake_id("base"), "commits_written": 42}, |
| 422 | "last_export": {}, |
| 423 | }) |
| 424 | errors: list[Exception] = [] |
| 425 | results: list[dict] = [] |
| 426 | |
| 427 | def _reader() -> None: |
| 428 | try: |
| 429 | results.append(_read_bs(root)) |
| 430 | except Exception as exc: # noqa: BLE001 |
| 431 | errors.append(exc) |
| 432 | |
| 433 | threads = [threading.Thread(target=_reader) for _ in range(8)] |
| 434 | for t in threads: |
| 435 | t.start() |
| 436 | for t in threads: |
| 437 | t.join() |
| 438 | |
| 439 | assert not errors |
| 440 | assert all(r["last_import"].get("commits_written") == 42 for r in results) |