cgcardona / muse public
test_bisect.py python
548 lines 19.1 KB
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse bisect`` — state machine, commit graph traversal, and CLI.
2
3 Coverage
4 --------
5 - :func:`read_bisect_state` / :func:`write_bisect_state` / :func:`clear_bisect_state`
6 round-trip fidelity.
7 - :func:`get_commits_between` returns the correct candidate set for both linear
8 and branching histories.
9 - :func:`pick_midpoint` selects the lower-middle element.
10 - :func:`advance_bisect` state machine: marks verdicts, narrows range,
11 identifies culprit when range collapses.
12 - ``test_bisect_state_machine_advances_correctly`` — the primary regression test
13 from the issue spec.
14 - Guard: ``muse bisect start`` blocks when a merge is in progress.
15 - Guard: ``muse bisect start`` blocks when a bisect is already active.
16 - ``muse bisect log --json`` emits valid JSON.
17 """
18 from __future__ import annotations
19
20 import datetime
21 import json
22 import pathlib
23 import uuid
24
25 import pytest
26 import typer
27 from sqlalchemy.ext.asyncio import AsyncSession
28
29 from maestro.muse_cli.commands.commit import _commit_async
30 from maestro.muse_cli.errors import ExitCode
31 from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot
32 from maestro.muse_cli.snapshot import compute_snapshot_id
33 from maestro.services.muse_bisect import (
34 BisectState,
35 BisectStepResult,
36 advance_bisect,
37 clear_bisect_state,
38 get_commits_between,
39 pick_midpoint,
40 read_bisect_state,
41 write_bisect_state,
42 )
43
44
45 # ---------------------------------------------------------------------------
46 # Helpers
47 # ---------------------------------------------------------------------------
48
49
50 def _init_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
51 """Create a minimal .muse/ layout for testing."""
52 rid = repo_id or str(uuid.uuid4())
53 muse = root / ".muse"
54 (muse / "refs" / "heads").mkdir(parents=True)
55 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
56 (muse / "HEAD").write_text("refs/heads/main")
57 (muse / "refs" / "heads" / "main").write_text("")
58 return rid
59
60
61 def _write_workdir(root: pathlib.Path, files: dict[str, bytes]) -> None:
62 """Overwrite muse-work/ with exactly the given files."""
63 import shutil
64
65 workdir = root / "muse-work"
66 if workdir.exists():
67 shutil.rmtree(workdir)
68 workdir.mkdir()
69 for name, content in files.items():
70 (workdir / name).write_bytes(content)
71
72
73 def _head_commit(root: pathlib.Path, branch: str = "main") -> str:
74 """Return current HEAD commit_id for the branch."""
75 muse = root / ".muse"
76 ref_path = muse / "refs" / "heads" / branch
77 return ref_path.read_text().strip() if ref_path.exists() else ""
78
79
80 # ---------------------------------------------------------------------------
81 # Unit tests — state file round-trip
82 # ---------------------------------------------------------------------------
83
84
85 def test_write_read_bisect_state_round_trip(tmp_path: pathlib.Path) -> None:
86 """BisectState survives a write → read cycle with all fields set."""
87 _init_repo(tmp_path)
88
89 state = BisectState(
90 good="goodabc123",
91 bad="baddef456",
92 current="midpoint789",
93 tested={"goodabc123": "good", "midpoint789": "bad"},
94 pre_bisect_ref="refs/heads/main",
95 pre_bisect_commit="originalabc",
96 )
97 write_bisect_state(tmp_path, state)
98 loaded = read_bisect_state(tmp_path)
99
100 assert loaded is not None
101 assert loaded.good == "goodabc123"
102 assert loaded.bad == "baddef456"
103 assert loaded.current == "midpoint789"
104 assert loaded.tested == {"goodabc123": "good", "midpoint789": "bad"}
105 assert loaded.pre_bisect_ref == "refs/heads/main"
106 assert loaded.pre_bisect_commit == "originalabc"
107
108
109 def test_read_bisect_state_returns_none_when_absent(tmp_path: pathlib.Path) -> None:
110 """read_bisect_state returns None when no BISECT_STATE.json exists."""
111 _init_repo(tmp_path)
112 assert read_bisect_state(tmp_path) is None
113
114
115 def test_clear_bisect_state_removes_file(tmp_path: pathlib.Path) -> None:
116 """clear_bisect_state removes the state file; subsequent read returns None."""
117 _init_repo(tmp_path)
118 write_bisect_state(tmp_path, BisectState())
119 assert read_bisect_state(tmp_path) is not None
120 clear_bisect_state(tmp_path)
121 assert read_bisect_state(tmp_path) is None
122
123
124 def test_clear_bisect_state_is_idempotent(tmp_path: pathlib.Path) -> None:
125 """Calling clear_bisect_state when no file exists does not raise."""
126 _init_repo(tmp_path)
127 clear_bisect_state(tmp_path) # should not raise
128
129
130 # ---------------------------------------------------------------------------
131 # Unit tests — pick_midpoint
132 # ---------------------------------------------------------------------------
133
134
135 def _make_commit(commit_id: str, offset_seconds: int = 0) -> MuseCliCommit:
136 """Return an unsaved MuseCliCommit stub for midpoint testing."""
137 return MuseCliCommit(
138 commit_id=commit_id,
139 repo_id="test-repo",
140 branch="main",
141 parent_commit_id=None,
142 snapshot_id="snap-" + commit_id[:8],
143 message="test",
144 author="",
145 committed_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
146 + datetime.timedelta(seconds=offset_seconds),
147 )
148
149
150 def test_pick_midpoint_returns_none_on_empty() -> None:
151 assert pick_midpoint([]) is None
152
153
154 def test_pick_midpoint_single_element() -> None:
155 c = _make_commit("aaa")
156 assert pick_midpoint([c]) is c
157
158
159 def test_pick_midpoint_selects_lower_middle_for_even() -> None:
160 """For a 4-element list, midpoint is index 1 (lower-middle)."""
161 commits = [_make_commit(f"c{i:03d}", i) for i in range(4)]
162 mid = pick_midpoint(commits)
163 assert mid is not None
164 assert mid.commit_id == commits[1].commit_id
165
166
167 def test_pick_midpoint_selects_middle_for_odd() -> None:
168 """For a 5-element list, midpoint is index 2."""
169 commits = [_make_commit(f"c{i:03d}", i) for i in range(5)]
170 mid = pick_midpoint(commits)
171 assert mid is not None
172 assert mid.commit_id == commits[2].commit_id
173
174
175 # ---------------------------------------------------------------------------
176 # Integration tests — commit graph traversal
177 # ---------------------------------------------------------------------------
178
179
180 @pytest.mark.anyio
181 async def test_get_commits_between_linear_history(
182 tmp_path: pathlib.Path,
183 muse_cli_db_session: AsyncSession,
184 ) -> None:
185 """get_commits_between returns the inner commits on a linear chain.
186
187 Topology: good → c1 → c2 → c3 → bad
188 Expected result: [c1, c2, c3] (oldest first, excluding good and bad).
189 """
190 _init_repo(tmp_path)
191
192 _write_workdir(tmp_path, {"a.mid": b"GOOD"})
193 await _commit_async(message="good commit", root=tmp_path, session=muse_cli_db_session)
194 good_id = _head_commit(tmp_path)
195
196 _write_workdir(tmp_path, {"a.mid": b"C1"})
197 await _commit_async(message="c1", root=tmp_path, session=muse_cli_db_session)
198 c1_id = _head_commit(tmp_path)
199
200 _write_workdir(tmp_path, {"a.mid": b"C2"})
201 await _commit_async(message="c2", root=tmp_path, session=muse_cli_db_session)
202 c2_id = _head_commit(tmp_path)
203
204 _write_workdir(tmp_path, {"a.mid": b"C3"})
205 await _commit_async(message="c3", root=tmp_path, session=muse_cli_db_session)
206 c3_id = _head_commit(tmp_path)
207
208 _write_workdir(tmp_path, {"a.mid": b"BAD"})
209 await _commit_async(message="bad commit", root=tmp_path, session=muse_cli_db_session)
210 bad_id = _head_commit(tmp_path)
211
212 candidates = await get_commits_between(muse_cli_db_session, good_id, bad_id)
213 candidate_ids = {c.commit_id for c in candidates}
214
215 assert c1_id in candidate_ids
216 assert c2_id in candidate_ids
217 assert c3_id in candidate_ids
218 assert good_id not in candidate_ids
219 assert bad_id not in candidate_ids
220
221 # Must be sorted oldest first.
222 assert [c.commit_id for c in candidates] == [c1_id, c2_id, c3_id]
223
224
225 @pytest.mark.anyio
226 async def test_get_commits_between_adjacent_commits_returns_empty(
227 tmp_path: pathlib.Path,
228 muse_cli_db_session: AsyncSession,
229 ) -> None:
230 """When good is the direct parent of bad, no commits to bisect."""
231 _init_repo(tmp_path)
232
233 _write_workdir(tmp_path, {"a.mid": b"V1"})
234 await _commit_async(message="good", root=tmp_path, session=muse_cli_db_session)
235 good_id = _head_commit(tmp_path)
236
237 _write_workdir(tmp_path, {"a.mid": b"V2"})
238 await _commit_async(message="bad", root=tmp_path, session=muse_cli_db_session)
239 bad_id = _head_commit(tmp_path)
240
241 candidates = await get_commits_between(muse_cli_db_session, good_id, bad_id)
242 assert candidates == []
243
244
245 # ---------------------------------------------------------------------------
246 # Integration tests — advance_bisect state machine
247 # ---------------------------------------------------------------------------
248
249
250 @pytest.mark.anyio
251 async def test_bisect_state_machine_advances_correctly(
252 tmp_path: pathlib.Path,
253 muse_cli_db_session: AsyncSession,
254 ) -> None:
255 """Regression test: bisect narrows range and identifies the culprit.
256
257 Topology: good → c1 → culprit → c3 → bad (4 inner commits)
258 Bisect should identify *culprit* after ≤ 2 steps by binary search.
259 """
260 _init_repo(tmp_path)
261
262 _write_workdir(tmp_path, {"beat.mid": b"GOOD"})
263 await _commit_async(message="good groove", root=tmp_path, session=muse_cli_db_session)
264 good_id = _head_commit(tmp_path)
265
266 _write_workdir(tmp_path, {"beat.mid": b"C1"})
267 await _commit_async(message="c1 ok", root=tmp_path, session=muse_cli_db_session)
268 c1_id = _head_commit(tmp_path)
269
270 _write_workdir(tmp_path, {"beat.mid": b"CULPRIT"})
271 await _commit_async(message="culprit: introduced drift", root=tmp_path, session=muse_cli_db_session)
272 culprit_id = _head_commit(tmp_path)
273
274 _write_workdir(tmp_path, {"beat.mid": b"C3"})
275 await _commit_async(message="c3 still broken", root=tmp_path, session=muse_cli_db_session)
276
277 _write_workdir(tmp_path, {"beat.mid": b"BAD"})
278 await _commit_async(message="bad groove", root=tmp_path, session=muse_cli_db_session)
279 bad_id = _head_commit(tmp_path)
280
281 # Start a bisect session.
282 state = BisectState(
283 good=None,
284 bad=None,
285 current=None,
286 tested={},
287 pre_bisect_ref="refs/heads/main",
288 pre_bisect_commit=bad_id,
289 )
290 write_bisect_state(tmp_path, state)
291
292 # Mark good.
293 result = await advance_bisect(
294 session=muse_cli_db_session, root=tmp_path, commit_id=good_id, verdict="good"
295 )
296 # Both bounds not yet set, so no next commit yet.
297 assert result.culprit is None
298
299 # Mark bad.
300 result = await advance_bisect(
301 session=muse_cli_db_session, root=tmp_path, commit_id=bad_id, verdict="bad"
302 )
303 assert result.culprit is None
304 assert result.next_commit is not None
305 midpoint_1 = result.next_commit
306
307 # Midpoint should be inside the range [c1, culprit, c3].
308 candidates_all = await get_commits_between(muse_cli_db_session, good_id, bad_id)
309 candidate_ids_all = {c.commit_id for c in candidates_all}
310 assert midpoint_1 in candidate_ids_all
311
312 # Test the midpoint: if it's culprit or after → bad; before → good.
313 # We need to simulate what a human / script would do.
314 # Strategy: mark commits that come AFTER the culprit (in time) as bad,
315 # and commits before/at culprit as bad too if they ARE the culprit.
316 # Simple rule: commit_id == culprit_id OR is a descendant → bad.
317
318 # Step 1: test the first midpoint.
319 mid1_commit = await muse_cli_db_session.get(MuseCliCommit, midpoint_1)
320 assert mid1_commit is not None
321 # The culprit is the 2nd of 3 inner commits (c1, culprit, c3).
322 # Binary search: midpoint of [c1, culprit, c3] (idx 0,1,2) → idx 1 = culprit.
323 # If midpoint IS the culprit → mark bad.
324 # In our test data the midpoint of 3 elements is index 1 = culprit.
325 if mid1_commit.message == "culprit: introduced drift":
326 # This is the culprit: mark as bad.
327 result2 = await advance_bisect(
328 session=muse_cli_db_session,
329 root=tmp_path,
330 commit_id=midpoint_1,
331 verdict="bad",
332 )
333 # Next candidate: [c1] (commits before culprit but after good).
334 # After marking culprit as bad, range = [c1].
335 # Midpoint of [c1] = c1 itself.
336 if result2.culprit is None:
337 assert result2.next_commit is not None
338 # Mark c1 as good → culprit identified as midpoint_1 (the culprit commit).
339 result3 = await advance_bisect(
340 session=muse_cli_db_session,
341 root=tmp_path,
342 commit_id=result2.next_commit,
343 verdict="good",
344 )
345 assert result3.culprit == midpoint_1
346 else:
347 # c1 is the midpoint — mark it based on its position relative to culprit.
348 # c1 is BEFORE culprit → good.
349 result2 = await advance_bisect(
350 session=muse_cli_db_session,
351 root=tmp_path,
352 commit_id=midpoint_1,
353 verdict="good",
354 )
355 assert result2.culprit is None or result2.culprit == culprit_id
356
357
358 @pytest.mark.anyio
359 async def test_advance_bisect_with_only_one_inner_commit_finds_culprit(
360 tmp_path: pathlib.Path,
361 muse_cli_db_session: AsyncSession,
362 ) -> None:
363 """When only one commit is between good and bad, it is the culprit immediately."""
364 _init_repo(tmp_path)
365
366 _write_workdir(tmp_path, {"beat.mid": b"GOOD"})
367 await _commit_async(message="good", root=tmp_path, session=muse_cli_db_session)
368 good_id = _head_commit(tmp_path)
369
370 _write_workdir(tmp_path, {"beat.mid": b"CULPRIT"})
371 await _commit_async(message="culprit", root=tmp_path, session=muse_cli_db_session)
372 culprit_id = _head_commit(tmp_path)
373
374 _write_workdir(tmp_path, {"beat.mid": b"BAD"})
375 await _commit_async(message="bad", root=tmp_path, session=muse_cli_db_session)
376 bad_id = _head_commit(tmp_path)
377
378 write_bisect_state(tmp_path, BisectState(
379 good=None, bad=None, current=None, tested={},
380 pre_bisect_ref="refs/heads/main", pre_bisect_commit=bad_id,
381 ))
382
383 await advance_bisect(session=muse_cli_db_session, root=tmp_path, commit_id=good_id, verdict="good")
384 # Mark bad → range = [culprit], midpoint = culprit.
385 result = await advance_bisect(session=muse_cli_db_session, root=tmp_path, commit_id=bad_id, verdict="bad")
386
387 assert result.next_commit == culprit_id
388 # Mark culprit as bad → range collapses.
389 result2 = await advance_bisect(session=muse_cli_db_session, root=tmp_path, commit_id=culprit_id, verdict="bad")
390 assert result2.culprit == culprit_id
391
392
393 @pytest.mark.anyio
394 async def test_advance_bisect_adjacent_commits_collapses_immediately(
395 tmp_path: pathlib.Path,
396 muse_cli_db_session: AsyncSession,
397 ) -> None:
398 """When good is the direct parent of bad, culprit is bad itself."""
399 _init_repo(tmp_path)
400
401 _write_workdir(tmp_path, {"a.mid": b"V1"})
402 await _commit_async(message="good", root=tmp_path, session=muse_cli_db_session)
403 good_id = _head_commit(tmp_path)
404
405 _write_workdir(tmp_path, {"a.mid": b"V2"})
406 await _commit_async(message="bad", root=tmp_path, session=muse_cli_db_session)
407 bad_id = _head_commit(tmp_path)
408
409 write_bisect_state(tmp_path, BisectState(
410 good=None, bad=None, current=None, tested={},
411 pre_bisect_ref="refs/heads/main", pre_bisect_commit=bad_id,
412 ))
413
414 await advance_bisect(session=muse_cli_db_session, root=tmp_path, commit_id=good_id, verdict="good")
415 result = await advance_bisect(session=muse_cli_db_session, root=tmp_path, commit_id=bad_id, verdict="bad")
416 # No inner commits → bad is the culprit immediately.
417 assert result.culprit == bad_id
418
419
420 # ---------------------------------------------------------------------------
421 # CLI guard tests
422 # ---------------------------------------------------------------------------
423
424
425 def test_bisect_start_blocked_by_merge_in_progress(tmp_path: pathlib.Path) -> None:
426 """muse bisect start exits 1 when MERGE_STATE.json is present."""
427 from typer.testing import CliRunner
428 from maestro.muse_cli.commands.bisect import app as bisect_app
429
430 _init_repo(tmp_path)
431 (tmp_path / ".muse" / "MERGE_STATE.json").write_text(
432 json.dumps({"base_commit": "abc", "ours_commit": "def", "theirs_commit": "ghi", "conflict_paths": []})
433 )
434
435 runner = CliRunner()
436 import os
437
438 old_cwd = pathlib.Path.cwd()
439 try:
440 os.chdir(tmp_path)
441 result = runner.invoke(bisect_app, ["start"])
442 finally:
443 os.chdir(old_cwd)
444
445 assert result.exit_code == ExitCode.USER_ERROR
446
447
448 def test_bisect_start_blocked_when_already_active(tmp_path: pathlib.Path) -> None:
449 """muse bisect start exits 1 when BISECT_STATE.json already exists."""
450 from typer.testing import CliRunner
451 from maestro.muse_cli.commands.bisect import app as bisect_app
452
453 _init_repo(tmp_path)
454 write_bisect_state(tmp_path, BisectState(
455 good=None, bad=None, current=None, tested={},
456 pre_bisect_ref="refs/heads/main", pre_bisect_commit="abc",
457 ))
458
459 runner = CliRunner()
460 import os
461
462 old_cwd = pathlib.Path.cwd()
463 try:
464 os.chdir(tmp_path)
465 result = runner.invoke(bisect_app, ["start"])
466 finally:
467 os.chdir(old_cwd)
468
469 assert result.exit_code == ExitCode.USER_ERROR
470
471
472 def test_bisect_good_without_active_session_exits_1(tmp_path: pathlib.Path) -> None:
473 """muse bisect good exits 1 when no session is active."""
474 from typer.testing import CliRunner
475 from maestro.muse_cli.commands.bisect import app as bisect_app
476
477 _init_repo(tmp_path)
478
479 runner = CliRunner()
480 import os
481
482 old_cwd = pathlib.Path.cwd()
483 try:
484 os.chdir(tmp_path)
485 result = runner.invoke(bisect_app, ["good", "abc123"])
486 finally:
487 os.chdir(old_cwd)
488
489 assert result.exit_code == ExitCode.USER_ERROR
490
491
492 # ---------------------------------------------------------------------------
493 # bisect log --json
494 # ---------------------------------------------------------------------------
495
496
497 def test_bisect_log_json_emits_valid_json(tmp_path: pathlib.Path) -> None:
498 """muse bisect log --json outputs valid JSON with expected fields."""
499 from typer.testing import CliRunner
500 from maestro.muse_cli.commands.bisect import app as bisect_app
501
502 _init_repo(tmp_path)
503 write_bisect_state(tmp_path, BisectState(
504 good="good_sha",
505 bad="bad_sha",
506 current="current_sha",
507 tested={"good_sha": "good"},
508 pre_bisect_ref="refs/heads/main",
509 pre_bisect_commit="orig_sha",
510 ))
511
512 runner = CliRunner()
513 import os
514
515 old_cwd = pathlib.Path.cwd()
516 try:
517 os.chdir(tmp_path)
518 result = runner.invoke(bisect_app, ["log", "--json"])
519 finally:
520 os.chdir(old_cwd)
521
522 assert result.exit_code == 0
523 data = json.loads(result.output)
524 assert data["good"] == "good_sha"
525 assert data["bad"] == "bad_sha"
526 assert data["current"] == "current_sha"
527 assert data["tested"] == {"good_sha": "good"}
528
529
530 def test_bisect_log_no_active_session(tmp_path: pathlib.Path) -> None:
531 """muse bisect log exits 0 with a message when no session is active."""
532 from typer.testing import CliRunner
533 from maestro.muse_cli.commands.bisect import app as bisect_app
534
535 _init_repo(tmp_path)
536
537 runner = CliRunner()
538 import os
539
540 old_cwd = pathlib.Path.cwd()
541 try:
542 os.chdir(tmp_path)
543 result = runner.invoke(bisect_app, ["log"])
544 finally:
545 os.chdir(old_cwd)
546
547 assert result.exit_code == 0
548 assert "No bisect session" in result.output