gabriel / muse public
test_reflog_watch.py python
143 lines 5.5 KB
Raw
sha256:799dc6f15b7dba389fdf5f088cd5e210da5fb66ec4b1ec9db7b16cde0e8fe734 feat: muse#54 item 10 — muse reflog watch live-tail Sonnet 5 patch 11 hours ago
1 """muse#54 item 10 -- `muse reflog watch [--branch <name>] [--json]`.
2
3 Live-tail of reflog entries: a 'snapshot' event per pre-existing entry,
4 then an 'appended' event for each newly written entry, until --once /
5 --timeout / --max-events ends the loop.
6 """
7 from __future__ import annotations
8
9 import json
10 import pathlib
11
12 import pytest
13
14 from tests.cli_test_helper import CliRunner
15 from muse.core.types import fake_id, Manifest, NULL_COMMIT_ID
16 from muse.core.paths import muse_dir, logs_dir
17
18 cli = None
19 runner = CliRunner()
20
21
22 def _env(root: pathlib.Path) -> Manifest:
23 return {"MUSE_REPO_ROOT": str(root)}
24
25
26 def _run(root: pathlib.Path, *args: str):
27 return runner.invoke(cli, list(args), env=_env(root))
28
29
30 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
31 dot_muse = muse_dir(tmp_path)
32 dot_muse.mkdir()
33 repo_id = fake_id("repo")
34 (dot_muse / "repo.json").write_text(json.dumps({
35 "repo_id": repo_id, "domain": "code", "default_branch": "main",
36 "created_at": "2025-01-01T00:00:00+00:00",
37 }), encoding="utf-8")
38 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
39 (dot_muse / "refs" / "heads").mkdir(parents=True)
40 (dot_muse / "snapshots").mkdir()
41 (dot_muse / "commits").mkdir()
42 (dot_muse / "objects").mkdir()
43 return tmp_path, repo_id
44
45
46 def _commit_file(root: pathlib.Path, content: str, message: str) -> str:
47 (root / "f.txt").write_text(content)
48 add = _run(root, "code", "add", "f.txt")
49 assert add.exit_code == 0, add.output
50 commit = _run(root, "commit", "-m", message, "--json")
51 assert commit.exit_code == 0, commit.output
52 return json.loads(commit.output)["commit_id"]
53
54
55 def _ndjson_lines(output: str) -> list[dict]:
56 return [json.loads(line) for line in output.splitlines() if line.strip()]
57
58
59 class TestReflogWatchOnce:
60 def test_once_emits_snapshot_only(self, tmp_path: pathlib.Path) -> None:
61 root, repo_id = _init_repo(tmp_path)
62 c1 = _commit_file(root, "v1\n", "c1")
63 c2 = _commit_file(root, "v2\n", "c2")
64
65 result = _run(root, "reflog", "watch", "--once", "--json")
66 assert result.exit_code == 0, result.output
67 events = _ndjson_lines(result.output)
68 assert len(events) == 2
69 assert all(e["event_type"] == "snapshot" for e in events)
70 # Chronological order: oldest first.
71 assert events[0]["new_id"] == c1
72 assert events[0]["index"] == 1
73 assert events[1]["new_id"] == c2
74 assert events[1]["index"] == 0
75
76 def test_once_empty_reflog_emits_nothing(self, tmp_path: pathlib.Path) -> None:
77 root, repo_id = _init_repo(tmp_path)
78 result = _run(root, "reflog", "watch", "--once", "--json")
79 assert result.exit_code == 0, result.output
80 assert _ndjson_lines(result.output) == []
81
82 def test_once_respects_branch_flag(self, tmp_path: pathlib.Path) -> None:
83 root, repo_id = _init_repo(tmp_path)
84 c1 = _commit_file(root, "v1\n", "c1")
85
86 result = _run(root, "reflog", "watch", "--once", "--branch", "main", "--json")
87 assert result.exit_code == 0, result.output
88 events = _ndjson_lines(result.output)
89 assert len(events) == 1
90 assert events[0]["new_id"] == c1
91
92 def test_invalid_branch_name_errors(self, tmp_path: pathlib.Path) -> None:
93 root, repo_id = _init_repo(tmp_path)
94 result = _run(root, "reflog", "watch", "--once", "--branch", "../escape", "--json")
95 assert result.exit_code == 1
96
97 def test_invalid_poll_interval_errors(self, tmp_path: pathlib.Path) -> None:
98 root, repo_id = _init_repo(tmp_path)
99 result = _run(root, "reflog", "watch", "--once", "--poll-interval", "0", "--json")
100 assert result.exit_code == 1
101
102 def test_max_events_caps_snapshot(self, tmp_path: pathlib.Path) -> None:
103 root, repo_id = _init_repo(tmp_path)
104 _commit_file(root, "v1\n", "c1")
105 _commit_file(root, "v2\n", "c2")
106 _commit_file(root, "v3\n", "c3")
107
108 result = _run(root, "reflog", "watch", "--once", "--max-events", "2", "--json")
109 assert result.exit_code == 0, result.output
110 events = _ndjson_lines(result.output)
111 assert len(events) == 2
112
113
114 class TestReflogWatchAppended:
115 def test_appended_event_detected_during_poll(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
116 root, repo_id = _init_repo(tmp_path)
117 c1 = _commit_file(root, "v1\n", "c1")
118 c2 = "2" * 64 # fabricated commit id -- no need for a real commit object
119
120 import muse.cli.commands.reflog as reflog_cmd
121
122 head_log = logs_dir(root) / "HEAD"
123 call_count = {"n": 0}
124
125 def _fake_sleep(seconds: float) -> None:
126 call_count["n"] += 1
127 if call_count["n"] == 1:
128 # Simulate a second entry landing while we're "sleeping" --
129 # append directly to the log file, no nested CLI invocation.
130 with head_log.open("a", encoding="utf-8") as f:
131 f.write(f"{c1} {c2} gabriel 1800000000 +0000\tcommit: c2\n")
132
133 monkeypatch.setattr(reflog_cmd.time, "sleep", _fake_sleep)
134
135 result = _run(root, "reflog", "watch", "--max-events", "2", "--json")
136 assert result.exit_code == 0, result.output
137 events = _ndjson_lines(result.output)
138 assert len(events) == 2
139 assert events[0]["event_type"] == "snapshot"
140 assert events[0]["new_id"] == c1
141 assert events[1]["event_type"] == "appended"
142 assert events[1]["new_id"] == f"sha256:{c2}"
143 assert events[1]["operation"] == "commit: c2"
File History 1 commit
sha256:799dc6f15b7dba389fdf5f088cd5e210da5fb66ec4b1ec9db7b16cde0e8fe734 feat: muse#54 item 10 — muse reflog watch live-tail Sonnet 5 patch 11 hours ago