cgcardona / muse public
test_muse_session.py python
407 lines 14.7 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse session`` — recording session metadata stored as JSON files.
2
3 All tests are purely file-based (no DB, no async) and use ``tmp_path`` for
4 isolation. The ``MUSE_REPO_ROOT`` env-var override keeps tests free of
5 ``os.chdir`` calls so they run safely in parallel.
6 """
7 from __future__ import annotations
8
9 import json
10 import os
11 import pathlib
12 import uuid
13
14 import pytest
15 from click.testing import Result
16 from typer.testing import CliRunner
17
18 from maestro.muse_cli.app import cli
19 from maestro.muse_cli.commands.session import (
20 _load_completed_sessions,
21 _sessions_dir,
22 app as session_app,
23 )
24 from maestro.muse_cli.errors import ExitCode
25
26 runner = CliRunner()
27
28
29 # ---------------------------------------------------------------------------
30 # Fixtures / helpers
31 # ---------------------------------------------------------------------------
32
33
34 def _init_repo(root: pathlib.Path) -> pathlib.Path:
35 """Create a minimal .muse/ layout and return the repo root."""
36 muse = root / ".muse"
37 (muse / "refs" / "heads").mkdir(parents=True)
38 (muse / "repo.json").write_text(
39 json.dumps({"repo_id": str(uuid.uuid4()), "schema_version": "1"})
40 )
41 (muse / "HEAD").write_text("refs/heads/main\n")
42 (muse / "refs" / "heads" / "main").write_text("")
43 return root
44
45
46 def _invoke(args: list[str], repo_root: pathlib.Path) -> Result:
47 """Invoke the top-level CLI with MUSE_REPO_ROOT set to *repo_root*."""
48 env = {**os.environ, "MUSE_REPO_ROOT": str(repo_root)}
49 return runner.invoke(cli, args, env=env)
50
51
52 def _invoke_session(args: list[str], repo_root: pathlib.Path) -> Result:
53 """Invoke the session sub-app with MUSE_REPO_ROOT set."""
54 env = {**os.environ, "MUSE_REPO_ROOT": str(repo_root)}
55 return runner.invoke(session_app, args, env=env)
56
57
58 # ---------------------------------------------------------------------------
59 # muse session start
60 # ---------------------------------------------------------------------------
61
62
63 def test_session_start_creates_current_json(tmp_path: pathlib.Path) -> None:
64 """``muse session start`` writes current.json with required fields."""
65 _init_repo(tmp_path)
66 result = _invoke(["session", "start"], tmp_path)
67
68 assert result.exit_code == 0, result.output
69 current = tmp_path / ".muse" / "sessions" / "current.json"
70 assert current.exists(), "current.json was not created"
71 data = json.loads(current.read_text())
72 assert "session_id" in data
73 assert "started_at" in data
74 assert data["ended_at"] is None
75 assert isinstance(data["participants"], list)
76 assert "✅" in result.output
77
78
79 def test_session_start_with_options(tmp_path: pathlib.Path) -> None:
80 """``muse session start`` captures participants, location, and intent."""
81 _init_repo(tmp_path)
82 result = _invoke(
83 [
84 "session",
85 "start",
86 "--participants",
87 "Alice,Bob",
88 "--location",
89 "Studio A",
90 "--intent",
91 "Record the bridge",
92 ],
93 tmp_path,
94 )
95
96 assert result.exit_code == 0, result.output
97 current = tmp_path / ".muse" / "sessions" / "current.json"
98 data = json.loads(current.read_text())
99 assert data["participants"] == ["Alice", "Bob"]
100 assert data["location"] == "Studio A"
101 assert data["intent"] == "Record the bridge"
102
103
104 def test_session_start_rejects_duplicate_session(tmp_path: pathlib.Path) -> None:
105 """Starting a second session while one is active exits with USER_ERROR."""
106 _init_repo(tmp_path)
107 _invoke(["session", "start"], tmp_path)
108 result = _invoke(["session", "start"], tmp_path)
109
110 assert result.exit_code == int(ExitCode.USER_ERROR)
111 assert "already active" in result.output.lower()
112
113
114 def test_session_start_no_repo_exits_2(tmp_path: pathlib.Path) -> None:
115 """``muse session start`` outside a repo exits 2."""
116 env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)}
117 result = runner.invoke(cli, ["session", "start"], env=env)
118 assert result.exit_code == int(ExitCode.REPO_NOT_FOUND)
119
120
121 # ---------------------------------------------------------------------------
122 # muse session end
123 # ---------------------------------------------------------------------------
124
125
126 def test_session_end_finalises_session(tmp_path: pathlib.Path) -> None:
127 """``muse session end`` moves current.json to <uuid>.json with ended_at set."""
128 _init_repo(tmp_path)
129 _invoke(["session", "start"], tmp_path)
130
131 current = tmp_path / ".muse" / "sessions" / "current.json"
132 session_id = json.loads(current.read_text())["session_id"]
133
134 result = _invoke(["session", "end"], tmp_path)
135 assert result.exit_code == 0, result.output
136
137 assert not current.exists(), "current.json should be removed after end"
138 final = tmp_path / ".muse" / "sessions" / f"{session_id}.json"
139 assert final.exists(), f"{session_id}.json was not created"
140 data = json.loads(final.read_text())
141 assert data["ended_at"] is not None
142 assert "✅" in result.output
143
144
145 def test_session_end_with_notes(tmp_path: pathlib.Path) -> None:
146 """``muse session end --notes`` saves closing notes in the session file."""
147 _init_repo(tmp_path)
148 _invoke(["session", "start"], tmp_path)
149
150 result = _invoke(["session", "end", "--notes", "Great take on measure 8"], tmp_path)
151 assert result.exit_code == 0, result.output
152
153 sessions_dir = tmp_path / ".muse" / "sessions"
154 finals = [p for p in sessions_dir.glob("*.json") if p.name != "current.json"]
155 assert len(finals) == 1
156 data = json.loads(finals[0].read_text())
157 assert data["notes"] == "Great take on measure 8"
158
159
160 def test_session_end_without_active_session(tmp_path: pathlib.Path) -> None:
161 """``muse session end`` with no active session exits with USER_ERROR."""
162 _init_repo(tmp_path)
163 result = _invoke(["session", "end"], tmp_path)
164
165 assert result.exit_code == int(ExitCode.USER_ERROR)
166 assert "no active session" in result.output.lower()
167
168
169 # ---------------------------------------------------------------------------
170 # muse session log
171 # ---------------------------------------------------------------------------
172
173
174 def test_session_log_empty(tmp_path: pathlib.Path) -> None:
175 """``muse session log`` reports no sessions when none exist."""
176 _init_repo(tmp_path)
177 result = _invoke(["session", "log"], tmp_path)
178
179 assert result.exit_code == 0, result.output
180 assert "no completed sessions" in result.output.lower()
181
182
183 def test_session_log_lists_sessions_newest_first(tmp_path: pathlib.Path) -> None:
184 """``muse session log`` lists sessions newest-first by started_at."""
185 _init_repo(tmp_path)
186
187 # Create two sessions back-to-back
188 _invoke(["session", "start", "--participants", "Alice"], tmp_path)
189 _invoke(["session", "end"], tmp_path)
190 _invoke(["session", "start", "--participants", "Bob"], tmp_path)
191 _invoke(["session", "end"], tmp_path)
192
193 result = _invoke(["session", "log"], tmp_path)
194 assert result.exit_code == 0, result.output
195 lines = [line for line in result.output.splitlines() if line.strip()]
196 assert len(lines) == 2
197 # Newest session should appear first — parse first column (8-char UUID prefix)
198 sessions_dir = tmp_path / ".muse" / "sessions"
199 all_sessions = _load_completed_sessions(sessions_dir)
200 assert str(all_sessions[0].get("started_at", "")) >= str(
201 all_sessions[1].get("started_at", "")
202 )
203
204
205 def test_session_log_skips_active_current(tmp_path: pathlib.Path) -> None:
206 """``muse session log`` does not list the active current.json."""
207 _init_repo(tmp_path)
208 _invoke(["session", "start"], tmp_path)
209
210 result = _invoke(["session", "log"], tmp_path)
211 assert result.exit_code == 0, result.output
212 assert "no completed sessions" in result.output.lower()
213
214
215 # ---------------------------------------------------------------------------
216 # muse session show
217 # ---------------------------------------------------------------------------
218
219
220 def test_session_show_full_json(tmp_path: pathlib.Path) -> None:
221 """``muse session show <id>`` prints the full JSON for that session."""
222 _init_repo(tmp_path)
223 _invoke(["session", "start", "--participants", "Carol"], tmp_path)
224 current = tmp_path / ".muse" / "sessions" / "current.json"
225 session_id = json.loads(current.read_text())["session_id"]
226 _invoke(["session", "end"], tmp_path)
227
228 result = _invoke(["session", "show", session_id], tmp_path)
229 assert result.exit_code == 0, result.output
230 parsed = json.loads(result.output)
231 assert parsed["session_id"] == session_id
232 assert parsed["participants"] == ["Carol"]
233
234
235 def test_session_show_by_prefix(tmp_path: pathlib.Path) -> None:
236 """``muse session show`` accepts a unique prefix of the session ID."""
237 _init_repo(tmp_path)
238 _invoke(["session", "start"], tmp_path)
239 current = tmp_path / ".muse" / "sessions" / "current.json"
240 session_id = json.loads(current.read_text())["session_id"]
241 _invoke(["session", "end"], tmp_path)
242
243 prefix = session_id[:8]
244 result = _invoke(["session", "show", prefix], tmp_path)
245 assert result.exit_code == 0, result.output
246 parsed = json.loads(result.output)
247 assert parsed["session_id"] == session_id
248
249
250 def test_session_show_unknown_id(tmp_path: pathlib.Path) -> None:
251 """``muse session show`` with an unknown ID exits with USER_ERROR."""
252 _init_repo(tmp_path)
253 result = _invoke(["session", "show", "nonexistent-id"], tmp_path)
254 assert result.exit_code == int(ExitCode.USER_ERROR)
255 assert "no session found" in result.output.lower()
256
257
258 def test_session_show_ambiguous_prefix(tmp_path: pathlib.Path) -> None:
259 """``muse session show`` with an ambiguous prefix exits with USER_ERROR."""
260 _init_repo(tmp_path)
261 sessions_dir = tmp_path / ".muse" / "sessions"
262 sessions_dir.mkdir(parents=True, exist_ok=True)
263 # Write two sessions whose IDs share a common prefix
264 for suffix in ("aaaa-bbbb", "aaaa-cccc"):
265 fake_id = f"00000000-{suffix}-0000-000000000000"
266 sessions_dir.joinpath(f"{fake_id}.json").write_text(
267 json.dumps(
268 {
269 "session_id": fake_id,
270 "started_at": "2024-01-01T00:00:00+00:00",
271 "ended_at": "2024-01-01T01:00:00+00:00",
272 "participants": [],
273 "location": "",
274 "intent": "",
275 "commits": [],
276 "notes": "",
277 }
278 )
279 )
280
281 result = _invoke(["session", "show", "00000000"], tmp_path)
282 assert result.exit_code == int(ExitCode.USER_ERROR)
283 assert "ambiguous" in result.output.lower()
284
285
286 # ---------------------------------------------------------------------------
287 # muse session credits
288 # ---------------------------------------------------------------------------
289
290
291 def test_session_credits_aggregates_participants(tmp_path: pathlib.Path) -> None:
292 """``muse session credits`` lists all participants with session counts."""
293 _init_repo(tmp_path)
294
295 # Session 1: Alice + Bob
296 _invoke(["session", "start", "--participants", "Alice,Bob"], tmp_path)
297 _invoke(["session", "end"], tmp_path)
298 # Session 2: Alice + Carol
299 _invoke(["session", "start", "--participants", "Alice,Carol"], tmp_path)
300 _invoke(["session", "end"], tmp_path)
301
302 result = _invoke(["session", "credits"], tmp_path)
303 assert result.exit_code == 0, result.output
304 assert "Alice" in result.output
305 assert "Bob" in result.output
306 assert "Carol" in result.output
307 # Alice appeared in 2 sessions — highest count
308 lines = result.output.splitlines()
309 alice_line = next(l for l in lines if "Alice" in l)
310 assert "2" in alice_line
311
312
313 def test_session_credits_no_participants(tmp_path: pathlib.Path) -> None:
314 """``muse session credits`` with no sessions reports no participants."""
315 _init_repo(tmp_path)
316 result = _invoke(["session", "credits"], tmp_path)
317
318 assert result.exit_code == 0, result.output
319 assert "no participants" in result.output.lower()
320
321
322 def test_session_credits_sorted_by_count_desc(tmp_path: pathlib.Path) -> None:
323 """Credits output is sorted by session count descending."""
324 _init_repo(tmp_path)
325 # Dave: 1 session, Eve: 2 sessions
326 _invoke(["session", "start", "--participants", "Dave,Eve"], tmp_path)
327 _invoke(["session", "end"], tmp_path)
328 _invoke(["session", "start", "--participants", "Eve"], tmp_path)
329 _invoke(["session", "end"], tmp_path)
330
331 result = _invoke(["session", "credits"], tmp_path)
332 assert result.exit_code == 0, result.output
333 lines = [l for l in result.output.splitlines() if "Eve" in l or "Dave" in l]
334 assert lines[0].find("Eve") != -1 or "2" in lines[0], (
335 "Eve should appear before Dave (higher count)"
336 )
337
338
339 # ---------------------------------------------------------------------------
340 # Schema validation
341 # ---------------------------------------------------------------------------
342
343
344 def test_session_json_schema(tmp_path: pathlib.Path) -> None:
345 """Completed session JSON contains all required schema fields."""
346 _init_repo(tmp_path)
347 _invoke(
348 [
349 "session",
350 "start",
351 "--participants",
352 "Frank",
353 "--location",
354 "Room 101",
355 "--intent",
356 "Groove track",
357 ],
358 tmp_path,
359 )
360 _invoke(["session", "end", "--notes", "Nailed it"], tmp_path)
361
362 sessions_dir = tmp_path / ".muse" / "sessions"
363 finals = list(sessions_dir.glob("*.json"))
364 assert len(finals) == 1
365 data = json.loads(finals[0].read_text())
366
367 required_fields = {
368 "session_id",
369 "started_at",
370 "ended_at",
371 "participants",
372 "location",
373 "intent",
374 "commits",
375 "notes",
376 }
377 missing = required_fields - data.keys()
378 assert not missing, f"Missing fields in session JSON: {missing}"
379 assert data["participants"] == ["Frank"]
380 assert data["location"] == "Room 101"
381 assert data["intent"] == "Groove track"
382 assert data["notes"] == "Nailed it"
383 assert data["commits"] == []
384 assert data["ended_at"] is not None
385
386
387 # ---------------------------------------------------------------------------
388 # Internal helper tests
389 # ---------------------------------------------------------------------------
390
391
392 def test_sessions_dir_created_on_demand(tmp_path: pathlib.Path) -> None:
393 """``_sessions_dir`` creates the directory if it does not exist."""
394 assert not (tmp_path / ".muse" / "sessions").exists()
395 result = _sessions_dir(tmp_path)
396 assert result.is_dir()
397
398
399 def test_load_completed_sessions_excludes_current(tmp_path: pathlib.Path) -> None:
400 """``_load_completed_sessions`` never includes current.json."""
401 sessions_dir = tmp_path / ".muse" / "sessions"
402 sessions_dir.mkdir(parents=True)
403 (sessions_dir / "current.json").write_text(
404 json.dumps({"session_id": "active", "started_at": "2024-01-01T00:00:00+00:00"})
405 )
406 results = _load_completed_sessions(sessions_dir)
407 assert results == []