gabriel / muse public
test_cmd_dag.py python
1,290 lines 51.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Tests for muse coord dag — dependency DAG on reservations.
2
3 Coverage goals
4 --------------
5 * Unit — every public function in ``muse.core.dag``
6 * Integration — add_dependencies → load → build_graph → detect/sort round-trip
7 * CLI — ``muse coord dag`` and ``muse coord reserve --depends-on`` via direct
8 function dispatch and stdout capture
9 * Security — UUID validation, path traversal, self-dependency rejection
10 * Stress — large linear chains, wide diamonds, many-node graphs
11
12 Test conventions
13 ----------------
14 * Every test receives a fresh repo fixture via ``_make_repo(tmp_path)``.
15 * Active-reservation state is injected by creating real Reservation files via
16 :func:`muse.core.coordination.create_reservation` + the repo fixture.
17 * Time is frozen where predictable timestamps matter.
18 * CLI dispatch calls ``run`` directly with a hand-assembled ``argparse.Namespace``.
19 """
20
21 from __future__ import annotations
22
23 import argparse
24 import datetime
25 import json
26 import pathlib
27 import time
28 import uuid
29 from collections.abc import Generator
30 from contextlib import AbstractContextManager
31 from unittest.mock import MagicMock, patch
32
33 import pytest
34
35 from muse.core.coordination import Reservation
36
37 from muse.core._types import MsgpackValue
38 from muse.core.dag import (
39 AdjacencyMap,
40 DependencyRecord,
41 _MAX_DEPS,
42 _dependencies_dir,
43 _validate_id,
44 add_dependencies,
45 build_graph,
46 detect_cycle,
47 ensure_dag_dirs,
48 get_blocking,
49 is_blocked,
50 load_all_dependencies,
51 load_dag,
52 load_dependencies,
53 topological_sort,
54 )
55 from muse.cli.commands.dag import run as dag_run
56 from muse.cli.commands.reserve import run as reserve_run
57
58 UTC = datetime.timezone.utc
59 _EPOCH = datetime.datetime(2026, 3, 30, 12, 0, 0, tzinfo=UTC)
60
61 # Helpers for generating valid UUID4 strings.
62 _A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
63 _B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"
64 _C = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"
65 _D = "dddddddd-dddd-4ddd-8ddd-dddddddddddd"
66 _E = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
67
68
69 def _uuid() -> str:
70 return str(uuid.uuid4())
71
72
73 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
74 """Return a minimal muse repo root."""
75 (tmp_path / ".muse").mkdir(parents=True)
76 return tmp_path
77
78
79 def _freeze(ts: datetime.datetime) -> AbstractContextManager[MagicMock]:
80 return patch("muse.core.dag._now_utc", return_value=ts)
81
82
83 def _dag_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
84 defaults = {
85 "reservation_id": None,
86 "topo": False,
87 "active_only": False,
88 "fmt": "json",
89 }
90 defaults.update(kwargs)
91 return argparse.Namespace(**defaults)
92
93
94 def _reserve_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
95 defaults = {
96 "addresses": ["billing.py::compute"],
97 "run_id": "agent-1",
98 "ttl": 3600,
99 "operation": None,
100 "fmt": "json",
101 "depends_on": [],
102 }
103 defaults.update(kwargs)
104 return argparse.Namespace(**defaults)
105
106
107 def _patch_repo(repo: pathlib.Path) -> AbstractContextManager[None]:
108 from contextlib import ExitStack, contextmanager
109
110 @contextmanager
111 def _ctx() -> Generator[None, None, None]:
112 with ExitStack() as stack:
113 stack.enter_context(patch("muse.cli.commands.reserve.require_repo", return_value=repo))
114 stack.enter_context(patch("muse.cli.commands.reserve.read_current_branch", return_value="dev"))
115 yield
116
117 return _ctx()
118
119
120 def _patch_dag_repo(repo: pathlib.Path) -> AbstractContextManager[MagicMock]:
121 return patch("muse.cli.commands.dag.require_repo", return_value=repo)
122
123
124 # ── _validate_id ─────────────────────────────────────────────────────────────
125
126
127 class TestValidateId:
128 """UUID4 validation rejects any non-UUID or dangerous string."""
129
130 def test_accepts_valid_uuid4(self) -> None:
131 _validate_id(_A) # no exception
132
133 def test_accepts_uppercase(self) -> None:
134 _validate_id(_A.upper())
135
136 def test_rejects_empty(self) -> None:
137 with pytest.raises(ValueError):
138 _validate_id("")
139
140 def test_rejects_non_uuid(self) -> None:
141 with pytest.raises(ValueError):
142 _validate_id("not-a-uuid")
143
144 def test_rejects_path_traversal(self) -> None:
145 with pytest.raises(ValueError):
146 _validate_id("../../etc/passwd")
147
148 def test_rejects_null_byte(self) -> None:
149 with pytest.raises(ValueError):
150 _validate_id("\x00" * 36)
151
152 def test_rejects_uuid_v3(self) -> None:
153 with pytest.raises(ValueError):
154 _validate_id("aaaaaaaa-aaaa-3aaa-8aaa-aaaaaaaaaaaa")
155
156 def test_rejects_slash(self) -> None:
157 with pytest.raises(ValueError):
158 _validate_id("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaa/aa")
159
160
161 # ── DependencyRecord ───────────────────────────────────────────────────────────
162
163
164 class TestDependencyRecord:
165 """DependencyRecord serialisation round-trip."""
166
167 def _make(self, **kw: MsgpackValue) -> DependencyRecord:
168 defaults = dict(
169 reservation_id=_A,
170 depends_on=[_B, _C],
171 created_at=_EPOCH,
172 )
173 defaults.update(kw)
174 return DependencyRecord(**defaults)
175
176 def test_to_dict_round_trip(self) -> None:
177 rec = self._make()
178 d = rec.to_dict()
179 rec2 = DependencyRecord.from_dict(d)
180 assert rec2.reservation_id == _A
181 assert rec2.depends_on == [_B, _C]
182
183 def test_from_dict_missing_created_at_defaults_to_now(self) -> None:
184 d = {"reservation_id": _A, "depends_on": [_B]}
185 rec = DependencyRecord.from_dict(d)
186 assert isinstance(rec.created_at, datetime.datetime)
187
188 def test_from_dict_empty_depends_on(self) -> None:
189 d = {"reservation_id": _A, "depends_on": [], "created_at": _EPOCH.isoformat()}
190 rec = DependencyRecord.from_dict(d)
191 assert rec.depends_on == []
192
193 def test_from_dict_non_list_depends_on_defaults_empty(self) -> None:
194 d = {"reservation_id": _A, "depends_on": None, "created_at": _EPOCH.isoformat()}
195 rec = DependencyRecord.from_dict(d)
196 assert rec.depends_on == []
197
198 def test_schema_version_in_dict(self) -> None:
199 rec = self._make()
200 d = rec.to_dict()
201 assert "schema_version" in d
202
203
204 # ── ensure_dag_dirs / load_all_dependencies ───────────────────────────────────
205
206
207 class TestEnsureDagDirs:
208 def test_creates_directory(self, tmp_path: pathlib.Path) -> None:
209 repo = _make_repo(tmp_path)
210 ensure_dag_dirs(repo)
211 assert _dependencies_dir(repo).is_dir()
212
213 def test_idempotent(self, tmp_path: pathlib.Path) -> None:
214 repo = _make_repo(tmp_path)
215 ensure_dag_dirs(repo)
216 ensure_dag_dirs(repo)
217
218 def test_empty_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
219 repo = _make_repo(tmp_path)
220 assert load_all_dependencies(repo) == []
221
222 def test_non_existent_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
223 repo = _make_repo(tmp_path)
224 assert load_all_dependencies(repo) == []
225
226 def test_skips_corrupt_file(self, tmp_path: pathlib.Path) -> None:
227 repo = _make_repo(tmp_path)
228 ensure_dag_dirs(repo)
229 (_dependencies_dir(repo) / f"{_A}.json").write_text("NOT JSON")
230 assert load_all_dependencies(repo) == []
231
232
233 # ── load_dependencies ─────────────────────────────────────────────────────────
234
235
236 class TestLoadDependencies:
237 def test_returns_none_for_missing(self, tmp_path: pathlib.Path) -> None:
238 repo = _make_repo(tmp_path)
239 ensure_dag_dirs(repo)
240 assert load_dependencies(repo, _A) is None
241
242 def test_invalid_id_raises(self, tmp_path: pathlib.Path) -> None:
243 repo = _make_repo(tmp_path)
244 with pytest.raises(ValueError):
245 load_dependencies(repo, "not-a-uuid")
246
247 def test_returns_record_after_add(self, tmp_path: pathlib.Path) -> None:
248 repo = _make_repo(tmp_path)
249 with _freeze(_EPOCH):
250 add_dependencies(repo, _A, [_B])
251 rec = load_dependencies(repo, _A)
252 assert rec is not None
253 assert rec.depends_on == [_B]
254
255
256 # ── build_graph ────────────────────────────────────────────────────────────────
257
258
259 class TestBuildGraph:
260 def _rec(self, rid: str, deps: list[str]) -> DependencyRecord:
261 return DependencyRecord(rid, deps, _EPOCH)
262
263 def test_empty_input(self) -> None:
264 g = build_graph([])
265 assert g == {}
266
267 def test_single_node_no_deps(self) -> None:
268 g = build_graph([self._rec(_A, [])])
269 assert g[_A] == set()
270
271 def test_adds_dependency_nodes(self) -> None:
272 g = build_graph([self._rec(_A, [_B])])
273 assert _A in g
274 assert _B in g
275 assert _B in g[_A]
276
277 def test_diamond_shape(self) -> None:
278 # A → B, A → C, B → D, C → D
279 g = build_graph([
280 self._rec(_A, [_B, _C]),
281 self._rec(_B, [_D]),
282 self._rec(_C, [_D]),
283 ])
284 assert g[_A] == {_B, _C}
285 assert g[_B] == {_D}
286 assert g[_C] == {_D}
287 assert g[_D] == set()
288
289 def test_leaf_nodes_included(self) -> None:
290 g = build_graph([self._rec(_A, [_B, _C])])
291 assert _B in g and g[_B] == set()
292 assert _C in g and g[_C] == set()
293
294
295 # ── detect_cycle ───────────────────────────────────────────────────────────────
296
297
298 class TestDetectCycle:
299 def test_empty_graph_no_cycle(self) -> None:
300 assert detect_cycle({}) is None
301
302 def test_single_node_no_cycle(self) -> None:
303 assert detect_cycle({_A: set()}) is None
304
305 def test_linear_chain_no_cycle(self) -> None:
306 g = {_A: {_B}, _B: {_C}, _C: set()}
307 assert detect_cycle(g) is None
308
309 def test_direct_self_loop(self) -> None:
310 g = {_A: {_A}}
311 cycle = detect_cycle(g)
312 assert cycle is not None
313 assert _A in cycle
314
315 def test_two_node_cycle(self) -> None:
316 g = {_A: {_B}, _B: {_A}}
317 cycle = detect_cycle(g)
318 assert cycle is not None
319 assert len(cycle) >= 2
320
321 def test_three_node_cycle(self) -> None:
322 g = {_A: {_B}, _B: {_C}, _C: {_A}}
323 cycle = detect_cycle(g)
324 assert cycle is not None
325 assert len(cycle) >= 3
326
327 def test_diamond_no_cycle(self) -> None:
328 # A → B, A → C, B → D, C → D (fan-in is fine)
329 g = {_A: {_B, _C}, _B: {_D}, _C: {_D}, _D: set()}
330 assert detect_cycle(g) is None
331
332 def test_cycle_path_starts_and_ends_same(self) -> None:
333 g = {_A: {_B}, _B: {_C}, _C: {_A}}
334 cycle = detect_cycle(g)
335 assert cycle is not None
336 assert cycle[0] == cycle[-1]
337
338 def test_unrelated_subgraph_no_cycle(self) -> None:
339 # A → B (no cycle), C → D (no cycle)
340 g = {_A: {_B}, _B: set(), _C: {_D}, _D: set()}
341 assert detect_cycle(g) is None
342
343
344 # ── topological_sort ───────────────────────────────────────────────────────────
345
346
347 class TestTopologicalSort:
348 def test_empty_graph(self) -> None:
349 result = topological_sort({})
350 assert result == []
351
352 def test_single_node(self) -> None:
353 result = topological_sort({_A: set()})
354 assert result == [_A]
355
356 def test_linear_chain_order(self) -> None:
357 # A → B → C: C should come first (fewest deps), then B, then A
358 g = {_A: {_B}, _B: {_C}, _C: set()}
359 result = topological_sort(g)
360 assert result.index(_C) < result.index(_B) < result.index(_A)
361
362 def test_diamond_order(self) -> None:
363 # D (leaf) → B, C (intermediate) → A (top)
364 g = {_A: {_B, _C}, _B: {_D}, _C: {_D}, _D: set()}
365 result = topological_sort(g)
366 assert result.index(_D) < result.index(_B)
367 assert result.index(_D) < result.index(_C)
368 assert result.index(_B) < result.index(_A)
369 assert result.index(_C) < result.index(_A)
370
371 def test_all_nodes_present(self) -> None:
372 g = {_A: {_B}, _B: {_C}, _C: set()}
373 result = topological_sort(g)
374 assert set(result) == {_A, _B, _C}
375
376 def test_subset_nodes(self) -> None:
377 g = {_A: {_B}, _B: {_C}, _C: set(), _D: set()}
378 # Only ask for A's subgraph
379 result = topological_sort(g, nodes=[_A])
380 assert _A in result and _B in result and _C in result
381 # D is unrelated — may or may not be included; just check order
382 assert result.index(_C) < result.index(_B) < result.index(_A)
383
384 def test_cycle_raises(self) -> None:
385 g = {_A: {_B}, _B: {_A}}
386 with pytest.raises(ValueError, match="cycle"):
387 topological_sort(g)
388
389 def test_independent_nodes_all_returned(self) -> None:
390 g = {_A: set(), _B: set(), _C: set()}
391 result = topological_sort(g)
392 assert set(result) == {_A, _B, _C}
393
394
395 # ── is_blocked / get_blocking ──────────────────────────────────────────────────
396
397
398 class TestIsBlockedAndGetBlocking:
399 def test_no_deps_not_blocked(self) -> None:
400 g = {_A: set()}
401 assert is_blocked(_A, g, frozenset({_B})) is False
402
403 def test_dep_inactive_not_blocked(self) -> None:
404 g = {_A: {_B}}
405 # B is not in active_ids → unblocked
406 assert is_blocked(_A, g, frozenset()) is False
407
408 def test_dep_active_is_blocked(self) -> None:
409 g = {_A: {_B}}
410 assert is_blocked(_A, g, frozenset({_B})) is True
411
412 def test_multiple_deps_one_active(self) -> None:
413 g = {_A: {_B, _C}}
414 assert is_blocked(_A, g, frozenset({_B})) is True
415
416 def test_unknown_node_not_blocked(self) -> None:
417 g = {}
418 assert is_blocked(_A, g, frozenset({_B})) is False
419
420 def test_get_blocking_empty(self) -> None:
421 g = {_A: {_B}}
422 assert get_blocking(_A, g, frozenset()) == []
423
424 def test_get_blocking_returns_active_deps(self) -> None:
425 g = {_A: {_B, _C}}
426 blocking = get_blocking(_A, g, frozenset({_B}))
427 assert blocking == [_B]
428
429 def test_get_blocking_sorted(self) -> None:
430 g = {_A: {_B, _C, _D}}
431 blocking = get_blocking(_A, g, frozenset({_C, _B}))
432 assert blocking == sorted([_B, _C])
433
434
435 # ── add_dependencies ───────────────────────────────────────────────────────────
436
437
438 class TestAddDependencies:
439 def test_creates_file_on_disk(self, tmp_path: pathlib.Path) -> None:
440 repo = _make_repo(tmp_path)
441 add_dependencies(repo, _A, [_B])
442 dep_file = _dependencies_dir(repo) / f"{_A}.json"
443 assert dep_file.is_file()
444
445 def test_file_is_valid_json(self, tmp_path: pathlib.Path) -> None:
446 repo = _make_repo(tmp_path)
447 add_dependencies(repo, _A, [_B])
448 raw = json.loads((_dependencies_dir(repo) / f"{_A}.json").read_text())
449 assert raw["reservation_id"] == _A
450 assert _B in raw["depends_on"]
451
452 def test_returns_correct_record(self, tmp_path: pathlib.Path) -> None:
453 repo = _make_repo(tmp_path)
454 rec = add_dependencies(repo, _A, [_B, _C])
455 assert rec.reservation_id == _A
456 assert set(rec.depends_on) == {_B, _C}
457
458 def test_deduplicates_deps(self, tmp_path: pathlib.Path) -> None:
459 repo = _make_repo(tmp_path)
460 rec = add_dependencies(repo, _A, [_B, _B, _B])
461 assert rec.depends_on == [_B]
462
463 def test_empty_deps_accepted(self, tmp_path: pathlib.Path) -> None:
464 repo = _make_repo(tmp_path)
465 rec = add_dependencies(repo, _A, [])
466 assert rec.depends_on == []
467
468 def test_write_once_raises_file_exists(self, tmp_path: pathlib.Path) -> None:
469 repo = _make_repo(tmp_path)
470 add_dependencies(repo, _A, [_B])
471 with pytest.raises(FileExistsError):
472 add_dependencies(repo, _A, [_C])
473
474 def test_self_dependency_raises(self, tmp_path: pathlib.Path) -> None:
475 repo = _make_repo(tmp_path)
476 with pytest.raises(ValueError, match="itself"):
477 add_dependencies(repo, _A, [_A])
478
479 def test_invalid_reservation_id_raises(self, tmp_path: pathlib.Path) -> None:
480 repo = _make_repo(tmp_path)
481 with pytest.raises(ValueError):
482 add_dependencies(repo, "not-a-uuid", [_B])
483
484 def test_invalid_dep_id_raises(self, tmp_path: pathlib.Path) -> None:
485 repo = _make_repo(tmp_path)
486 with pytest.raises(ValueError):
487 add_dependencies(repo, _A, ["../../etc/passwd"])
488
489 def test_too_many_deps_raises(self, tmp_path: pathlib.Path) -> None:
490 repo = _make_repo(tmp_path)
491 # Generate _MAX_DEPS + 1 unique valid UUIDs
492 many = [str(uuid.uuid4()) for _ in range(_MAX_DEPS + 1)]
493 with pytest.raises(ValueError, match="too many"):
494 add_dependencies(repo, _A, many)
495
496 def test_max_deps_accepted(self, tmp_path: pathlib.Path) -> None:
497 repo = _make_repo(tmp_path)
498 many = [str(uuid.uuid4()) for _ in range(_MAX_DEPS)]
499 rec = add_dependencies(repo, _A, many)
500 assert len(rec.depends_on) == _MAX_DEPS
501
502 def test_rejects_cycle_two_node(self, tmp_path: pathlib.Path) -> None:
503 """A → B then B → A must be rejected."""
504 repo = _make_repo(tmp_path)
505 add_dependencies(repo, _A, [_B])
506 with pytest.raises(ValueError, match="cycle"):
507 add_dependencies(repo, _B, [_A])
508
509 def test_rejects_cycle_three_node(self, tmp_path: pathlib.Path) -> None:
510 """A → B → C → A must be rejected at the third edge."""
511 repo = _make_repo(tmp_path)
512 add_dependencies(repo, _A, [_B])
513 add_dependencies(repo, _B, [_C])
514 with pytest.raises(ValueError, match="cycle"):
515 add_dependencies(repo, _C, [_A])
516
517 def test_allows_diamond(self, tmp_path: pathlib.Path) -> None:
518 """A → B, A → C, B → D, C → D is a valid DAG (fan-in, not a cycle)."""
519 repo = _make_repo(tmp_path)
520 add_dependencies(repo, _A, [_B, _C])
521 add_dependencies(repo, _B, [_D])
522 add_dependencies(repo, _C, [_D]) # must not raise
523 g = load_dag(repo)
524 assert detect_cycle(g) is None
525
526
527 # ── load_dag ───────────────────────────────────────────────────────────────────
528
529
530 class TestLoadDag:
531 def test_empty_repo(self, tmp_path: pathlib.Path) -> None:
532 repo = _make_repo(tmp_path)
533 assert load_dag(repo) == {}
534
535 def test_single_record(self, tmp_path: pathlib.Path) -> None:
536 repo = _make_repo(tmp_path)
537 add_dependencies(repo, _A, [_B])
538 g = load_dag(repo)
539 assert g[_A] == {_B}
540 assert _B in g
541
542 def test_multiple_records(self, tmp_path: pathlib.Path) -> None:
543 repo = _make_repo(tmp_path)
544 add_dependencies(repo, _A, [_B])
545 add_dependencies(repo, _B, [_C])
546 g = load_dag(repo)
547 assert g[_A] == {_B}
548 assert g[_B] == {_C}
549
550
551 # ── Integration: full lifecycle ────────────────────────────────────────────────
552
553
554 class TestIntegration:
555 """End-to-end: create deps → load → compute order → check blocked."""
556
557 def test_pipeline_ordering(self, tmp_path: pathlib.Path) -> None:
558 """Five-stage pipeline: E→D, D→C, C→B, B→A. Topo order: A,B,C,D,E."""
559 repo = _make_repo(tmp_path)
560 add_dependencies(repo, _B, [_A])
561 add_dependencies(repo, _C, [_B])
562 add_dependencies(repo, _D, [_C])
563 add_dependencies(repo, _E, [_D])
564
565 g = load_dag(repo)
566 assert detect_cycle(g) is None
567
568 order = topological_sort(g)
569 assert order.index(_A) < order.index(_B)
570 assert order.index(_B) < order.index(_C)
571 assert order.index(_C) < order.index(_D)
572 assert order.index(_D) < order.index(_E)
573
574 def test_blocked_status_with_active_reservations(self, tmp_path: pathlib.Path) -> None:
575 """B depends on A; A is active → B is blocked; A is active → not blocked."""
576 repo = _make_repo(tmp_path)
577 add_dependencies(repo, _B, [_A])
578 g = load_dag(repo)
579
580 # A is still active
581 assert is_blocked(_B, g, frozenset({_A})) is True
582 assert get_blocking(_B, g, frozenset({_A})) == [_A]
583
584 # A has been released
585 assert is_blocked(_B, g, frozenset()) is False
586 assert get_blocking(_B, g, frozenset()) == []
587
588 def test_unblocked_when_no_deps(self, tmp_path: pathlib.Path) -> None:
589 repo = _make_repo(tmp_path)
590 add_dependencies(repo, _A, [])
591 g = load_dag(repo)
592 assert is_blocked(_A, g, frozenset({_B, _C})) is False
593
594 def test_diamond_blocked_status(self, tmp_path: pathlib.Path) -> None:
595 """A → B, A → C, B → D, C → D. When D is active, B and C are blocked."""
596 repo = _make_repo(tmp_path)
597 add_dependencies(repo, _A, [_B, _C])
598 add_dependencies(repo, _B, [_D])
599 add_dependencies(repo, _C, [_D])
600 g = load_dag(repo)
601
602 active = frozenset({_D})
603 # is_blocked checks DIRECT deps only.
604 # A's direct deps are B and C — neither is in active_ids{D}.
605 assert is_blocked(_A, g, active) is False
606 assert is_blocked(_B, g, active) is True # B depends on D (active)
607 assert is_blocked(_C, g, active) is True # C depends on D (active)
608 assert is_blocked(_D, g, active) is False # D has no deps
609
610
611 # ── Security tests ─────────────────────────────────────────────────────────────
612
613
614 class TestSecurity:
615 def test_path_traversal_in_add_reservation_id(self, tmp_path: pathlib.Path) -> None:
616 repo = _make_repo(tmp_path)
617 with pytest.raises(ValueError):
618 add_dependencies(repo, "../../etc/passwd", [_B])
619
620 def test_path_traversal_in_dep_id(self, tmp_path: pathlib.Path) -> None:
621 repo = _make_repo(tmp_path)
622 with pytest.raises(ValueError):
623 add_dependencies(repo, _A, ["../../harm"])
624
625 def test_null_byte_in_reservation_id(self, tmp_path: pathlib.Path) -> None:
626 repo = _make_repo(tmp_path)
627 with pytest.raises(ValueError):
628 add_dependencies(repo, "\x00" * 36, [_B])
629
630 def test_null_byte_in_dep_id(self, tmp_path: pathlib.Path) -> None:
631 repo = _make_repo(tmp_path)
632 with pytest.raises(ValueError):
633 add_dependencies(repo, _A, ["\x00" * 36])
634
635 def test_path_traversal_in_load_dependencies(self, tmp_path: pathlib.Path) -> None:
636 repo = _make_repo(tmp_path)
637 with pytest.raises(ValueError):
638 load_dependencies(repo, "../../etc/shadow")
639
640 def test_cycle_cannot_be_written(self, tmp_path: pathlib.Path) -> None:
641 """Even under rapid concurrent conditions, a cycle must never reach disk."""
642 repo = _make_repo(tmp_path)
643 add_dependencies(repo, _A, [_B])
644 add_dependencies(repo, _B, [_C])
645 # This would create A→B→C→A
646 with pytest.raises(ValueError, match="cycle"):
647 add_dependencies(repo, _C, [_A])
648 # Verify C has no record on disk
649 assert (_dependencies_dir(repo) / f"{_C}.json").exists() is False
650
651
652 # ── CLI: muse coord dag ────────────────────────────────────────────────────────
653
654
655 class TestCliDag:
656 def _make_deps(self, repo: pathlib.Path) -> None:
657 add_dependencies(repo, _B, [_A])
658 add_dependencies(repo, _C, [_B])
659
660 def test_full_dag_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
661 repo = _make_repo(tmp_path)
662 self._make_deps(repo)
663 args = _dag_ns(fmt="json")
664 with _patch_dag_repo(repo):
665 dag_run(args)
666 out = json.loads(capsys.readouterr().out)
667 assert out["total_nodes"] == 3
668 assert out["total_edges"] == 2
669 assert "nodes" in out
670
671 def test_full_dag_text(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
672 repo = _make_repo(tmp_path)
673 self._make_deps(repo)
674 args = _dag_ns(fmt="text")
675 with _patch_dag_repo(repo):
676 dag_run(args)
677 out = capsys.readouterr().out
678 assert "Dependency DAG" in out
679
680 def test_empty_dag_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
681 repo = _make_repo(tmp_path)
682 args = _dag_ns(fmt="json")
683 with _patch_dag_repo(repo):
684 dag_run(args)
685 out = json.loads(capsys.readouterr().out)
686 assert out["total_nodes"] == 0
687 assert out["nodes"] == []
688
689 def test_single_reservation_mode_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
690 repo = _make_repo(tmp_path)
691 add_dependencies(repo, _B, [_A])
692 args = _dag_ns(fmt="json", reservation_id=_B)
693 with _patch_dag_repo(repo):
694 dag_run(args)
695 out = json.loads(capsys.readouterr().out)
696 assert out["reservation_id"] == _B
697 assert _A in out["depends_on"]
698
699 def test_single_reservation_mode_text(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
700 repo = _make_repo(tmp_path)
701 add_dependencies(repo, _B, [_A])
702 args = _dag_ns(fmt="text", reservation_id=_B)
703 with _patch_dag_repo(repo):
704 dag_run(args)
705 out = capsys.readouterr().out
706 # Should show the reservation ID prefix
707 assert _B[:8] in out
708
709 def test_cycle_detected_exits_1(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
710 """A cycle on disk causes exit code 1."""
711 repo = _make_repo(tmp_path)
712 # Manually write a cycle to disk, bypassing add_dependencies validation
713 ensure_dag_dirs(repo)
714 import json as _json
715 from muse._version import __version__
716 rec_a = {"schema_version": __version__, "reservation_id": _A,
717 "depends_on": [_B], "created_at": _EPOCH.isoformat()}
718 rec_b = {"schema_version": __version__, "reservation_id": _B,
719 "depends_on": [_A], "created_at": _EPOCH.isoformat()}
720 (_dependencies_dir(repo) / f"{_A}.json").write_text(_json.dumps(rec_a))
721 (_dependencies_dir(repo) / f"{_B}.json").write_text(_json.dumps(rec_b))
722
723 args = _dag_ns(fmt="json")
724 with _patch_dag_repo(repo):
725 with pytest.raises(SystemExit) as exc:
726 dag_run(args)
727 assert exc.value.code == 1
728
729 def test_blocked_count_in_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
730 """A reservation with an active dependency must show blocked_count > 0."""
731 repo = _make_repo(tmp_path)
732 add_dependencies(repo, _B, [_A])
733
734 # Inject _A as an active reservation via patching active_reservations.
735 from muse.core.coordination import Reservation
736 fake_res = Reservation(
737 reservation_id=_A,
738 run_id="agent-a",
739 branch="dev",
740 addresses=["foo.py::bar"],
741 created_at=_EPOCH,
742 expires_at=_EPOCH + datetime.timedelta(hours=1),
743 operation=None,
744 )
745 with _patch_dag_repo(repo), \
746 patch("muse.cli.commands.dag.active_reservations", return_value=[fake_res]):
747 args = _dag_ns(fmt="json")
748 dag_run(args)
749
750 out = json.loads(capsys.readouterr().out)
751 assert out["blocked_count"] == 1
752
753 def test_topo_index_in_nodes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
754 repo = _make_repo(tmp_path)
755 add_dependencies(repo, _B, [_A])
756 args = _dag_ns(fmt="json")
757 with _patch_dag_repo(repo):
758 dag_run(args)
759 out = json.loads(capsys.readouterr().out)
760 indices = [n["topo_index"] for n in out["nodes"]]
761 assert all(i is not None for i in indices)
762 # A must have lower index than B (A has no deps, B depends on A)
763 a_node = next(n for n in out["nodes"] if n["reservation_id"] == _A)
764 b_node = next(n for n in out["nodes"] if n["reservation_id"] == _B)
765 assert a_node["topo_index"] < b_node["topo_index"]
766
767 def test_ansi_not_in_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
768 """Node IDs stored on disk must not produce raw ANSI in text output."""
769 repo = _make_repo(tmp_path)
770 add_dependencies(repo, _B, [_A])
771 args = _dag_ns(fmt="text")
772 with _patch_dag_repo(repo):
773 dag_run(args)
774 out = capsys.readouterr().out
775 assert "\x1b" not in out
776
777
778 # ── CLI: muse coord reserve --depends-on ──────────────────────────────────────
779
780
781 class TestReserveWithDependsOn:
782 """reserve --depends-on writes both a reservation and a dependency record."""
783
784 def test_reserve_without_depends_on(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
785 repo = _make_repo(tmp_path)
786 args = _reserve_ns(depends_on=[], addresses=["foo.py::bar"])
787 with _patch_repo(repo):
788 reserve_run(args)
789 out = json.loads(capsys.readouterr().out)
790 assert out["depends_on"] == []
791 assert out["dependency_error"] is None
792
793 def test_reserve_with_single_dep(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
794 repo = _make_repo(tmp_path)
795 args = _reserve_ns(depends_on=[_A], addresses=["foo.py::bar"])
796 with _patch_repo(repo):
797 reserve_run(args)
798 out = json.loads(capsys.readouterr().out)
799 assert _A in out["depends_on"]
800 # Verify file exists on disk
801 res_id = out["reservation_id"]
802 dep_rec = load_dependencies(repo, res_id)
803 assert dep_rec is not None
804 assert _A in dep_rec.depends_on
805
806 def test_reserve_with_multiple_deps(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
807 repo = _make_repo(tmp_path)
808 args = _reserve_ns(depends_on=[_A, _B], addresses=["foo.py::bar"])
809 with _patch_repo(repo):
810 reserve_run(args)
811 out = json.loads(capsys.readouterr().out)
812 assert set(out["depends_on"]) == {_A, _B}
813
814 def test_reserve_invalid_dep_id_exits_1(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
815 """Invalid UUID in --depends-on → dependency_error set, exit 1."""
816 repo = _make_repo(tmp_path)
817 args = _reserve_ns(depends_on=["not-a-uuid"], addresses=["foo.py::bar"])
818 with _patch_repo(repo):
819 with pytest.raises(SystemExit) as exc:
820 reserve_run(args)
821 assert exc.value.code == 1
822 out = json.loads(capsys.readouterr().out)
823 assert out["dependency_error"] is not None
824
825 def test_reserve_dep_creates_dag_record(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
826 """The reservation ID from --depends-on is reflected in load_dag."""
827 repo = _make_repo(tmp_path)
828 # First reservation (will be the dependency)
829 args1 = _reserve_ns(depends_on=[], addresses=["a.py::fn"])
830 with _patch_repo(repo):
831 reserve_run(args1)
832 dep_id = json.loads(capsys.readouterr().out)["reservation_id"]
833
834 # Second reservation depends on the first
835 args2 = _reserve_ns(depends_on=[dep_id], addresses=["b.py::fn"])
836 with _patch_repo(repo):
837 reserve_run(args2)
838 new_id = json.loads(capsys.readouterr().out)["reservation_id"]
839
840 g = load_dag(repo)
841 assert dep_id in g[new_id]
842
843
844 # ── Stress tests ───────────────────────────────────────────────────────────────
845
846
847 class TestStress:
848 def test_linear_chain_100(self, tmp_path: pathlib.Path) -> None:
849 """100-node chain A0 → A1 → … → A99: no cycle, correct topo order."""
850 repo = _make_repo(tmp_path)
851 ids = [str(uuid.uuid4()) for _ in range(100)]
852 for i in range(1, len(ids)):
853 add_dependencies(repo, ids[i], [ids[i - 1]])
854
855 g = load_dag(repo)
856 assert detect_cycle(g) is None
857
858 order = topological_sort(g)
859 # ids[0] must come before ids[1] must come before … ids[99]
860 for i in range(1, len(ids)):
861 assert order.index(ids[i - 1]) < order.index(ids[i])
862
863 def test_wide_diamond_50(self, tmp_path: pathlib.Path) -> None:
864 """1 root, 50 middle nodes, 1 leaf: fan-out then fan-in."""
865 repo = _make_repo(tmp_path)
866 root_id = str(uuid.uuid4())
867 leaf_id = str(uuid.uuid4())
868 middles = [str(uuid.uuid4()) for _ in range(50)]
869
870 # root → each middle
871 for m in middles:
872 add_dependencies(repo, m, [root_id])
873 # leaf → each middle (leaf depends on all 50)
874 add_dependencies(repo, leaf_id, middles)
875
876 g = load_dag(repo)
877 assert detect_cycle(g) is None
878 order = topological_sort(g)
879 assert order.index(root_id) < order.index(middles[0])
880 assert order.index(middles[-1]) < order.index(leaf_id)
881
882 def test_add_500_independent_deps(self, tmp_path: pathlib.Path) -> None:
883 """500 reservations each with one independent dependency — quick scan."""
884 repo = _make_repo(tmp_path)
885 start = time.monotonic()
886 for _ in range(500):
887 add_dependencies(repo, str(uuid.uuid4()), [str(uuid.uuid4())])
888 elapsed = time.monotonic() - start
889 assert elapsed < 10.0, f"500 independent adds took {elapsed:.2f}s"
890
891 def test_load_dag_500_nodes(self, tmp_path: pathlib.Path) -> None:
892 """load_dag on 500 records must complete quickly."""
893 repo = _make_repo(tmp_path)
894 for _ in range(500):
895 add_dependencies(repo, str(uuid.uuid4()), [str(uuid.uuid4())])
896 start = time.monotonic()
897 g = load_dag(repo)
898 elapsed = time.monotonic() - start
899 assert len(g) >= 500
900 assert elapsed < 5.0, f"load_dag 500 nodes took {elapsed:.2f}s"
901
902 def test_topo_sort_1000_node_chain(self) -> None:
903 """In-memory topo sort of 1000-node linear chain is fast."""
904 ids = [str(uuid.uuid4()) for _ in range(1000)]
905 graph: AdjacencyMap = {}
906 for i, rid in enumerate(ids):
907 graph[rid] = {ids[i - 1]} if i > 0 else set()
908
909 start = time.monotonic()
910 order = topological_sort(graph)
911 elapsed = time.monotonic() - start
912 assert len(order) == 1000
913 assert elapsed < 1.0, f"topo sort 1000 nodes took {elapsed:.2f}s"
914
915 def test_detect_cycle_1000_node_acyclic(self) -> None:
916 """Cycle detection on a 1000-node chain (no cycle) must be fast."""
917 ids = [str(uuid.uuid4()) for _ in range(1000)]
918 graph: AdjacencyMap = {}
919 for i, rid in enumerate(ids):
920 graph[rid] = {ids[i - 1]} if i > 0 else set()
921
922 start = time.monotonic()
923 result = detect_cycle(graph)
924 elapsed = time.monotonic() - start
925 assert result is None
926 assert elapsed < 1.0, f"detect_cycle 1000 nodes took {elapsed:.2f}s"
927
928 def test_dag_json_200_active_nodes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
929 """CLI dag with 200 nodes completes quickly."""
930 repo = _make_repo(tmp_path)
931 ids = [str(uuid.uuid4()) for _ in range(200)]
932 for i in range(1, len(ids)):
933 add_dependencies(repo, ids[i], [ids[i - 1]])
934 args = _dag_ns(fmt="json")
935 start = time.monotonic()
936 with _patch_dag_repo(repo):
937 dag_run(args)
938 elapsed = time.monotonic() - start
939 out = json.loads(capsys.readouterr().out)
940 assert out["total_nodes"] == 200
941 assert elapsed < 5.0, f"200-node dag JSON took {elapsed:.2f}s"
942
943
944 # ── New: input validation ──────────────────────────────────────────────────────
945
946
947 class TestCliDagInputValidation:
948 """UUID validation on --reservation-id fires before any file I/O."""
949
950 def test_valid_uuid_accepted_single_mode(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
951 repo = _make_repo(tmp_path)
952 add_dependencies(repo, _B, [_A])
953 args = _dag_ns(fmt="json", reservation_id=_B)
954 with _patch_dag_repo(repo):
955 dag_run(args)
956 out = json.loads(capsys.readouterr().out)
957 assert out["reservation_id"] == _B
958
959 def test_invalid_uuid_exits_1_text(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
960 repo = _make_repo(tmp_path)
961 args = _dag_ns(fmt="text", reservation_id="not-a-uuid")
962 with _patch_dag_repo(repo):
963 with pytest.raises(SystemExit) as exc:
964 dag_run(args)
965 assert exc.value.code == 1
966 err = capsys.readouterr().err
967 assert "❌" in err
968
969 def test_invalid_uuid_exits_1_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
970 repo = _make_repo(tmp_path)
971 args = _dag_ns(fmt="json", reservation_id="not-a-uuid")
972 with _patch_dag_repo(repo):
973 with pytest.raises(SystemExit) as exc:
974 dag_run(args)
975 assert exc.value.code == 1
976 out = json.loads(capsys.readouterr().out)
977 assert "error" in out
978 assert out["status"] == "bad_reservation_id"
979
980 def test_path_traversal_exits_1(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
981 repo = _make_repo(tmp_path)
982 args = _dag_ns(fmt="json", reservation_id="../../etc/passwd")
983 with _patch_dag_repo(repo):
984 with pytest.raises(SystemExit) as exc:
985 dag_run(args)
986 assert exc.value.code == 1
987
988 def test_null_byte_in_reservation_id_exits_1(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
989 repo = _make_repo(tmp_path)
990 args = _dag_ns(fmt="json", reservation_id="\x00" * 36)
991 with _patch_dag_repo(repo):
992 with pytest.raises(SystemExit) as exc:
993 dag_run(args)
994 assert exc.value.code == 1
995
996 def test_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
997 """require_repo must never be called when --reservation-id is invalid."""
998 repo = _make_repo(tmp_path)
999 args = _dag_ns(fmt="json", reservation_id="bad-uuid")
1000 require_calls: list[bool] = []
1001
1002 def _fake_require() -> pathlib.Path:
1003 require_calls.append(True)
1004 return repo
1005
1006 with patch("muse.cli.commands.dag.require_repo", side_effect=_fake_require):
1007 with pytest.raises(SystemExit):
1008 dag_run(args)
1009 assert require_calls == [], "require_repo was called before validation"
1010
1011 def test_none_reservation_id_shows_full_dag(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1012 repo = _make_repo(tmp_path)
1013 add_dependencies(repo, _B, [_A])
1014 args = _dag_ns(fmt="json", reservation_id=None)
1015 with _patch_dag_repo(repo):
1016 dag_run(args)
1017 out = json.loads(capsys.readouterr().out)
1018 assert "nodes" in out
1019 assert out["total_nodes"] > 0
1020
1021 def test_json_error_is_compact_single_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1022 """JSON error output must be on a single line (compact, no indent)."""
1023 repo = _make_repo(tmp_path)
1024 args = _dag_ns(fmt="json", reservation_id="bad")
1025 with _patch_dag_repo(repo):
1026 with pytest.raises(SystemExit):
1027 dag_run(args)
1028 raw = capsys.readouterr().out.strip()
1029 assert "\n" not in raw
1030 data = json.loads(raw)
1031 assert "error" in data
1032
1033
1034 # ── New: compact JSON output ───────────────────────────────────────────────────
1035
1036
1037 class TestCliDagCompactJson:
1038 """All JSON output must be compact (no indent=2)."""
1039
1040 def test_full_dag_json_is_single_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1041 repo = _make_repo(tmp_path)
1042 add_dependencies(repo, _B, [_A])
1043 args = _dag_ns(fmt="json")
1044 with _patch_dag_repo(repo):
1045 dag_run(args)
1046 raw = capsys.readouterr().out.strip()
1047 assert "\n" not in raw
1048 json.loads(raw) # must be valid JSON
1049
1050 def test_single_mode_json_is_single_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1051 repo = _make_repo(tmp_path)
1052 add_dependencies(repo, _B, [_A])
1053 args = _dag_ns(fmt="json", reservation_id=_B)
1054 with _patch_dag_repo(repo):
1055 dag_run(args)
1056 raw = capsys.readouterr().out.strip()
1057 assert "\n" not in raw
1058 json.loads(raw)
1059
1060 def test_empty_dag_json_is_single_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1061 repo = _make_repo(tmp_path)
1062 args = _dag_ns(fmt="json")
1063 with _patch_dag_repo(repo):
1064 dag_run(args)
1065 raw = capsys.readouterr().out.strip()
1066 assert "\n" not in raw
1067 data = json.loads(raw)
1068 assert data["total_nodes"] == 0
1069
1070 def test_full_dag_json_schema_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1071 repo = _make_repo(tmp_path)
1072 add_dependencies(repo, _B, [_A])
1073 args = _dag_ns(fmt="json")
1074 with _patch_dag_repo(repo):
1075 dag_run(args)
1076 out = json.loads(capsys.readouterr().out)
1077 for key in ("schema_version", "total_nodes", "total_edges",
1078 "blocked_count", "active_only", "cycle", "nodes"):
1079 assert key in out, f"missing key: {key}"
1080
1081 def test_single_mode_json_schema_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1082 repo = _make_repo(tmp_path)
1083 add_dependencies(repo, _B, [_A])
1084 args = _dag_ns(fmt="json", reservation_id=_B)
1085 with _patch_dag_repo(repo):
1086 dag_run(args)
1087 out = json.loads(capsys.readouterr().out)
1088 for key in ("reservation_id", "depends_on", "active", "blocked", "blocking", "cycle"):
1089 assert key in out, f"missing key: {key}"
1090
1091 def test_active_only_field_in_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1092 repo = _make_repo(tmp_path)
1093 args = _dag_ns(fmt="json", active_only=False)
1094 with _patch_dag_repo(repo):
1095 dag_run(args)
1096 out = json.loads(capsys.readouterr().out)
1097 assert out["active_only"] is False
1098
1099 def test_active_only_true_reflected_in_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1100 repo = _make_repo(tmp_path)
1101 args = _dag_ns(fmt="json", active_only=True)
1102 with _patch_dag_repo(repo):
1103 dag_run(args)
1104 out = json.loads(capsys.readouterr().out)
1105 assert out["active_only"] is True
1106
1107
1108 # ── New: --active-only flag ────────────────────────────────────────────────────
1109
1110
1111 class TestCliDagActiveOnly:
1112 """--active-only restricts the graph to currently active reservations."""
1113
1114 def _make_fake_active(self, reservation_id: str) -> Reservation:
1115 return Reservation(
1116 reservation_id=reservation_id,
1117 run_id="agent-test",
1118 branch="dev",
1119 addresses=["foo.py::bar"],
1120 created_at=_EPOCH,
1121 expires_at=_EPOCH + datetime.timedelta(hours=1),
1122 operation=None,
1123 )
1124
1125 def test_active_only_false_shows_all_nodes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1126 repo = _make_repo(tmp_path)
1127 add_dependencies(repo, _B, [_A])
1128 args = _dag_ns(fmt="json", active_only=False)
1129 with _patch_dag_repo(repo):
1130 dag_run(args)
1131 out = json.loads(capsys.readouterr().out)
1132 # Both _A and _B must be in graph (neither is active, no filter)
1133 node_ids = {n["reservation_id"] for n in out["nodes"]}
1134 assert _A in node_ids and _B in node_ids
1135
1136 def test_active_only_true_empty_active_set(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1137 repo = _make_repo(tmp_path)
1138 add_dependencies(repo, _B, [_A])
1139 args = _dag_ns(fmt="json", active_only=True)
1140 with _patch_dag_repo(repo), \
1141 patch("muse.cli.commands.dag.active_reservations", return_value=[]):
1142 dag_run(args)
1143 out = json.loads(capsys.readouterr().out)
1144 # No active reservations → empty graph
1145 assert out["total_nodes"] == 0
1146 assert out["nodes"] == []
1147
1148 def test_active_only_filters_to_active_nodes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1149 repo = _make_repo(tmp_path)
1150 add_dependencies(repo, _B, [_A])
1151 add_dependencies(repo, _C, [_B])
1152 # Only _B is active
1153 fake_active = [self._make_fake_active(_B)]
1154 args = _dag_ns(fmt="json", active_only=True)
1155 with _patch_dag_repo(repo), \
1156 patch("muse.cli.commands.dag.active_reservations", return_value=fake_active):
1157 dag_run(args)
1158 out = json.loads(capsys.readouterr().out)
1159 node_ids = {n["reservation_id"] for n in out["nodes"]}
1160 assert _B in node_ids
1161 assert _A not in node_ids
1162 assert _C not in node_ids
1163
1164 def test_active_only_all_active_shows_all(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1165 repo = _make_repo(tmp_path)
1166 add_dependencies(repo, _B, [_A])
1167 fake_active = [self._make_fake_active(_A), self._make_fake_active(_B)]
1168 args = _dag_ns(fmt="json", active_only=True)
1169 with _patch_dag_repo(repo), \
1170 patch("muse.cli.commands.dag.active_reservations", return_value=fake_active):
1171 dag_run(args)
1172 out = json.loads(capsys.readouterr().out)
1173 node_ids = {n["reservation_id"] for n in out["nodes"]}
1174 assert _A in node_ids and _B in node_ids
1175
1176 def test_active_only_reflected_true_in_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1177 repo = _make_repo(tmp_path)
1178 args = _dag_ns(fmt="json", active_only=True)
1179 with _patch_dag_repo(repo), \
1180 patch("muse.cli.commands.dag.active_reservations", return_value=[]):
1181 dag_run(args)
1182 out = json.loads(capsys.readouterr().out)
1183 assert out["active_only"] is True
1184
1185 def test_active_only_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1186 repo = _make_repo(tmp_path)
1187 add_dependencies(repo, _B, [_A])
1188 add_dependencies(repo, _C, [_B])
1189 # Only _B is active — _A and _C are filtered from graph nodes
1190 fake_active = [self._make_fake_active(_B)]
1191 args = _dag_ns(fmt="json", active_only=True)
1192 with _patch_dag_repo(repo), \
1193 patch("muse.cli.commands.dag.active_reservations", return_value=fake_active):
1194 dag_run(args)
1195 out = json.loads(capsys.readouterr().out)
1196 node_ids = {n["reservation_id"] for n in out["nodes"]}
1197 # Only _B passes the active-only filter
1198 assert _B in node_ids
1199 assert _C not in node_ids
1200 assert _A not in node_ids
1201 assert out["total_nodes"] == 1
1202
1203
1204 # ── New: --topo flag ───────────────────────────────────────────────────────────
1205
1206
1207 class TestCliDagTopoFlag:
1208 """--topo adds a TOPO column; without it a flat table is shown."""
1209
1210 def test_without_topo_no_topo_column(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1211 repo = _make_repo(tmp_path)
1212 add_dependencies(repo, _B, [_A])
1213 args = _dag_ns(fmt="text", topo=False)
1214 with _patch_dag_repo(repo):
1215 dag_run(args)
1216 out = capsys.readouterr().out
1217 # flat table has STATUS column but no TOPO column header
1218 assert "STATUS" in out
1219 assert "TOPO" not in out
1220
1221 def test_with_topo_shows_topo_column(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1222 repo = _make_repo(tmp_path)
1223 add_dependencies(repo, _B, [_A])
1224 args = _dag_ns(fmt="text", topo=True)
1225 with _patch_dag_repo(repo):
1226 dag_run(args)
1227 out = capsys.readouterr().out
1228 assert "TOPO" in out
1229 assert "STATUS" in out
1230
1231 def test_topo_flag_does_not_affect_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1232 """JSON output is identical regardless of --topo (topo_index always present)."""
1233 repo = _make_repo(tmp_path)
1234 add_dependencies(repo, _B, [_A])
1235
1236 args_with = _dag_ns(fmt="json", topo=True)
1237 with _patch_dag_repo(repo):
1238 dag_run(args_with)
1239 out_with = json.loads(capsys.readouterr().out)
1240
1241 args_without = _dag_ns(fmt="json", topo=False)
1242 with _patch_dag_repo(repo):
1243 dag_run(args_without)
1244 out_without = json.loads(capsys.readouterr().out)
1245
1246 # Both must have topo_index on every node
1247 for node in out_with["nodes"]:
1248 assert node["topo_index"] is not None
1249 for node in out_without["nodes"]:
1250 assert node["topo_index"] is not None
1251
1252 def test_without_topo_nodes_shown_in_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1253 repo = _make_repo(tmp_path)
1254 add_dependencies(repo, _B, [_A])
1255 args = _dag_ns(fmt="text", topo=False)
1256 with _patch_dag_repo(repo):
1257 dag_run(args)
1258 out = capsys.readouterr().out
1259 # Node IDs (truncated) must appear
1260 assert _B[:8] in out
1261
1262 def test_with_topo_node_id_in_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1263 repo = _make_repo(tmp_path)
1264 add_dependencies(repo, _B, [_A])
1265 args = _dag_ns(fmt="text", topo=True)
1266 with _patch_dag_repo(repo):
1267 dag_run(args)
1268 out = capsys.readouterr().out
1269 assert _B[:8] in out
1270
1271 def test_topo_order_correct_in_text(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1272 """With --topo, nodes with no deps must appear before dependents."""
1273 repo = _make_repo(tmp_path)
1274 add_dependencies(repo, _B, [_A])
1275 args = _dag_ns(fmt="text", topo=True)
1276 with _patch_dag_repo(repo):
1277 dag_run(args)
1278 out = capsys.readouterr().out
1279 # _A has lower topo index → must appear before _B in output
1280 pos_a = out.find(_A[:8])
1281 pos_b = out.find(_B[:8])
1282 assert pos_a < pos_b, "dependency must appear before dependent in topo mode"
1283
1284 def test_empty_dag_topo_no_crash(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1285 repo = _make_repo(tmp_path)
1286 args = _dag_ns(fmt="text", topo=True)
1287 with _patch_dag_repo(repo):
1288 dag_run(args)
1289 out = capsys.readouterr().out
1290 assert "no dependency records" in out
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago