gabriel / muse public
test_bridge_hooks.py python
740 lines 29.2 KB
Raw
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 29 days ago
1 """TDD contract for .muse/bridge-hooks.toml — pre/post bridge hook system.
2
3 Issue #55 (musehub): AaronRene needs hooks to enforce security audits and
4 auto-open PRs on every git-export without relying on individual memory.
5
6 Config format (.muse/bridge-hooks.toml):
7 [pre_bridge]
8 hooks = [
9 { run = "npm audit fix", on_fail = "block" },
10 ]
11
12 [post_bridge]
13 hooks = [
14 { run = "gh pr create --base main --head muse-mirror", on_fail = "warn" },
15 ]
16
17 Hook execution contract:
18 - pre_bridge hooks run in the Muse repo root (before sync_to_git).
19 - post_bridge hooks run in the git mirror directory (after git_commit).
20 - on_fail = "block" → non-zero exit aborts the bridge run (SystemExit).
21 - on_fail = "warn" → non-zero exit prints a warning and continues.
22 - MUSE_BRIDGE_GIT_DIR, MUSE_BRIDGE_GIT_BRANCH, MUSE_BRIDGE_COMMIT_ID are
23 injected as environment variables for every hook invocation.
24
25 Phases
26 ------
27 BH-1 load_bridge_hooks — reads and validates .muse/bridge-hooks.toml
28 BH-2 run_hook — executes a single hook with correct cwd / env
29 BH-3 run_hooks — runs a list of hooks in order, honouring on_fail
30 BH-4 git-export integration — pre/post hooks fire at the right moments
31 """
32 from __future__ import annotations
33
34 import argparse
35 import pathlib
36 import subprocess
37 import sys
38 import textwrap
39 from typing import TypedDict
40 from unittest.mock import MagicMock, patch, call
41
42 import pytest
43
44 from muse.cli.commands.bridge import BridgeHook
45
46 _Env = dict[str, str] # process environment map
47 _Manifest = dict[str, str] # snapshot manifest: path → object_id
48 _ArgVal = str | bool | int | None | list[str] # argparse Namespace field values
49
50
51 class _RunCapture(TypedDict):
52 cmd: str | list[str]
53 cwd: pathlib.Path | None
54
55
56 # ---------------------------------------------------------------------------
57 # Phase BH-1: load_bridge_hooks
58 # ---------------------------------------------------------------------------
59
60 class TestLoadBridgeHooks:
61 """BH-1: load_bridge_hooks reads .muse/bridge-hooks.toml."""
62
63 def test_returns_empty_hooks_when_file_missing(self, tmp_path: pathlib.Path) -> None:
64 """No .muse/bridge-hooks.toml → empty pre and post lists, no error."""
65 from muse.cli.commands.bridge import load_bridge_hooks
66
67 hooks = load_bridge_hooks(tmp_path)
68
69 assert hooks.pre_bridge == [], "pre_bridge must be empty when file is missing"
70 assert hooks.post_bridge == [], "post_bridge must be empty when file is missing"
71
72 def test_loads_pre_bridge_hooks(self, tmp_path: pathlib.Path) -> None:
73 """pre_bridge section is parsed into a list of BridgeHook objects."""
74 from muse.cli.commands.bridge import load_bridge_hooks, BridgeHook
75
76 hooks_file = tmp_path / ".muse" / "bridge-hooks.toml"
77 hooks_file.parent.mkdir(parents=True, exist_ok=True)
78 hooks_file.write_text(textwrap.dedent("""\
79 [pre_bridge]
80 hooks = [
81 { run = "npm audit fix", on_fail = "block" },
82 { run = "echo ready", on_fail = "warn" },
83 ]
84 """))
85
86 result = load_bridge_hooks(tmp_path)
87
88 assert len(result.pre_bridge) == 2
89 assert result.pre_bridge[0] == BridgeHook(run="npm audit fix", on_fail="block")
90 assert result.pre_bridge[1] == BridgeHook(run="echo ready", on_fail="warn")
91
92 def test_loads_post_bridge_hooks(self, tmp_path: pathlib.Path) -> None:
93 """post_bridge section is parsed correctly."""
94 from muse.cli.commands.bridge import load_bridge_hooks, BridgeHook
95
96 hooks_file = tmp_path / ".muse" / "bridge-hooks.toml"
97 hooks_file.parent.mkdir(parents=True, exist_ok=True)
98 hooks_file.write_text(textwrap.dedent("""\
99 [post_bridge]
100 hooks = [
101 { run = "gh pr create --base main --head muse-mirror", on_fail = "warn" },
102 ]
103 """))
104
105 result = load_bridge_hooks(tmp_path)
106
107 assert len(result.post_bridge) == 1
108 assert result.post_bridge[0] == BridgeHook(
109 run="gh pr create --base main --head muse-mirror", on_fail="warn"
110 )
111 assert result.pre_bridge == []
112
113 def test_returns_empty_sections_when_section_missing(self, tmp_path: pathlib.Path) -> None:
114 """File exists but a section is absent → empty list for that section."""
115 from muse.cli.commands.bridge import load_bridge_hooks
116
117 hooks_file = tmp_path / ".muse" / "bridge-hooks.toml"
118 hooks_file.parent.mkdir(parents=True, exist_ok=True)
119 hooks_file.write_text(textwrap.dedent("""\
120 [pre_bridge]
121 hooks = [{ run = "echo hi", on_fail = "block" }]
122 """))
123
124 result = load_bridge_hooks(tmp_path)
125
126 assert len(result.pre_bridge) == 1
127 assert result.post_bridge == [], "absent post_bridge section must yield empty list"
128
129 def test_invalid_toml_raises_user_error(self, tmp_path: pathlib.Path) -> None:
130 """Malformed TOML prints a clear error and raises SystemExit."""
131 from muse.cli.commands.bridge import load_bridge_hooks
132
133 hooks_file = tmp_path / ".muse" / "bridge-hooks.toml"
134 hooks_file.parent.mkdir(parents=True, exist_ok=True)
135 hooks_file.write_text("[[[ invalid toml")
136
137 with pytest.raises(SystemExit):
138 load_bridge_hooks(tmp_path)
139
140 def test_invalid_on_fail_value_raises_user_error(self, tmp_path: pathlib.Path) -> None:
141 """on_fail must be 'block' or 'warn' — anything else raises SystemExit."""
142 from muse.cli.commands.bridge import load_bridge_hooks
143
144 hooks_file = tmp_path / ".muse" / "bridge-hooks.toml"
145 hooks_file.parent.mkdir(parents=True, exist_ok=True)
146 hooks_file.write_text(textwrap.dedent("""\
147 [pre_bridge]
148 hooks = [{ run = "echo hi", on_fail = "explode" }]
149 """))
150
151 with pytest.raises(SystemExit):
152 load_bridge_hooks(tmp_path)
153
154 def test_missing_run_field_raises_user_error(self, tmp_path: pathlib.Path) -> None:
155 """A hook entry without 'run' raises SystemExit."""
156 from muse.cli.commands.bridge import load_bridge_hooks
157
158 hooks_file = tmp_path / ".muse" / "bridge-hooks.toml"
159 hooks_file.parent.mkdir(parents=True, exist_ok=True)
160 hooks_file.write_text(textwrap.dedent("""\
161 [pre_bridge]
162 hooks = [{ on_fail = "block" }]
163 """))
164
165 with pytest.raises(SystemExit):
166 load_bridge_hooks(tmp_path)
167
168
169 # ---------------------------------------------------------------------------
170 # Phase BH-2/3: run_hook / run_hooks
171 # ---------------------------------------------------------------------------
172
173 class TestRunHook:
174 """BH-2: run_hook executes a single hook with correct cwd and env."""
175
176 def test_hook_runs_in_given_cwd(self, tmp_path: pathlib.Path) -> None:
177 """run_hook must execute the command in the specified directory."""
178 from muse.cli.commands.bridge import run_hook, BridgeHook
179
180 hook = BridgeHook(run="pwd", on_fail="block")
181 captured: list[_RunCapture] = []
182
183 def _fake_run(
184 cmd: str | list[str],
185 shell: bool = False,
186 cwd: pathlib.Path | None = None,
187 env: _Env | None = None,
188 ) -> subprocess.CompletedProcess[str]:
189 captured.append({"cmd": cmd, "cwd": cwd})
190 return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
191
192 with patch("subprocess.run", side_effect=_fake_run):
193 run_hook(hook, cwd=tmp_path, env={})
194
195 assert captured, "subprocess.run was not called"
196 assert captured[0]["cwd"] == tmp_path
197
198 def test_hook_injects_env_vars(self, tmp_path: pathlib.Path) -> None:
199 """MUSE_BRIDGE_* env vars must be present in the subprocess environment."""
200 from muse.cli.commands.bridge import run_hook, BridgeHook
201
202 hook = BridgeHook(run="echo $MUSE_BRIDGE_GIT_DIR", on_fail="block")
203 captured_env: _Env = {}
204
205 def _fake_run(
206 cmd: str | list[str],
207 shell: bool = False,
208 cwd: pathlib.Path | None = None,
209 env: _Env | None = None,
210 ) -> subprocess.CompletedProcess[str]:
211 captured_env.update(env or {})
212 return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
213
214 extra_env = {
215 "MUSE_BRIDGE_GIT_DIR": "/tmp/git-mirror",
216 "MUSE_BRIDGE_GIT_BRANCH": "muse-mirror",
217 "MUSE_BRIDGE_COMMIT_ID": "sha256:abc123",
218 }
219
220 with patch("subprocess.run", side_effect=_fake_run):
221 run_hook(hook, cwd=tmp_path, env=extra_env)
222
223 assert captured_env.get("MUSE_BRIDGE_GIT_DIR") == "/tmp/git-mirror"
224 assert captured_env.get("MUSE_BRIDGE_GIT_BRANCH") == "muse-mirror"
225 assert captured_env.get("MUSE_BRIDGE_COMMIT_ID") == "sha256:abc123"
226
227 def test_block_hook_raises_on_nonzero_exit(self, tmp_path: pathlib.Path) -> None:
228 """on_fail='block' + non-zero exit → SystemExit."""
229 from muse.cli.commands.bridge import run_hook, BridgeHook
230
231 hook = BridgeHook(run="exit 1", on_fail="block")
232
233 def _fake_run(
234 cmd: str | list[str],
235 shell: bool = False,
236 cwd: pathlib.Path | None = None,
237 env: _Env | None = None,
238 ) -> subprocess.CompletedProcess[str]:
239 return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="audit failed")
240
241 with patch("subprocess.run", side_effect=_fake_run):
242 with pytest.raises(SystemExit):
243 run_hook(hook, cwd=tmp_path, env={})
244
245 def test_warn_hook_does_not_raise_on_nonzero_exit(
246 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
247 ) -> None:
248 """on_fail='warn' + non-zero exit → prints warning, does NOT raise."""
249 from muse.cli.commands.bridge import run_hook, BridgeHook
250
251 hook = BridgeHook(run="exit 1", on_fail="warn")
252
253 def _fake_run(
254 cmd: str | list[str],
255 shell: bool = False,
256 cwd: pathlib.Path | None = None,
257 env: _Env | None = None,
258 ) -> subprocess.CompletedProcess[str]:
259 return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="pr already exists")
260
261 with patch("subprocess.run", side_effect=_fake_run):
262 run_hook(hook, cwd=tmp_path, env={}) # must not raise
263
264 captured = capsys.readouterr()
265 assert "warn" in (captured.err + captured.out).lower(), (
266 "a warning message must be printed for on_fail='warn' failures"
267 )
268
269 def test_successful_hook_does_not_raise(self, tmp_path: pathlib.Path) -> None:
270 """Exit code 0 → no exception for either on_fail value."""
271 from muse.cli.commands.bridge import run_hook, BridgeHook
272
273 for on_fail in ("block", "warn"):
274 hook = BridgeHook(run="echo ok", on_fail=on_fail)
275
276 def _fake_run(
277 cmd: str | list[str],
278 shell: bool = False,
279 cwd: pathlib.Path | None = None,
280 env: _Env | None = None,
281 ) -> subprocess.CompletedProcess[str]:
282 return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")
283
284 with patch("subprocess.run", side_effect=_fake_run):
285 run_hook(hook, cwd=tmp_path, env={}) # must not raise
286
287
288 class TestRunHooks:
289 """BH-3: run_hooks runs a list in order and stops on block failure."""
290
291 def test_runs_all_hooks_in_order(self, tmp_path: pathlib.Path) -> None:
292 """All hooks in the list are executed in order."""
293 from muse.cli.commands.bridge import run_hooks, BridgeHook
294
295 order: list[str] = []
296
297 def _fake_run(
298 cmd: str | list[str],
299 shell: bool = False,
300 cwd: pathlib.Path | None = None,
301 env: _Env | None = None,
302 ) -> subprocess.CompletedProcess[str]:
303 order.append(cmd) # type: ignore[arg-type]
304 return subprocess.CompletedProcess(cmd, 0)
305
306 hooks = [
307 BridgeHook(run="first", on_fail="block"),
308 BridgeHook(run="second", on_fail="warn"),
309 BridgeHook(run="third", on_fail="block"),
310 ]
311
312 with patch("subprocess.run", side_effect=_fake_run):
313 run_hooks(hooks, cwd=tmp_path, env={})
314
315 assert order == [["first"], ["second"], ["third"]]
316
317 def test_block_failure_stops_subsequent_hooks(self, tmp_path: pathlib.Path) -> None:
318 """A block failure must prevent later hooks from running."""
319 from muse.cli.commands.bridge import run_hooks, BridgeHook
320
321 ran: list[str] = []
322
323 def _fake_run(
324 cmd: str | list[str],
325 shell: bool = False,
326 cwd: pathlib.Path | None = None,
327 env: _Env | None = None,
328 ) -> subprocess.CompletedProcess[str]:
329 ran.append(cmd) # type: ignore[arg-type]
330 rc = 1 if cmd == ["fail"] else 0
331 return subprocess.CompletedProcess(cmd, rc)
332
333 hooks = [
334 BridgeHook(run="first", on_fail="warn"),
335 BridgeHook(run="fail", on_fail="block"),
336 BridgeHook(run="should-not-run", on_fail="warn"),
337 ]
338
339 with patch("subprocess.run", side_effect=_fake_run):
340 with pytest.raises(SystemExit):
341 run_hooks(hooks, cwd=tmp_path, env={})
342
343 assert ["should-not-run"] not in ran, (
344 "hooks after a blocking failure must not be executed"
345 )
346
347 def test_warn_failure_continues_to_next_hook(self, tmp_path: pathlib.Path) -> None:
348 """A warn failure must not stop the hook chain."""
349 from muse.cli.commands.bridge import run_hooks, BridgeHook
350
351 ran: list[str] = []
352
353 def _fake_run(
354 cmd: str | list[str],
355 shell: bool = False,
356 cwd: pathlib.Path | None = None,
357 env: _Env | None = None,
358 ) -> subprocess.CompletedProcess[str]:
359 ran.append(cmd) # type: ignore[arg-type]
360 rc = 1 if cmd == ["fails-but-warns"] else 0
361 return subprocess.CompletedProcess(cmd, rc)
362
363 hooks = [
364 BridgeHook(run="fails-but-warns", on_fail="warn"),
365 BridgeHook(run="still-runs", on_fail="block"),
366 ]
367
368 with patch("subprocess.run", side_effect=_fake_run):
369 run_hooks(hooks, cwd=tmp_path, env={})
370
371 assert ["still-runs"] in ran, "hook chain must continue after a warn failure"
372
373 def test_empty_hook_list_is_a_noop(self, tmp_path: pathlib.Path) -> None:
374 """Empty list → no subprocess calls, no exceptions."""
375 from muse.cli.commands.bridge import run_hooks
376
377 with patch("subprocess.run") as mock_run:
378 run_hooks([], cwd=tmp_path, env={})
379
380 mock_run.assert_not_called()
381
382
383 # ---------------------------------------------------------------------------
384 # Phase BH-4: git-export integration
385 # ---------------------------------------------------------------------------
386
387 class TestGitExportHookIntegration:
388 """BH-4: pre/post hooks fire at the correct points in run_git_export."""
389
390 def _make_args(self, git_dir: pathlib.Path, **overrides: _ArgVal) -> argparse.Namespace:
391 """Minimal argparse.Namespace for run_git_export."""
392 defaults = dict(
393 git_dir=str(git_dir),
394 json_out=False,
395 dry_run=False,
396 no_push=True,
397 force_push=False,
398 git_branch="muse-mirror",
399 git_remote="origin",
400 muse_ref=None,
401 excludes=[],
402 strip_muse_metadata=True,
403 fix_modes=False,
404 allow_empty=True,
405 commit_message="mirror: muse {commit_id}",
406 watch=None,
407 export_rerere=False,
408 export_shelves=False,
409 )
410 defaults.update(overrides)
411 return argparse.Namespace(**defaults)
412
413 def test_pre_bridge_hooks_run_before_sync(
414 self, tmp_path: pathlib.Path
415 ) -> None:
416 """pre_bridge hooks must execute before sync_to_git is called."""
417 from muse.cli.commands.bridge import load_bridge_hooks, BridgeHooks, BridgeHook
418
419 muse_root = tmp_path / "muse_repo"
420 muse_root.mkdir()
421 git_dir = tmp_path / "git_repo"
422 git_dir.mkdir()
423 (git_dir / ".git").mkdir()
424
425 hooks_file = muse_root / ".muse" / "bridge-hooks.toml"
426 hooks_file.parent.mkdir(parents=True)
427 hooks_file.write_text(textwrap.dedent("""\
428 [pre_bridge]
429 hooks = [{ run = "echo pre", on_fail = "block" }]
430 """))
431
432 call_order: list[str] = []
433
434 def _fake_pre_hook(hook: BridgeHook, *, cwd: pathlib.Path, env: _Env) -> None:
435 call_order.append("pre_hook")
436
437 def _fake_sync(
438 manifest: _Manifest,
439 *,
440 excludes: list[str] | None = None,
441 strip_muse: bool = True,
442 fix_modes: bool = False,
443 ) -> int:
444 call_order.append("sync_to_git")
445 return 0
446
447 with (
448 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
449 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
450 patch("muse.cli.commands.bridge.run_hook", side_effect=_fake_pre_hook),
451 patch("muse.cli.commands.bridge._ensure_git_branch"),
452 patch("muse.cli.commands.bridge.read_bridge_state", return_value={}),
453 patch("muse.cli.commands.bridge.write_bridge_state"),
454 ):
455 instance = MockExporter.return_value
456 instance.resolve_muse_ref.return_value = ("sha256:" + "a" * 64, "sha256:" + "b" * 64)
457 instance.read_snapshot.return_value = {}
458 instance.sync_to_git.side_effect = _fake_sync
459 instance.git_commit.return_value = "abc123"
460 instance.muse_branch = "main"
461
462 from muse.cli.commands.bridge import run_git_export
463 run_git_export(self._make_args(git_dir))
464
465 pre_idx = next((i for i, v in enumerate(call_order) if v == "pre_hook"), None)
466 sync_idx = next((i for i, v in enumerate(call_order) if v == "sync_to_git"), None)
467
468 assert pre_idx is not None, "pre_bridge hook was not called"
469 assert sync_idx is not None, "sync_to_git was not called"
470 assert pre_idx < sync_idx, (
471 f"pre_bridge hook must run before sync_to_git "
472 f"(pre_hook at {pre_idx}, sync_to_git at {sync_idx})"
473 )
474
475 def test_post_bridge_hooks_run_after_commit(
476 self, tmp_path: pathlib.Path
477 ) -> None:
478 """post_bridge hooks must execute after git_commit."""
479 from muse.cli.commands.bridge import BridgeHook
480
481 muse_root = tmp_path / "muse_repo"
482 muse_root.mkdir()
483 git_dir = tmp_path / "git_repo"
484 git_dir.mkdir()
485 (git_dir / ".git").mkdir()
486
487 hooks_file = muse_root / ".muse" / "bridge-hooks.toml"
488 hooks_file.parent.mkdir(parents=True)
489 hooks_file.write_text(textwrap.dedent("""\
490 [post_bridge]
491 hooks = [{ run = "gh pr create", on_fail = "warn" }]
492 """))
493
494 call_order: list[str] = []
495
496 def _fake_post_hook(hook: BridgeHook, *, cwd: pathlib.Path, env: _Env) -> None:
497 call_order.append(("post_hook", str(cwd)))
498
499 def _fake_commit(
500 commit_id: str,
501 commit_message: str,
502 *,
503 allow_empty: bool = True,
504 ) -> str:
505 call_order.append("git_commit")
506 return "deadbeef"
507
508 with (
509 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
510 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
511 patch("muse.cli.commands.bridge.run_hook", side_effect=_fake_post_hook),
512 patch("muse.cli.commands.bridge._ensure_git_branch"),
513 patch("muse.cli.commands.bridge.read_bridge_state", return_value={}),
514 patch("muse.cli.commands.bridge.write_bridge_state"),
515 ):
516 instance = MockExporter.return_value
517 instance.resolve_muse_ref.return_value = ("sha256:" + "a" * 64, "sha256:" + "b" * 64)
518 instance.read_snapshot.return_value = {}
519 instance.sync_to_git.return_value = 5
520 instance.git_commit.side_effect = _fake_commit
521 instance.muse_branch = "main"
522
523 from muse.cli.commands.bridge import run_git_export
524 run_git_export(self._make_args(git_dir))
525
526 commit_idx = next((i for i, v in enumerate(call_order) if v == "git_commit"), None)
527 post_idx = next(
528 (i for i, v in enumerate(call_order)
529 if isinstance(v, tuple) and v[0] == "post_hook"),
530 None,
531 )
532
533 assert commit_idx is not None, "git_commit was not called"
534 assert post_idx is not None, "post_bridge hook was not called"
535 assert post_idx > commit_idx, (
536 f"post_bridge hook must run after git_commit "
537 f"(commit at {commit_idx}, post_hook at {post_idx})"
538 )
539
540 def test_post_bridge_hooks_run_in_git_dir(
541 self, tmp_path: pathlib.Path
542 ) -> None:
543 """post_bridge hooks must use git_dir as cwd (for gh pr create etc.)."""
544 muse_root = tmp_path / "muse_repo"
545 muse_root.mkdir()
546 git_dir = tmp_path / "git_repo"
547 git_dir.mkdir()
548 (git_dir / ".git").mkdir()
549
550 hooks_file = muse_root / ".muse" / "bridge-hooks.toml"
551 hooks_file.parent.mkdir(parents=True)
552 hooks_file.write_text(textwrap.dedent("""\
553 [post_bridge]
554 hooks = [{ run = "gh pr create", on_fail = "warn" }]
555 """))
556
557 post_cwds: list[pathlib.Path] = []
558
559 def _capture_post_hook(hook: BridgeHook, *, cwd: pathlib.Path, env: _Env) -> None:
560 post_cwds.append(cwd)
561
562 with (
563 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
564 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
565 patch("muse.cli.commands.bridge.run_hook", side_effect=_capture_post_hook),
566 patch("muse.cli.commands.bridge._ensure_git_branch"),
567 patch("muse.cli.commands.bridge.read_bridge_state", return_value={}),
568 patch("muse.cli.commands.bridge.write_bridge_state"),
569 ):
570 instance = MockExporter.return_value
571 instance.resolve_muse_ref.return_value = ("sha256:" + "a" * 64, "sha256:" + "b" * 64)
572 instance.read_snapshot.return_value = {}
573 instance.sync_to_git.return_value = 5
574 instance.git_commit.return_value = "deadbeef"
575 instance.muse_branch = "main"
576
577 from muse.cli.commands.bridge import run_git_export
578 run_git_export(self._make_args(git_dir))
579
580 assert post_cwds, "post_bridge hook cwd was not captured"
581 assert post_cwds[0] == git_dir, (
582 f"post_bridge hooks must run in git_dir={git_dir}, got {post_cwds[0]}"
583 )
584
585 def test_pre_bridge_hooks_run_in_muse_root(
586 self, tmp_path: pathlib.Path
587 ) -> None:
588 """pre_bridge hooks must use the muse repo root as cwd."""
589 muse_root = tmp_path / "muse_repo"
590 muse_root.mkdir()
591 git_dir = tmp_path / "git_repo"
592 git_dir.mkdir()
593 (git_dir / ".git").mkdir()
594
595 hooks_file = muse_root / ".muse" / "bridge-hooks.toml"
596 hooks_file.parent.mkdir(parents=True)
597 hooks_file.write_text(textwrap.dedent("""\
598 [pre_bridge]
599 hooks = [{ run = "npm audit fix", on_fail = "block" }]
600 """))
601
602 pre_cwds: list[pathlib.Path] = []
603
604 def _capture_pre_hook(hook: BridgeHook, *, cwd: pathlib.Path, env: _Env) -> None:
605 pre_cwds.append(cwd)
606
607 with (
608 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
609 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
610 patch("muse.cli.commands.bridge.run_hook", side_effect=_capture_pre_hook),
611 patch("muse.cli.commands.bridge._ensure_git_branch"),
612 patch("muse.cli.commands.bridge.read_bridge_state", return_value={}),
613 patch("muse.cli.commands.bridge.write_bridge_state"),
614 ):
615 instance = MockExporter.return_value
616 instance.resolve_muse_ref.return_value = ("sha256:" + "a" * 64, "sha256:" + "b" * 64)
617 instance.read_snapshot.return_value = {}
618 instance.sync_to_git.return_value = 5
619 instance.git_commit.return_value = "deadbeef"
620 instance.muse_branch = "main"
621
622 from muse.cli.commands.bridge import run_git_export
623 run_git_export(self._make_args(git_dir))
624
625 assert pre_cwds, "pre_bridge hook cwd was not captured"
626 assert pre_cwds[0] == muse_root, (
627 f"pre_bridge hooks must run in muse_root={muse_root}, got {pre_cwds[0]}"
628 )
629
630 def test_blocking_pre_hook_failure_aborts_export(
631 self, tmp_path: pathlib.Path
632 ) -> None:
633 """A blocking pre_bridge failure must abort before sync_to_git."""
634 muse_root = tmp_path / "muse_repo"
635 muse_root.mkdir()
636 git_dir = tmp_path / "git_repo"
637 git_dir.mkdir()
638 (git_dir / ".git").mkdir()
639
640 hooks_file = muse_root / ".muse" / "bridge-hooks.toml"
641 hooks_file.parent.mkdir(parents=True)
642 hooks_file.write_text(textwrap.dedent("""\
643 [pre_bridge]
644 hooks = [{ run = "npm audit", on_fail = "block" }]
645 """))
646
647 def _fail_pre_hook(hook: BridgeHook, *, cwd: pathlib.Path, env: _Env) -> None:
648 raise SystemExit(1)
649
650 with (
651 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
652 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
653 patch("muse.cli.commands.bridge.run_hook", side_effect=_fail_pre_hook),
654 patch("muse.cli.commands.bridge._ensure_git_branch"),
655 ):
656 instance = MockExporter.return_value
657 instance.resolve_muse_ref.return_value = ("sha256:" + "a" * 64, "sha256:" + "b" * 64)
658 instance.read_snapshot.return_value = {}
659
660 from muse.cli.commands.bridge import run_git_export
661 with pytest.raises(SystemExit):
662 run_git_export(self._make_args(git_dir))
663
664 instance.sync_to_git.assert_not_called()
665
666 def test_env_vars_injected_into_hooks(self, tmp_path: pathlib.Path) -> None:
667 """MUSE_BRIDGE_* env vars are passed to every hook invocation."""
668 muse_root = tmp_path / "muse_repo"
669 muse_root.mkdir()
670 git_dir = tmp_path / "git_repo"
671 git_dir.mkdir()
672 (git_dir / ".git").mkdir()
673
674 hooks_file = muse_root / ".muse" / "bridge-hooks.toml"
675 hooks_file.parent.mkdir(parents=True)
676 hooks_file.write_text(textwrap.dedent("""\
677 [pre_bridge]
678 hooks = [{ run = "echo $MUSE_BRIDGE_COMMIT_ID", on_fail = "warn" }]
679 """))
680
681 captured_envs: list[_Env] = []
682
683 def _capture_hook(hook: BridgeHook, *, cwd: pathlib.Path, env: _Env) -> None:
684 captured_envs.append(dict(env))
685
686 commit_id = "sha256:" + "a" * 64
687
688 with (
689 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
690 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
691 patch("muse.cli.commands.bridge.run_hook", side_effect=_capture_hook),
692 patch("muse.cli.commands.bridge._ensure_git_branch"),
693 patch("muse.cli.commands.bridge.read_bridge_state", return_value={}),
694 patch("muse.cli.commands.bridge.write_bridge_state"),
695 ):
696 instance = MockExporter.return_value
697 instance.resolve_muse_ref.return_value = (commit_id, "sha256:" + "b" * 64)
698 instance.read_snapshot.return_value = {}
699 instance.sync_to_git.return_value = 3
700 instance.git_commit.return_value = "deadbeef"
701 instance.muse_branch = "main"
702
703 from muse.cli.commands.bridge import run_git_export
704 run_git_export(self._make_args(git_dir))
705
706 assert captured_envs, "run_hook was not called"
707 env = captured_envs[0]
708 assert env.get("MUSE_BRIDGE_COMMIT_ID") == commit_id, (
709 f"MUSE_BRIDGE_COMMIT_ID not injected; got env keys: {list(env.keys())}"
710 )
711 assert "MUSE_BRIDGE_GIT_DIR" in env
712 assert "MUSE_BRIDGE_GIT_BRANCH" in env
713
714 def test_no_hooks_file_export_runs_normally(self, tmp_path: pathlib.Path) -> None:
715 """Absence of bridge-hooks.toml must not affect a normal export."""
716 muse_root = tmp_path / "muse_repo"
717 muse_root.mkdir()
718 (muse_root / ".muse").mkdir() # no bridge-hooks.toml
719 git_dir = tmp_path / "git_repo"
720 git_dir.mkdir()
721 (git_dir / ".git").mkdir()
722
723 with (
724 patch("muse.cli.commands.bridge.find_repo_root", return_value=muse_root),
725 patch("muse.cli.commands.bridge.GitExporter") as MockExporter,
726 patch("muse.cli.commands.bridge._ensure_git_branch"),
727 patch("muse.cli.commands.bridge.read_bridge_state", return_value={}),
728 patch("muse.cli.commands.bridge.write_bridge_state"),
729 ):
730 instance = MockExporter.return_value
731 instance.resolve_muse_ref.return_value = ("sha256:" + "a" * 64, "sha256:" + "b" * 64)
732 instance.read_snapshot.return_value = {}
733 instance.sync_to_git.return_value = 2
734 instance.git_commit.return_value = "abc"
735 instance.muse_branch = "main"
736
737 from muse.cli.commands.bridge import run_git_export
738 run_git_export(self._make_args(git_dir)) # must not raise
739
740 instance.sync_to_git.assert_called_once()
File History 2 commits
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 29 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago