gabriel / muse public
test_bridge_phase1.py python
440 lines 17.3 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago
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)
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago