"""Phase 1 TDD tests for ``muse bridge`` — namespace structure and bridge state. Tests are organised into seven tiers: Tier 1 — Unit BridgeState TypedDict, read_bridge_state, write_bridge_state Tier 2 — Contract CLI namespace: --help shows all subcommands and required flags Tier 3 — Integration CliRunner invocation of each subcommand Tier 4 — Property Bridge state round-trips for arbitrary valid state dicts Tier 5 — Regression git-bridge.toml backward-compat: file with partial keys is readable Tier 6 — Security Path traversal in bridge state location rejected / contained Tier 7 — Stress Concurrent reads and writes to bridge state do not corrupt data """ from __future__ import annotations import pathlib import threading import tomllib from unittest.mock import patch import pytest from hypothesis import HealthCheck, given, settings from hypothesis import strategies as st from muse.core.bridge.state import BridgeState from muse.core.paths import git_bridge_state_path, init_repo_dirs from muse.core.types import fake_id from tests.cli_test_helper import CliRunner runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _invoke(*args: str) -> "CliRunner": return runner.invoke(None, list(args)) def _read_bs(root: pathlib.Path) -> BridgeState: """Import lazily so tests can be collected before bridge.py exists.""" from muse.core.bridge.state import read_bridge_state return read_bridge_state(root) def _write_bs(root: pathlib.Path, state: BridgeState) -> None: from muse.core.bridge.state import write_bridge_state write_bridge_state(root, state) def _fake_muse_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Create a minimal .muse/ layout so find_repo_root() succeeds.""" return init_repo_dirs(tmp_path) # =========================================================================== # Tier 1 — Unit: TypedDict, read_bridge_state, write_bridge_state # =========================================================================== class TestBridgeStateTypedDict: """BridgeState TypedDict has the right structure.""" def test_import_and_instantiate(self) -> None: """BridgeState can be imported and used as a TypedDict.""" from muse.core.bridge.state import BridgeState state: BridgeState = { "last_import": { "git_sha": "a" * 40, "git_ref": "main", "git_remote": "origin", "muse_branch": "main", "muse_commit_id": fake_id("commit-1"), "imported_at": "2026-01-01T00:00:00Z", "commits_written": 3, }, "last_export": { "muse_branch": "main", "muse_commit_id": fake_id("commit-2"), "git_remote": "origin", "git_ref": "muse-mirror", "git_sha": "b" * 40, "exported_at": "2026-01-01T01:00:00Z", }, } assert state["last_import"]["commits_written"] == 3 assert state["last_export"]["git_remote"] == "origin" def test_muse_commit_id_is_prefixed(self) -> None: """muse_commit_id values always carry sha256: prefix.""" from muse.core.bridge.state import BridgeState cid = fake_id("test-commit") assert cid.startswith("sha256:") state: BridgeState = { "last_import": { "git_sha": "c" * 40, "git_ref": "main", "git_remote": "origin", "muse_branch": "main", "muse_commit_id": cid, "imported_at": "2026-01-01T00:00:00Z", "commits_written": 1, }, "last_export": {}, } assert state["last_import"]["muse_commit_id"].startswith("sha256:") class TestReadBridgeState: """read_bridge_state returns a default empty state when file is missing.""" def test_missing_file_returns_empty_state(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) state = _read_bs(root) assert "last_import" in state assert "last_export" in state assert state["last_import"] == {} assert state["last_export"] == {} def test_existing_file_is_parsed(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) cid = fake_id("my-commit") toml_text = f""" [last_import] git_sha = "{'a' * 40}" git_ref = "main" git_remote = "origin" muse_branch = "main" muse_commit_id = "{cid}" imported_at = "2026-01-15T10:00:00Z" commits_written = 7 """ (git_bridge_state_path(root)).write_text(toml_text) state = _read_bs(root) assert state["last_import"]["commits_written"] == 7 assert state["last_import"]["muse_commit_id"] == cid assert state["last_import"]["git_ref"] == "main" def test_partial_file_fills_missing_sections(self, tmp_path: pathlib.Path) -> None: """File with only [last_import] — last_export defaults to empty dict.""" root = _fake_muse_repo(tmp_path) (git_bridge_state_path(root)).write_text("[last_import]\ngit_ref = \"dev\"\n") state = _read_bs(root) assert state["last_import"]["git_ref"] == "dev" assert state["last_export"] == {} class TestWriteBridgeState: """write_bridge_state persists state to .muse/git-bridge.toml.""" def test_writes_toml_file(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) cid = fake_id("written-commit") state = { "last_import": { "git_sha": "d" * 40, "git_ref": "main", "git_remote": "origin", "muse_branch": "main", "muse_commit_id": cid, "imported_at": "2026-02-01T00:00:00Z", "commits_written": 5, }, "last_export": {}, } _write_bs(root, state) toml_path = git_bridge_state_path(root) assert toml_path.exists() parsed = tomllib.loads(toml_path.read_text()) assert parsed["last_import"]["muse_commit_id"] == cid assert parsed["last_import"]["commits_written"] == 5 def test_muse_commit_id_preserves_prefix(self, tmp_path: pathlib.Path) -> None: """sha256: prefix is never stripped during write.""" root = _fake_muse_repo(tmp_path) cid = fake_id("prefix-check") assert cid.startswith("sha256:") _write_bs(root, {"last_import": {"muse_commit_id": cid}, "last_export": {}}) toml_path = git_bridge_state_path(root) parsed = tomllib.loads(toml_path.read_text()) assert parsed["last_import"]["muse_commit_id"].startswith("sha256:") def test_round_trip(self, tmp_path: pathlib.Path) -> None: """write then read gives back the same state.""" root = _fake_muse_repo(tmp_path) original: BridgeState = { "last_import": {"git_sha": "e" * 40, "git_ref": "dev", "commits_written": 12}, "last_export": {"muse_commit_id": fake_id("rt"), "git_sha": "f" * 40}, } _write_bs(root, original) recovered = _read_bs(root) assert recovered["last_import"]["git_sha"] == "e" * 40 assert recovered["last_import"]["commits_written"] == 12 assert recovered["last_export"]["git_sha"] == "f" * 40 # =========================================================================== # Tier 2 — Contract: CLI namespace shows all subcommands and required flags # =========================================================================== class TestBridgeCliContract: """muse bridge --help shows git-import, git-export, git-status.""" def test_bridge_help_shows_git_import(self) -> None: result = _invoke("bridge", "--help") assert "git-import" in result.output or "git-import" in result.stderr def test_bridge_help_shows_git_export(self) -> None: result = _invoke("bridge", "--help") assert "git-export" in result.output or "git-export" in result.stderr def test_bridge_help_shows_git_status(self) -> None: result = _invoke("bridge", "--help") assert "git-status" in result.output or "git-status" in result.stderr class TestGitImportFlags: """muse bridge git-import --help exposes all required flags.""" @pytest.mark.parametrize("flag", [ "--target", "--branch", "--all", "--from-ref", "--incremental", "--attribution-map", "--sign", "--dry-run", "--json", ]) def test_flag_present(self, flag: str) -> None: result = _invoke("bridge", "git-import", "--help") combined = result.output + result.stderr assert flag in combined, f"Flag {flag!r} missing from git-import --help" class TestGitExportFlags: """muse bridge git-export --help exposes all required flags.""" @pytest.mark.parametrize("flag", [ "--muse-ref", "--git-dir", "--git-branch", "--git-remote", "--no-push", "--dry-run", "--json", ]) def test_flag_present(self, flag: str) -> None: result = _invoke("bridge", "git-export", "--help") combined = result.output + result.stderr assert flag in combined, f"Flag {flag!r} missing from git-export --help" class TestGitStatusFlags: """muse bridge git-status --help exposes all required flags.""" @pytest.mark.parametrize("flag", ["--git-dir", "--json"]) def test_flag_present(self, flag: str) -> None: result = _invoke("bridge", "git-status", "--help") combined = result.output + result.stderr assert flag in combined, f"Flag {flag!r} missing from git-status --help" # =========================================================================== # Tier 3 — Integration: CliRunner invocation # =========================================================================== class TestBridgeIntegration: """Basic invocation through CliRunner exits cleanly or with known codes.""" def test_git_import_dry_run_no_source(self) -> None: """git-import --dry-run with no SOURCE falls back to cwd gracefully.""" result = _invoke("bridge", "git-import", "--dry-run") # Should exit with user error (no valid git repo at cwd), not a crash assert result.exit_code in (0, 1, 2) def test_git_export_no_git_dir_exits_user_error(self) -> None: """git-export without --git-dir should exit with error, not crash.""" with patch("muse.core.repo.find_repo_root", return_value=None): result = _invoke("bridge", "git-export", "--git-dir", "/nonexistent/path") assert result.exit_code in (1, 2) def test_git_status_no_repo_exits_cleanly(self) -> None: """git-status with no Muse repo exits with user error, not traceback.""" with patch("muse.core.repo.find_repo_root", return_value=None): result = _invoke("bridge", "git-status") assert result.exit_code in (0, 1, 2) # =========================================================================== # Tier 4 — Property: bridge state round-trips # =========================================================================== _SAFE_TEXT = st.text( alphabet=st.characters(whitelist_categories=("L", "N"), whitelist_characters="-_/:"), min_size=0, max_size=50, ) class TestBridgeStateProperty: """Property: bridge state survives a write/read round-trip.""" @given( git_ref=_SAFE_TEXT, commits_written=st.integers(min_value=0, max_value=10_000), ) @settings(max_examples=30, suppress_health_check=[HealthCheck.function_scoped_fixture]) def test_import_state_round_trips( self, tmp_path: pathlib.Path, git_ref: str, commits_written: int ) -> None: root = _fake_muse_repo(tmp_path) cid = fake_id(f"prop-{git_ref[:10]}-{commits_written}") state = { "last_import": { "git_sha": "a" * 40, "git_ref": git_ref, "commits_written": commits_written, "muse_commit_id": cid, }, "last_export": {}, } _write_bs(root, state) recovered = _read_bs(root) assert recovered["last_import"]["commits_written"] == commits_written assert recovered["last_import"]["muse_commit_id"] == cid # =========================================================================== # Tier 5 — Regression: partial / legacy state files are tolerated # =========================================================================== class TestBridgeStateRegression: """Previously, partial state files would crash. Now they must be tolerated.""" def test_empty_toml_file_returns_empty_state(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) (git_bridge_state_path(root)).write_text("") state = _read_bs(root) assert state["last_import"] == {} assert state["last_export"] == {} def test_no_last_import_section(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) (git_bridge_state_path(root)).write_text("[last_export]\ngit_ref = \"muse-mirror\"\n") state = _read_bs(root) assert state["last_import"] == {} assert state["last_export"]["git_ref"] == "muse-mirror" def test_extra_unknown_keys_are_preserved(self, tmp_path: pathlib.Path) -> None: """Unknown TOML keys in bridge state are passed through, not rejected.""" root = _fake_muse_repo(tmp_path) (git_bridge_state_path(root)).write_text( "[last_import]\nfuture_key = \"v2\"\n" ) state = _read_bs(root) assert state["last_import"].get("future_key") == "v2" # =========================================================================== # Tier 6 — Security: path traversal in bridge state # =========================================================================== class TestBridgeStateSecurity: """Bridge state is always stored inside .muse/ — path traversal is rejected.""" def test_read_state_is_always_inside_muse(self, tmp_path: pathlib.Path) -> None: """read_bridge_state always reads from /.muse/git-bridge.toml.""" root = _fake_muse_repo(tmp_path) from muse.core.bridge.state import read_bridge_state import inspect src = inspect.getsource(read_bridge_state) # Must reference .muse/ in the path, not accept arbitrary paths from env assert ".muse" in src or "git-bridge" in src def test_write_state_stays_inside_muse(self, tmp_path: pathlib.Path) -> None: """write_bridge_state only writes to /.muse/git-bridge.toml.""" root = _fake_muse_repo(tmp_path) _write_bs(root, {"last_import": {}, "last_export": {}}) written = list(tmp_path.rglob("git-bridge.toml")) assert len(written) == 1 assert ".muse" in str(written[0]) def test_muse_commit_id_without_prefix_raises(self, tmp_path: pathlib.Path) -> None: """write_bridge_state rejects muse_commit_id without sha256: prefix.""" root = _fake_muse_repo(tmp_path) bare_hex = "a" * 64 # no sha256: prefix with pytest.raises((ValueError, SystemExit)): _write_bs(root, { "last_import": {"muse_commit_id": bare_hex}, "last_export": {}, }) # =========================================================================== # Tier 7 — Stress: concurrent reads and writes # =========================================================================== class TestBridgeStateStress: """Concurrent reads and writes to the bridge state file do not corrupt data.""" def test_concurrent_writes_do_not_corrupt(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) errors: list[Exception] = [] def _worker(i: int) -> None: try: _write_bs(root, { "last_import": { "git_sha": hex(i)[2:].zfill(40)[:40], "commits_written": i, "muse_commit_id": fake_id(f"stress-{i}"), }, "last_export": {}, }) except Exception as exc: # noqa: BLE001 errors.append(exc) threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Errors during concurrent writes: {errors}" # File must still be valid TOML after concurrent writes state = _read_bs(root) assert "last_import" in state def test_concurrent_reads_are_safe(self, tmp_path: pathlib.Path) -> None: root = _fake_muse_repo(tmp_path) _write_bs(root, { "last_import": {"muse_commit_id": fake_id("base"), "commits_written": 42}, "last_export": {}, }) errors: list[Exception] = [] results: list[dict] = [] def _reader() -> None: try: results.append(_read_bs(root)) except Exception as exc: # noqa: BLE001 errors.append(exc) threads = [threading.Thread(target=_reader) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors assert all(r["last_import"].get("commits_written") == 42 for r in results)