gabriel / muse public
test_cmd_forecast.py python
1,112 lines 54.0 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago
1 """Tests for ``muse coord forecast``.
2
3 Coverage matrix
4 ---------------
5 Unit
6 ~~~~
7 * :class:`_ConflictPrediction` — construction, ``to_dict``
8 * Pass 1: direct address overlap detection
9 * Pass 2: blast-radius overlap — call graph available path
10 * Pass 2: call graph unavailable — ``(OSError, KeyError, ValueError, AttributeError)``
11 produces a warning, not a silent skip and not a crash
12 * Pass 2: no commits yet → warning emitted
13 * Pass 3: operation conflict detection
14
15 Integration
16 ~~~~~~~~~~~
17 * ``muse coord forecast`` — empty swarm
18 * ``muse coord forecast`` — address_overlap detected
19 * ``muse coord forecast`` — operation_conflict detected
20 * ``muse coord forecast --branch`` filtering
21 * ``muse coord forecast --format json`` — schema complete, all required keys
22 * ``muse coord forecast --json`` — shorthand works
23 * ``muse coord forecast`` text output — warnings visible when call graph absent
24 * JSON: ``call_graph_available`` is ``False`` when index missing
25 * JSON: ``partial_forecast`` true when any pass was skipped
26 * JSON: ``warnings`` non-empty when call graph absent
27 * JSON: ``duration_ms`` present and non-negative
28 * JSON: ``current_branch`` and ``branch_filter`` present
29 * JSON: compact output (no indent=2 newlines)
30 * Text output: released reservations excluded (active_reservations)
31
32 Input validation
33 ~~~~~~~~~~~~~~~~
34 * ``--run-id`` at max length (256): accepted
35 * ``--run-id`` over max length: exits USER_ERROR (1), compact JSON error
36 * ``--min-confidence`` at boundary 0.0: accepted (shows all)
37 * ``--min-confidence`` at boundary 1.0: accepted (shows only certainties)
38 * ``--min-confidence`` = 1.5: exits USER_ERROR (1), compact JSON error
39 * ``--min-confidence`` = -0.1: exits USER_ERROR (1), compact JSON error
40 * validation fires before any file I/O
41
42 --run-id filter
43 ~~~~~~~~~~~~~~~
44 * filters to conflicts involving the named agent
45 * unknown agent returns empty conflicts list
46 * does not affect reservations/intents counts in output
47
48 --min-confidence filter
49 ~~~~~~~~~~~~~~~~~~~~~~~
50 * hides conflicts below threshold
51 * shows conflicts at exactly the threshold
52 * 0.9 threshold: only high-risk conflicts
53 * risk counts (high/medium/low) reflect post-filter list
54
55 Security
56 ~~~~~~~~
57 * ANSI escape sequences in run_id / branch / address sanitized before output
58 * Control characters stripped from agent labels in text output
59 * address_glob is never used for filesystem access
60
61 Stress
62 ~~~~~~
63 * 100 reservations × 100 addresses → Pass 1 in < 2 s
64 * 50 reservations → Pass 2 with mock call graph in < 1 s
65 * 200 intents → Pass 3 in < 1 s
66 * 500 reservations / 500 intents — combined forecast in < 5 s
67 """
68
69 from __future__ import annotations
70
71 import argparse
72 import datetime
73 import json
74 import os
75 import pathlib
76 import time
77 from unittest.mock import MagicMock, patch
78
79 import pytest
80
81 from muse.cli.commands.forecast import _MAX_RUN_ID_LEN
82 from muse.core.types import MsgpackDict, fake_id
83 from muse.core.paths import muse_dir
84 from muse.core.coordination import Reservation
85 from muse.core.errors import ExitCode
86
87
88 # ── Helpers ───────────────────────────────────────────────────────────────────
89
90
91 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
92 import json as _json
93 dot_muse = muse_dir(tmp_path)
94 dot_muse.mkdir()
95 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
96 (dot_muse / "repo.json").write_text(
97 _json.dumps({"repo_id": fake_id("repo"), "name": "test-repo"})
98 )
99 return tmp_path
100
101
102 def _now_utc() -> datetime.datetime:
103 return datetime.datetime.now(datetime.timezone.utc)
104
105
106 def _run_forecast(
107 repo: pathlib.Path,
108 *,
109 branch_filter: str | None = None,
110 run_id_filter: str | None = None,
111 min_confidence: float = 0.0,
112 fmt: str = "json",
113 ) -> tuple[int, MsgpackDict]:
114 """Run the forecast command and return (exit_code, parsed_output)."""
115 from muse.cli.commands.forecast import run as forecast_run
116
117 ns = argparse.Namespace(
118 branch_filter=branch_filter,
119 run_id_filter=run_id_filter,
120 min_confidence=min_confidence,
121 json_out=fmt == "json",
122 )
123 old = os.getcwd()
124 os.chdir(repo)
125 try:
126 forecast_run(ns)
127 return 0, {}
128 except SystemExit as exc:
129 return exc.code, {}
130 finally:
131 os.chdir(old)
132
133
134 def _run_forecast_json(
135 repo: pathlib.Path,
136 *,
137 branch_filter: str | None = None,
138 run_id_filter: str | None = None,
139 min_confidence: float = 0.0,
140 capsys: pytest.CaptureFixture[str],
141 ) -> MsgpackDict:
142 """Run forecast in JSON mode and return the parsed output dict."""
143 _run_forecast(
144 repo,
145 branch_filter=branch_filter,
146 run_id_filter=run_id_filter,
147 min_confidence=min_confidence,
148 fmt="json",
149 )
150 return json.loads(capsys.readouterr().out)
151
152
153 def _run_forecast_text(
154 repo: pathlib.Path,
155 *,
156 branch_filter: str | None = None,
157 run_id_filter: str | None = None,
158 min_confidence: float = 0.0,
159 capsys: pytest.CaptureFixture[str],
160 ) -> str:
161 """Run forecast in text mode and return the captured output."""
162 _run_forecast(
163 repo,
164 branch_filter=branch_filter,
165 run_id_filter=run_id_filter,
166 min_confidence=min_confidence,
167 fmt="text",
168 )
169 return capsys.readouterr().out
170
171
172 # ── Fixtures ──────────────────────────────────────────────────────────────────
173
174
175 @pytest.fixture()
176 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
177 return _make_repo(tmp_path)
178
179
180 @pytest.fixture()
181 def repo_with_conflict(repo: pathlib.Path) -> tuple[pathlib.Path, Reservation, Reservation]:
182 """Two active reservations sharing an address on different branches."""
183 from muse.core.coordination import create_reservation
184 res_a = create_reservation(
185 repo,
186 run_id="agent-41",
187 branch="main",
188 addresses=["src/billing.py::compute_total"],
189 ttl_seconds=3600,
190 )
191 res_b = create_reservation(
192 repo,
193 run_id="agent-42",
194 branch="feature/billing",
195 addresses=["src/billing.py::compute_total"],
196 ttl_seconds=3600,
197 )
198 return repo, res_a, res_b
199
200
201 @pytest.fixture()
202 def repo_with_op_conflict(repo: pathlib.Path) -> pathlib.Path:
203 """One agent intends delete, another intends modify on same address."""
204 from muse.core.coordination import create_reservation, create_intent
205 res_a = create_reservation(
206 repo, run_id="agent-del", branch="main",
207 addresses=["src/api.py::old_endpoint"], ttl_seconds=3600,
208 )
209 res_b = create_reservation(
210 repo, run_id="agent-mod", branch="feat/new",
211 addresses=["src/api.py::old_endpoint"], ttl_seconds=3600,
212 )
213 create_intent(
214 repo, res_a.reservation_id, "agent-del", "main",
215 ["src/api.py::old_endpoint"], "delete", "removing deprecated endpoint",
216 )
217 create_intent(
218 repo, res_b.reservation_id, "agent-mod", "feat/new",
219 ["src/api.py::old_endpoint"], "modify", "extend response shape",
220 )
221 return repo
222
223
224 # ─────────────────────────────────────────────────────────────────────────────
225 # Unit tests — _ConflictPrediction
226 # ─────────────────────────────────────────────────────────────────────────────
227
228
229 class TestConflictPrediction:
230 def test_to_dict_keys(self) -> None:
231 from muse.cli.commands.forecast import _ConflictPrediction
232 c = _ConflictPrediction(
233 conflict_type="address_overlap",
234 addresses=["src/billing.py::compute_total"],
235 agents=["agent-41@main", "agent-42@feat"],
236 confidence=1.0,
237 description="direct overlap",
238 )
239 d = c.to_dict()
240 assert set(d.keys()) == {
241 "conflict_type", "addresses", "agents", "confidence", "description"
242 }
243
244 def test_confidence_rounded(self) -> None:
245 from muse.cli.commands.forecast import _ConflictPrediction
246 c = _ConflictPrediction("x", [], [], 0.749999, "")
247 assert c.to_dict()["confidence"] == 0.75
248
249 def test_all_conflict_types(self) -> None:
250 from muse.cli.commands.forecast import _ConflictPrediction
251 for ctype in ("address_overlap", "blast_radius_overlap", "operation_conflict"):
252 c = _ConflictPrediction(ctype, [], [], 0.9, "desc")
253 assert c.to_dict()["conflict_type"] == ctype
254
255
256 # ─────────────────────────────────────────────────────────────────────────────
257 # Unit tests — Pass 1: address overlap
258 # ─────────────────────────────────────────────────────────────────────────────
259
260
261 class TestPass1AddressOverlap:
262 def test_overlap_detected(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
263 from muse.core.coordination import create_reservation
264 create_reservation(
265 repo, run_id="ag-1", branch="b1",
266 addresses=["src/x.py::fn"], ttl_seconds=3600,
267 )
268 create_reservation(
269 repo, run_id="ag-2", branch="b2",
270 addresses=["src/x.py::fn"], ttl_seconds=3600,
271 )
272 data = _run_forecast_json(repo, capsys=capsys)
273 conflict_types = [c["conflict_type"] for c in data["conflicts"]]
274 assert "address_overlap" in conflict_types
275
276 def test_no_overlap_when_different_addresses(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
277 from muse.core.coordination import create_reservation
278 create_reservation(
279 repo, run_id="ag-1", branch="b", addresses=["x.py::a"], ttl_seconds=3600,
280 )
281 create_reservation(
282 repo, run_id="ag-2", branch="b", addresses=["x.py::b"], ttl_seconds=3600,
283 )
284 data = _run_forecast_json(repo, capsys=capsys)
285 conflict_types = [c["conflict_type"] for c in data["conflicts"]]
286 assert "address_overlap" not in conflict_types
287
288 def test_same_agent_same_address_no_conflict(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
289 from muse.core.coordination import create_reservation
290 create_reservation(
291 repo, run_id="same-agent", branch="b",
292 addresses=["x.py::fn"], ttl_seconds=3600,
293 )
294 data = _run_forecast_json(repo, capsys=capsys)
295 assert data["conflicts"] == []
296
297 def test_overlap_confidence_is_1(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
298 repo, *_ = repo_with_conflict
299 data = _run_forecast_json(repo, capsys=capsys)
300 overlaps = [c for c in data["conflicts"] if c["conflict_type"] == "address_overlap"]
301 assert all(c["confidence"] == 1.0 for c in overlaps)
302
303 def test_multiple_overlaps_all_reported(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
304 from muse.core.coordination import create_reservation
305 for i in range(3):
306 create_reservation(
307 repo, run_id=f"ag-{i}", branch="b",
308 addresses=[f"x.py::fn{i}", "shared.py::util"],
309 ttl_seconds=3600,
310 )
311 data = _run_forecast_json(repo, capsys=capsys)
312 overlaps = [c for c in data["conflicts"] if c["conflict_type"] == "address_overlap"]
313 # shared.py::util is claimed by 3 agents → 1 overlap entry.
314 assert len(overlaps) >= 1
315
316
317 # ─────────────────────────────────────────────────────────────────────────────
318 # Unit tests — Pass 2: blast-radius (call graph)
319 # ─────────────────────────────────────────────────────────────────────────────
320
321
322 class TestPass2BlastRadius:
323 def test_call_graph_unavailable_oserror_warns(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
324 """OSError from build_reverse_graph → warning, not silent pass or crash."""
325 repo, *_ = repo_with_conflict
326 with (
327 patch(
328 "muse.cli.commands.forecast.resolve_commit_ref",
329 return_value=MagicMock(commit_id="abc123"),
330 ),
331 patch(
332 "muse.cli.commands.forecast.get_commit_snapshot_manifest",
333 return_value={"src/billing.py": "sha"},
334 ),
335 patch(
336 "muse.cli.commands.forecast.build_reverse_graph",
337 side_effect=OSError("index file missing"),
338 ),
339 ):
340 data = _run_forecast_json(repo, capsys=capsys)
341
342 assert data["call_graph_available"] is False
343 assert any("call graph unavailable" in w for w in data["warnings"])
344 # Direct overlap still detected (Pass 1 always runs).
345 assert any(c["conflict_type"] == "address_overlap" for c in data["conflicts"])
346
347 def test_call_graph_unavailable_keyerror_warns(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
348 repo, *_ = repo_with_conflict
349 with (
350 patch("muse.cli.commands.forecast.resolve_commit_ref",
351 return_value=MagicMock(commit_id="abc123")),
352 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
353 return_value={}),
354 patch("muse.cli.commands.forecast.build_reverse_graph",
355 side_effect=KeyError("missing key")),
356 ):
357 data = _run_forecast_json(repo, capsys=capsys)
358
359 assert data["call_graph_available"] is False
360 assert data["warnings"]
361
362 def test_call_graph_unavailable_valueerror_warns(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
363 repo, *_ = repo_with_conflict
364 with (
365 patch("muse.cli.commands.forecast.resolve_commit_ref",
366 return_value=MagicMock(commit_id="abc123")),
367 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
368 return_value={}),
369 patch("muse.cli.commands.forecast.build_reverse_graph",
370 side_effect=ValueError("bad data")),
371 ):
372 data = _run_forecast_json(repo, capsys=capsys)
373
374 assert data["call_graph_available"] is False
375 assert data["warnings"]
376
377 def test_call_graph_unavailable_attributeerror_warns(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
378 repo, *_ = repo_with_conflict
379 with (
380 patch("muse.cli.commands.forecast.resolve_commit_ref",
381 return_value=MagicMock(commit_id="abc123")),
382 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
383 return_value={}),
384 patch("muse.cli.commands.forecast.build_reverse_graph",
385 side_effect=AttributeError("NoneType has no attribute")),
386 ):
387 data = _run_forecast_json(repo, capsys=capsys)
388
389 assert data["call_graph_available"] is False
390 assert data["warnings"]
391
392 def test_unexpected_exception_propagates(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
393 """RuntimeError (unexpected) must NOT be silently swallowed."""
394 repo, *_ = repo_with_conflict
395 with (
396 patch("muse.cli.commands.forecast.resolve_commit_ref",
397 return_value=MagicMock(commit_id="abc123")),
398 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
399 return_value={}),
400 patch("muse.cli.commands.forecast.build_reverse_graph",
401 side_effect=RuntimeError("index corrupted")),
402 ):
403 with pytest.raises(RuntimeError, match="index corrupted"):
404 _run_forecast(repo, fmt="json")
405
406 def test_no_commits_warns(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
407 from muse.core.coordination import create_reservation
408 create_reservation(
409 repo, run_id="ag", branch="main",
410 addresses=["x.py::fn"], ttl_seconds=3600,
411 )
412 with patch(
413 "muse.cli.commands.forecast.resolve_commit_ref",
414 return_value=None,
415 ):
416 data = _run_forecast_json(repo, capsys=capsys)
417
418 assert data["call_graph_available"] is False
419 assert any("no commits" in w for w in data["warnings"])
420
421 def test_blast_radius_overlap_detected_with_mock_graph(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
422 """When call graph is available, blast_radius_overlap is detected."""
423 from muse.core.coordination import create_reservation
424 create_reservation(
425 repo, run_id="ag-1", branch="main",
426 addresses=["billing.py::compute_total"], ttl_seconds=3600,
427 )
428 create_reservation(
429 repo, run_id="ag-2", branch="feat",
430 addresses=["api.py::process_payment"], ttl_seconds=3600,
431 )
432 # Mock: each address sees the other in its transitive callers,
433 # so the test is correct regardless of dict iteration order.
434 _caller_map = {
435 "billing.py::compute_total": {1: ["api.py::process_payment"]},
436 "api.py::process_payment": {1: ["billing.py::compute_total"]},
437 }
438 with (
439 patch("muse.cli.commands.forecast.resolve_commit_ref",
440 return_value=MagicMock(commit_id="abc")),
441 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
442 return_value={"billing.py": "sha", "api.py": "sha2"}),
443 patch("muse.cli.commands.forecast.build_reverse_graph",
444 return_value={}),
445 patch("muse.cli.commands.forecast.transitive_callers",
446 side_effect=lambda name, rev, max_depth=0: _caller_map.get(name, {})),
447 ):
448 data = _run_forecast_json(repo, capsys=capsys)
449
450 assert data["call_graph_available"] is True
451 assert any(
452 c["conflict_type"] == "blast_radius_overlap" for c in data["conflicts"]
453 )
454 blast = next(
455 c for c in data["conflicts"] if c["conflict_type"] == "blast_radius_overlap"
456 )
457 assert blast["confidence"] == 0.75
458
459
460 # ─────────────────────────────────────────────────────────────────────────────
461 # Unit tests — Pass 3: operation conflicts
462 # ─────────────────────────────────────────────────────────────────────────────
463
464
465 class TestPass3OperationConflict:
466 def test_delete_vs_modify_conflict(self, repo_with_op_conflict: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
467 repo = repo_with_op_conflict
468 data = _run_forecast_json(repo, capsys=capsys)
469 op_conflicts = [
470 c for c in data["conflicts"] if c["conflict_type"] == "operation_conflict"
471 ]
472 assert len(op_conflicts) == 1
473 assert op_conflicts[0]["confidence"] == 0.9
474 assert "src/api.py::old_endpoint" in op_conflicts[0]["addresses"]
475
476 def test_delete_vs_rename_conflict(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
477 from muse.core.coordination import create_intent, create_reservation
478 res_a = create_reservation(
479 repo, run_id="ag-del", branch="b",
480 addresses=["x.py::fn"], ttl_seconds=3600,
481 )
482 res_b = create_reservation(
483 repo, run_id="ag-ren", branch="b2",
484 addresses=["x.py::fn"], ttl_seconds=3600,
485 )
486 create_intent(repo, res_a.reservation_id, "ag-del", "b",
487 ["x.py::fn"], "delete", "")
488 create_intent(repo, res_b.reservation_id, "ag-ren", "b2",
489 ["x.py::fn"], "rename", "")
490 data = _run_forecast_json(repo, capsys=capsys)
491 op_conflicts = [c for c in data["conflicts"] if c["conflict_type"] == "operation_conflict"]
492 assert op_conflicts
493
494 def test_delete_vs_extract_conflict(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
495 from muse.core.coordination import create_intent, create_reservation
496 res_a = create_reservation(repo, run_id="ag-del", branch="b",
497 addresses=["x.py::fn"], ttl_seconds=3600)
498 res_b = create_reservation(repo, run_id="ag-ext", branch="b2",
499 addresses=["x.py::fn"], ttl_seconds=3600)
500 create_intent(repo, res_a.reservation_id, "ag-del", "b",
501 ["x.py::fn"], "delete", "")
502 create_intent(repo, res_b.reservation_id, "ag-ext", "b2",
503 ["x.py::fn"], "extract", "")
504 data = _run_forecast_json(repo, capsys=capsys)
505 assert any(c["conflict_type"] == "operation_conflict" for c in data["conflicts"])
506
507 def test_no_conflict_same_op_same_agent(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
508 from muse.core.coordination import create_intent, create_reservation
509 res = create_reservation(repo, run_id="ag", branch="b",
510 addresses=["x.py::fn"], ttl_seconds=3600)
511 create_intent(repo, res.reservation_id, "ag", "b",
512 ["x.py::fn"], "modify", "")
513 create_intent(repo, res.reservation_id, "ag", "b",
514 ["x.py::fn"], "modify", "again")
515 data = _run_forecast_json(repo, capsys=capsys)
516 assert not any(c["conflict_type"] == "operation_conflict" for c in data["conflicts"])
517
518 def test_two_modifies_no_conflict(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
519 """Two agents both modifying → no operation_conflict (both allowed)."""
520 from muse.core.coordination import create_intent, create_reservation
521 res_a = create_reservation(repo, run_id="ag-1", branch="b1",
522 addresses=["x.py::fn"], ttl_seconds=3600)
523 res_b = create_reservation(repo, run_id="ag-2", branch="b2",
524 addresses=["x.py::fn"], ttl_seconds=3600)
525 create_intent(repo, res_a.reservation_id, "ag-1", "b1",
526 ["x.py::fn"], "modify", "")
527 create_intent(repo, res_b.reservation_id, "ag-2", "b2",
528 ["x.py::fn"], "modify", "")
529 data = _run_forecast_json(repo, capsys=capsys)
530 assert not any(c["conflict_type"] == "operation_conflict" for c in data["conflicts"])
531
532
533 # ─────────────────────────────────────────────────────────────────────────────
534 # Integration tests — full CLI
535 # ─────────────────────────────────────────────────────────────────────────────
536
537
538 class TestForecastIntegration:
539 def test_empty_swarm_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
540 data = _run_forecast_json(repo, capsys=capsys)
541 assert data["conflicts"] == []
542 assert data["active_reservations"] == 0
543 assert data["intents_count"] == 0
544
545 def test_json_schema_complete(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
546 data = _run_forecast_json(repo, capsys=capsys)
547 required_keys = {
548 "schema", "current_branch", "branch_filter",
549 "active_reservations", "intents_count", "call_graph_available",
550 "warnings", "conflicts", "high_risk", "medium_risk",
551 "low_risk", "duration_ms",
552 }
553 assert required_keys.issubset(data.keys())
554
555 def test_duration_ms_is_float(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
556 data = _run_forecast_json(repo, capsys=capsys)
557 assert isinstance(data["duration_ms"], float)
558 assert data["duration_ms"] >= 0
559
560 def test_current_branch_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
561 data = _run_forecast_json(repo, capsys=capsys)
562 assert isinstance(data["current_branch"], str)
563
564 def test_branch_filter_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
565 data = _run_forecast_json(repo, branch_filter="feat/x", capsys=capsys)
566 assert data["branch_filter"] == "feat/x"
567
568 def test_branch_filter_null_when_not_set(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
569 data = _run_forecast_json(repo, capsys=capsys)
570 assert data["branch_filter"] is None
571
572 def test_high_medium_low_risk_counts(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
573 repo, *_ = repo_with_conflict
574 data = _run_forecast_json(repo, capsys=capsys)
575 assert data["high_risk"] + data["medium_risk"] + data["low_risk"] == len(data["conflicts"])
576
577 def test_address_overlap_in_high_or_medium(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
578 repo, *_ = repo_with_conflict
579 data = _run_forecast_json(repo, capsys=capsys)
580 # address_overlap has confidence 1.0 → high risk
581 assert data["high_risk"] >= 1
582
583 def test_branch_filter_excludes_other_branches(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
584 from muse.core.coordination import create_reservation
585 create_reservation(repo, run_id="ag-1", branch="main",
586 addresses=["x.py::fn"], ttl_seconds=3600)
587 create_reservation(repo, run_id="ag-2", branch="feat",
588 addresses=["x.py::fn"], ttl_seconds=3600)
589 # Filter to "main" only — no overlap visible (only 1 agent on main).
590 data = _run_forecast_json(repo, branch_filter="main", capsys=capsys)
591 assert data["active_reservations"] == 1
592 assert data["conflicts"] == []
593
594 def test_format_json_flag(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
595 data = _run_forecast_json(repo, capsys=capsys)
596 assert "conflicts" in data
597
598 def test_text_format_default(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
599 text = _run_forecast_text(repo, capsys=capsys)
600 assert "Conflict forecast" in text
601
602 def test_text_shows_no_conflicts_message(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
603 text = _run_forecast_text(repo, capsys=capsys)
604 assert "No conflicts predicted" in text
605
606 def test_text_shows_conflict(self, repo_with_conflict: tuple[pathlib.Path, Reservation, Reservation], capsys: pytest.CaptureFixture[str]) -> None:
607 repo, *_ = repo_with_conflict
608 text = _run_forecast_text(repo, capsys=capsys)
609 assert "address_overlap" in text
610
611 def test_text_shows_elapsed(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
612 text = _run_forecast_text(repo, capsys=capsys)
613 assert "s)" in text
614
615 def test_text_shows_warning_when_call_graph_absent(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
616 from muse.core.coordination import create_reservation
617 create_reservation(repo, run_id="ag", branch="b",
618 addresses=["x.py::fn"], ttl_seconds=3600)
619 with patch("muse.cli.commands.forecast.resolve_commit_ref", return_value=None):
620 text = _run_forecast_text(repo, capsys=capsys)
621 assert "Note:" in text
622
623 def test_warnings_list_in_json_when_no_commits(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
624 with patch("muse.cli.commands.forecast.resolve_commit_ref", return_value=None):
625 data = _run_forecast_json(repo, capsys=capsys)
626 assert isinstance(data["warnings"], list)
627 assert data["warnings"]
628
629 def test_released_reservation_excluded_from_active(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
630 from muse.core.coordination import create_release, create_reservation
631 res = create_reservation(repo, run_id="ag", branch="b",
632 addresses=["x.py::fn"], ttl_seconds=3600)
633 create_release(repo, res.reservation_id, "ag", "completed")
634 data = _run_forecast_json(repo, capsys=capsys)
635 # active_reservations uses active_reservations() which excludes released.
636 assert data["active_reservations"] == 0
637
638 def test_expired_reservation_excluded(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
639 from muse.core.coordination import _reservations_dir, create_reservation
640 import json as _json
641 res = create_reservation(repo, run_id="ag", branch="b",
642 addresses=["x.py::fn"], ttl_seconds=1)
643 # Back-date.
644 path = _reservations_dir(repo) / f"{res.reservation_id}.json"
645 data = _json.loads(path.read_text())
646 data["expires_at"] = (
647 datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=10)
648 ).isoformat()
649 path.write_text(_json.dumps(data))
650 out = _run_forecast_json(repo, capsys=capsys)
651 assert out["active_reservations"] == 0
652
653
654 # ─────────────────────────────────────────────────────────────────────────────
655 # Security tests
656 # ─────────────────────────────────────────────────────────────────────────────
657
658
659 class TestForecastSecurity:
660 def test_ansi_in_run_id_stripped_from_text(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
661 from muse.core.coordination import create_reservation
662 ansi_run_id = "\x1b[31mmalicious\x1b[0m"
663 create_reservation(
664 repo, run_id=ansi_run_id, branch="b",
665 addresses=["x.py::fn"], ttl_seconds=3600,
666 )
667 create_reservation(
668 repo, run_id="normal-agent", branch="b2",
669 addresses=["x.py::fn"], ttl_seconds=3600,
670 )
671 text = _run_forecast_text(repo, capsys=capsys)
672 # ANSI escape bytes must not appear in text output.
673 assert "\x1b[" not in text
674 assert "malicious" in text # sanitized content still shown
675
676 def test_ansi_in_branch_stripped_from_text(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
677 from muse.core.coordination import create_reservation
678 malicious_branch = "\x1b[32mfeat/malicious\x1b[0m"
679 create_reservation(
680 repo, run_id="ag-1", branch=malicious_branch,
681 addresses=["x.py::fn"], ttl_seconds=3600,
682 )
683 create_reservation(
684 repo, run_id="ag-2", branch="b2",
685 addresses=["x.py::fn"], ttl_seconds=3600,
686 )
687 text = _run_forecast_text(repo, capsys=capsys)
688 assert "\x1b[" not in text
689
690 def test_control_chars_in_address_stripped_from_text(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
691 from muse.core.coordination import create_reservation
692 malicious_addr = "src/x.py::fn\x00\r\n"
693 create_reservation(
694 repo, run_id="ag-1", branch="b",
695 addresses=[malicious_addr], ttl_seconds=3600,
696 )
697 create_reservation(
698 repo, run_id="ag-2", branch="b2",
699 addresses=[malicious_addr], ttl_seconds=3600,
700 )
701 text = _run_forecast_text(repo, capsys=capsys)
702 assert "\x00" not in text
703 assert "\r" not in text
704
705 def test_address_filtering_uses_no_filesystem(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
706 """address_glob filtering in Pass 1/3 must not access the filesystem."""
707 from muse.core.coordination import create_reservation
708 # Address looks like a path traversal — must be treated as a pure string.
709 create_reservation(
710 repo, run_id="ag-1", branch="b",
711 addresses=["../../etc/passwd::root"], ttl_seconds=3600,
712 )
713 create_reservation(
714 repo, run_id="ag-2", branch="b2",
715 addresses=["../../etc/passwd::root"], ttl_seconds=3600,
716 )
717 # Should not raise FileNotFoundError or similar.
718 data = _run_forecast_json(repo, capsys=capsys)
719 assert any(c["conflict_type"] == "address_overlap" for c in data["conflicts"])
720
721
722 # ─────────────────────────────────────────────────────────────────────────────
723 # Stress tests
724 # ─────────────────────────────────────────────────────────────────────────────
725
726
727 class TestForecastStress:
728 def test_100_reservations_100_addresses_pass1_under_2s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
729 """Pass 1 with 100 reservations sharing many addresses completes in < 2 s."""
730 from muse.core.coordination import create_reservation
731
732 N = 100
733 shared_addresses = [f"src/mod{i % 10}.py::fn{i % 10}" for i in range(10)]
734 for i in range(N):
735 create_reservation(
736 repo,
737 run_id=f"ag-{i}",
738 branch=f"feat/branch-{i}",
739 addresses=[shared_addresses[i % 10], f"src/unique{i}.py::fn"],
740 ttl_seconds=3600,
741 )
742
743 t0 = time.monotonic()
744 data = _run_forecast_json(repo, capsys=capsys)
745 elapsed = time.monotonic() - t0
746
747 assert elapsed < 2.0, f"Pass 1 took {elapsed:.2f}s — too slow"
748 # Each shared address has many agents → many overlaps.
749 assert data["conflicts"]
750
751 def test_50_reservations_mock_callgraph_under_1s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
752 """Pass 2 with 50 reservations and a mock call graph completes in < 1 s."""
753 from muse.core.coordination import create_reservation
754
755 N = 50
756 for i in range(N):
757 create_reservation(
758 repo,
759 run_id=f"ag-{i}",
760 branch=f"b{i}",
761 addresses=[f"src/f{i}.py::fn{i}"],
762 ttl_seconds=3600,
763 )
764
765 # Mock call graph: no transitive relationships → no blast-radius conflicts.
766 with (
767 patch("muse.cli.commands.forecast.resolve_commit_ref",
768 return_value=MagicMock(commit_id="abc")),
769 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
770 return_value={f"src/f{i}.py": f"sha{i}" for i in range(N)}),
771 patch("muse.cli.commands.forecast.build_reverse_graph",
772 return_value={}),
773 ):
774 t0 = time.monotonic()
775 data = _run_forecast_json(repo, capsys=capsys)
776 elapsed = time.monotonic() - t0
777
778 assert elapsed < 1.0, f"Pass 2 took {elapsed:.2f}s — too slow"
779 assert data["call_graph_available"] is True
780
781 def test_200_intents_pass3_under_1s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
782 """Pass 3 with 200 intents completes in < 1 s."""
783 from muse.core.coordination import create_intent, create_reservation
784
785 N = 200
786 for i in range(N):
787 res = create_reservation(
788 repo, run_id=f"ag-{i}", branch=f"b{i}",
789 addresses=[f"f{i}.py::fn"], ttl_seconds=3600,
790 )
791 op = "delete" if i % 2 == 0 else "modify"
792 create_intent(
793 repo, res.reservation_id, f"ag-{i}", f"b{i}",
794 [f"f{i}.py::fn"], op, "",
795 )
796
797 t0 = time.monotonic()
798 data = _run_forecast_json(repo, capsys=capsys)
799 elapsed = time.monotonic() - t0
800
801 assert elapsed < 1.0, f"Pass 3 took {elapsed:.2f}s — too slow"
802 # Pairs of delete+modify on same address → operation_conflict entries.
803 # Only agents sharing the same address create a conflict, so here
804 # each fn is unique per agent → no operation conflicts expected.
805 assert isinstance(data["conflicts"], list)
806
807 def test_all_three_passes_combined_under_3s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
808 """Combined forecast across all three passes with realistic workload."""
809 from muse.core.coordination import create_intent, create_reservation
810
811 # 20 agents, 5 shared addresses (overlap), 5 unique.
812 shared = [f"shared.py::fn{i}" for i in range(5)]
813 for i in range(20):
814 res = create_reservation(
815 repo, run_id=f"ag-{i}", branch=f"b{i % 4}",
816 addresses=[shared[i % 5], f"unique{i}.py::fn"],
817 ttl_seconds=3600,
818 )
819 if i % 3 == 0:
820 create_intent(
821 repo, res.reservation_id, f"ag-{i}", f"b{i % 4}",
822 [shared[i % 5]], "delete" if i % 6 == 0 else "modify", "",
823 )
824
825 with (
826 patch("muse.cli.commands.forecast.resolve_commit_ref", return_value=None),
827 ):
828 t0 = time.monotonic()
829 data = _run_forecast_json(repo, capsys=capsys)
830 elapsed = time.monotonic() - t0
831
832 assert elapsed < 3.0, f"Combined forecast took {elapsed:.2f}s — too slow"
833 assert data["active_reservations"] == 20
834
835 def test_500_reservations_500_intents_under_5s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
836 """Large swarm: 500 reservations + 500 intents across all three passes."""
837 from muse.core.coordination import create_intent, create_reservation
838
839 N = 500
840 shared_addresses = [f"src/hot{i % 20}.py::fn" for i in range(20)]
841 for i in range(N):
842 res = create_reservation(
843 repo,
844 run_id=f"ag-{i}",
845 branch=f"b{i % 10}",
846 addresses=[shared_addresses[i % 20], f"unique{i}.py::fn"],
847 ttl_seconds=3600,
848 )
849 op = "delete" if i % 50 == 0 else "modify"
850 create_intent(
851 repo, res.reservation_id, f"ag-{i}", f"b{i % 10}",
852 [shared_addresses[i % 20]], op, "",
853 )
854
855 with patch("muse.cli.commands.forecast.resolve_commit_ref", return_value=None):
856 t0 = time.monotonic()
857 data = _run_forecast_json(repo, capsys=capsys)
858 elapsed = time.monotonic() - t0
859
860 assert elapsed < 5.0, f"500-agent forecast took {elapsed:.2f}s — too slow"
861 assert data["active_reservations"] == N
862 assert data["partial_forecast"] is True
863
864
865 # ─────────────────────────────────────────────────────────────────────────────
866 # Input validation tests
867 # ─────────────────────────────────────────────────────────────────────────────
868
869
870 class TestForecastInputValidation:
871 def test_run_id_at_max_length_accepted(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
872 run_id = "a" * _MAX_RUN_ID_LEN
873 data = _run_forecast_json(repo, run_id_filter=run_id, capsys=capsys)
874 assert data["run_id_filter"] == run_id
875
876 def test_run_id_over_max_length_exits_user_error(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
877 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
878 code, _ = _run_forecast(repo, run_id_filter=run_id, fmt="text")
879 assert code == ExitCode.USER_ERROR
880
881 def test_run_id_over_max_json_returns_error_field(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
882 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
883 _run_forecast(repo, run_id_filter=run_id, fmt="json")
884 out = capsys.readouterr().out.strip()
885 data = json.loads(out)
886 assert "error" in data
887 assert data.get("status") == "bad_args"
888
889 def test_run_id_error_output_is_compact(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
890 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
891 _run_forecast(repo, run_id_filter=run_id, fmt="json")
892 out = capsys.readouterr().out.strip()
893 assert "\n" not in out
894
895 def test_min_confidence_zero_accepted(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
896 data = _run_forecast_json(repo, min_confidence=0.0, capsys=capsys)
897 assert data["min_confidence"] == 0.0
898
899 def test_min_confidence_one_accepted(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
900 data = _run_forecast_json(repo, min_confidence=1.0, capsys=capsys)
901 assert data["min_confidence"] == 1.0
902
903 def test_min_confidence_above_1_exits_user_error(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
904 code, _ = _run_forecast(repo, min_confidence=1.5, fmt="text")
905 assert code == ExitCode.USER_ERROR
906
907 def test_min_confidence_negative_exits_user_error(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
908 code, _ = _run_forecast(repo, min_confidence=-0.1, fmt="text")
909 assert code == ExitCode.USER_ERROR
910
911 def test_min_confidence_above_1_json_returns_error_field(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
912 _run_forecast(repo, min_confidence=1.5, fmt="json")
913 out = capsys.readouterr().out.strip()
914 data = json.loads(out)
915 assert "error" in data
916 assert data.get("status") == "bad_args"
917
918 def test_min_confidence_error_output_is_compact(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
919 _run_forecast(repo, min_confidence=99.0, fmt="json")
920 out = capsys.readouterr().out.strip()
921 assert "\n" not in out
922
923
924 # ─────────────────────────────────────────────────────────────────────────────
925 # --run-id filter tests
926 # ─────────────────────────────────────────────────────────────────────────────
927
928
929 class TestForecastRunIdFilter:
930 def test_run_id_filter_keeps_only_matching_conflicts(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
931 from muse.core.coordination import create_reservation
932 create_reservation(repo, run_id="ag-target", branch="b1",
933 addresses=["x.py::fn"], ttl_seconds=3600)
934 create_reservation(repo, run_id="ag-other", branch="b2",
935 addresses=["x.py::fn"], ttl_seconds=3600)
936 create_reservation(repo, run_id="ag-third", branch="b3",
937 addresses=["y.py::fn"], ttl_seconds=3600)
938 create_reservation(repo, run_id="ag-fourth", branch="b4",
939 addresses=["y.py::fn"], ttl_seconds=3600)
940
941 # Without filter: 2 conflicts (x.py and y.py).
942 all_data = _run_forecast_json(repo, capsys=capsys)
943 assert len(all_data["conflicts"]) == 2
944
945 # Filter to ag-target: only x.py conflict (ag-target is not on y.py).
946 filtered = _run_forecast_json(repo, run_id_filter="ag-target", capsys=capsys)
947 assert all(
948 any(a.startswith("ag-target@") for a in c["agents"])
949 for c in filtered["conflicts"]
950 )
951
952 def test_run_id_filter_unknown_agent_returns_empty(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
953 from muse.core.coordination import create_reservation
954 create_reservation(repo, run_id="ag-1", branch="b1",
955 addresses=["x.py::fn"], ttl_seconds=3600)
956 create_reservation(repo, run_id="ag-2", branch="b2",
957 addresses=["x.py::fn"], ttl_seconds=3600)
958 data = _run_forecast_json(repo, run_id_filter="no-such-agent", capsys=capsys)
959 assert data["conflicts"] == []
960
961 def test_run_id_filter_does_not_change_reservation_count(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
962 from muse.core.coordination import create_reservation
963 create_reservation(repo, run_id="ag-1", branch="b1",
964 addresses=["x.py::fn"], ttl_seconds=3600)
965 create_reservation(repo, run_id="ag-2", branch="b2",
966 addresses=["x.py::fn"], ttl_seconds=3600)
967 data = _run_forecast_json(repo, run_id_filter="ag-1", capsys=capsys)
968 # Reservation count reflects the full filtered-by-branch set, not by run_id.
969 assert data["active_reservations"] == 2
970
971 def test_run_id_filter_in_json_field(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
972 data = _run_forecast_json(repo, run_id_filter="my-agent", capsys=capsys)
973 assert data["run_id_filter"] == "my-agent"
974
975 def test_run_id_filter_null_when_not_set(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
976 data = _run_forecast_json(repo, capsys=capsys)
977 assert data["run_id_filter"] is None
978
979 def test_run_id_filter_operation_conflict(self, repo_with_op_conflict: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
980 repo = repo_with_op_conflict
981 # agent-del is involved in the op conflict; filter to it.
982 data = _run_forecast_json(repo, run_id_filter="agent-del", capsys=capsys)
983 assert any(c["conflict_type"] == "operation_conflict" for c in data["conflicts"])
984
985 def test_run_id_filter_excludes_operation_conflict_for_uninvolved_agent(
986 self, repo_with_op_conflict: pathlib.Path, capsys: pytest.CaptureFixture[str]
987 ) -> None:
988 repo = repo_with_op_conflict
989 data = _run_forecast_json(repo, run_id_filter="uninvolved-bot", capsys=capsys)
990 assert data["conflicts"] == []
991
992
993 # ─────────────────────────────────────────────────────────────────────────────
994 # --min-confidence filter tests
995 # ─────────────────────────────────────────────────────────────────────────────
996
997
998 class TestForecastMinConfidence:
999 def _setup_all_conflict_types(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1000 """Return (address_overlap, operation_conflict) — both detected."""
1001 from muse.core.coordination import create_intent, create_reservation
1002 res_a = create_reservation(repo, run_id="ag-1", branch="b1",
1003 addresses=["x.py::fn"], ttl_seconds=3600)
1004 res_b = create_reservation(repo, run_id="ag-2", branch="b2",
1005 addresses=["x.py::fn"], ttl_seconds=3600)
1006 create_intent(repo, res_a.reservation_id, "ag-1", "b1",
1007 ["x.py::fn"], "delete", "")
1008 create_intent(repo, res_b.reservation_id, "ag-2", "b2",
1009 ["x.py::fn"], "modify", "")
1010
1011 def test_min_confidence_0_shows_all(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1012 self._setup_all_conflict_types(repo, capsys)
1013 data = _run_forecast_json(repo, min_confidence=0.0, capsys=capsys)
1014 types = {c["conflict_type"] for c in data["conflicts"]}
1015 assert "address_overlap" in types
1016 assert "operation_conflict" in types
1017
1018 def test_min_confidence_0_9_hides_medium(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1019 self._setup_all_conflict_types(repo, capsys)
1020 # Inject a mock blast_radius_overlap at 0.75.
1021 with (
1022 patch("muse.cli.commands.forecast.resolve_commit_ref",
1023 return_value=MagicMock(commit_id="abc")),
1024 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
1025 return_value={}),
1026 patch("muse.cli.commands.forecast.build_reverse_graph",
1027 return_value={}),
1028 patch("muse.cli.commands.forecast.transitive_callers",
1029 return_value={}),
1030 ):
1031 data = _run_forecast_json(repo, min_confidence=0.9, capsys=capsys)
1032 for c in data["conflicts"]:
1033 assert c["confidence"] >= 0.9
1034
1035 def test_min_confidence_at_exact_threshold_included(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1036 self._setup_all_conflict_types(repo, capsys)
1037 # address_overlap has confidence 1.0; 1.0 >= 1.0 → included.
1038 data = _run_forecast_json(repo, min_confidence=1.0, capsys=capsys)
1039 types = {c["conflict_type"] for c in data["conflicts"]}
1040 assert "address_overlap" in types
1041
1042 def test_min_confidence_1_hides_operation_conflict(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1043 self._setup_all_conflict_types(repo, capsys)
1044 # operation_conflict has confidence 0.9; 0.9 < 1.0 → hidden.
1045 data = _run_forecast_json(repo, min_confidence=1.0, capsys=capsys)
1046 assert not any(c["conflict_type"] == "operation_conflict" for c in data["conflicts"])
1047
1048 def test_risk_counts_reflect_filtered_list(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1049 self._setup_all_conflict_types(repo, capsys)
1050 data = _run_forecast_json(repo, min_confidence=0.95, capsys=capsys)
1051 # After filtering to >= 0.95, only address_overlap (1.0) remains.
1052 assert data["high_risk"] + data["medium_risk"] + data["low_risk"] == len(data["conflicts"])
1053
1054 def test_min_confidence_in_json_field(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1055 data = _run_forecast_json(repo, min_confidence=0.75, capsys=capsys)
1056 assert data["min_confidence"] == 0.75
1057
1058
1059 # ─────────────────────────────────────────────────────────────────────────────
1060 # JSON format — compact output + new schema fields
1061 # ─────────────────────────────────────────────────────────────────────────────
1062
1063
1064 class TestForecastJsonFormat:
1065 def test_json_output_is_compact(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1066 """No indent=2 — output must be a single line."""
1067 _run_forecast(repo, fmt="json")
1068 out = capsys.readouterr().out.strip()
1069 assert "\n" not in out, "JSON output must be compact (no newlines)"
1070
1071 def test_partial_forecast_true_when_no_commits(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1072 with patch("muse.cli.commands.forecast.resolve_commit_ref", return_value=None):
1073 data = _run_forecast_json(repo, capsys=capsys)
1074 assert data["partial_forecast"] is True
1075
1076 def test_partial_forecast_false_when_call_graph_available(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1077 from muse.core.coordination import create_reservation
1078 create_reservation(repo, run_id="ag", branch="main",
1079 addresses=["x.py::fn"], ttl_seconds=3600)
1080 with (
1081 patch("muse.cli.commands.forecast.resolve_commit_ref",
1082 return_value=MagicMock(commit_id="abc")),
1083 patch("muse.cli.commands.forecast.get_commit_snapshot_manifest",
1084 return_value={}),
1085 patch("muse.cli.commands.forecast.build_reverse_graph", return_value={}),
1086 ):
1087 data = _run_forecast_json(repo, capsys=capsys)
1088 assert data["partial_forecast"] is False
1089
1090 def test_run_id_filter_field_in_schema(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1091 data = _run_forecast_json(repo, capsys=capsys)
1092 assert "run_id_filter" in data
1093
1094 def test_min_confidence_field_in_schema(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1095 data = _run_forecast_json(repo, capsys=capsys)
1096 assert "min_confidence" in data
1097
1098 def test_partial_forecast_field_in_schema(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1099 data = _run_forecast_json(repo, capsys=capsys)
1100 assert "partial_forecast" in data
1101
1102 def test_all_required_keys_present(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1103 data = _run_forecast_json(repo, capsys=capsys)
1104 required = {
1105 "schema", "current_branch", "branch_filter",
1106 "run_id_filter", "min_confidence",
1107 "active_reservations", "intents_count",
1108 "call_graph_available", "partial_forecast",
1109 "warnings", "conflicts", "high_risk", "medium_risk",
1110 "low_risk", "duration_ms",
1111 }
1112 assert required.issubset(data.keys())
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago