test_bridge_hooks.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.hooks 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 449 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 450 | patch("muse.core.bridge.exporter.run_hook", side_effect=_fake_pre_hook), |
| 451 | patch("muse.core.bridge.exporter._ensure_git_branch"), |
| 452 | patch("muse.core.bridge.exporter.read_bridge_state", return_value={}), |
| 453 | patch("muse.core.bridge.exporter.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.core.bridge.exporter 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.core.bridge.hooks 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 510 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 511 | patch("muse.core.bridge.exporter.run_hook", side_effect=_fake_post_hook), |
| 512 | patch("muse.core.bridge.exporter._ensure_git_branch"), |
| 513 | patch("muse.core.bridge.exporter.read_bridge_state", return_value={}), |
| 514 | patch("muse.core.bridge.exporter.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.core.bridge.exporter 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 564 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 565 | patch("muse.core.bridge.exporter.run_hook", side_effect=_capture_post_hook), |
| 566 | patch("muse.core.bridge.exporter._ensure_git_branch"), |
| 567 | patch("muse.core.bridge.exporter.read_bridge_state", return_value={}), |
| 568 | patch("muse.core.bridge.exporter.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.core.bridge.exporter 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 609 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 610 | patch("muse.core.bridge.exporter.run_hook", side_effect=_capture_pre_hook), |
| 611 | patch("muse.core.bridge.exporter._ensure_git_branch"), |
| 612 | patch("muse.core.bridge.exporter.read_bridge_state", return_value={}), |
| 613 | patch("muse.core.bridge.exporter.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.core.bridge.exporter 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 652 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 653 | patch("muse.core.bridge.exporter.run_hook", side_effect=_fail_pre_hook), |
| 654 | patch("muse.core.bridge.exporter._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.core.bridge.exporter 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 690 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 691 | patch("muse.core.bridge.exporter.run_hook", side_effect=_capture_hook), |
| 692 | patch("muse.core.bridge.exporter._ensure_git_branch"), |
| 693 | patch("muse.core.bridge.exporter.read_bridge_state", return_value={}), |
| 694 | patch("muse.core.bridge.exporter.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.core.bridge.exporter 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.core.bridge.exporter.find_repo_root", return_value=muse_root), |
| 725 | patch("muse.core.bridge.exporter.GitExporter") as MockExporter, |
| 726 | patch("muse.core.bridge.exporter._ensure_git_branch"), |
| 727 | patch("muse.core.bridge.exporter.read_bridge_state", return_value={}), |
| 728 | patch("muse.core.bridge.exporter.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.core.bridge.exporter 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() |