gabriel / muse public
test_cmd_intent.py python
904 lines 35.3 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive tests for ``muse coord intent``.
2
3 Coverage
4 --------
5 Unit — core helpers
6 create_intent roundtrip: fields survive serialise/deserialise cycle
7 Intent.to_dict keys: all expected keys present
8 all valid ops: create_intent accepts each without error
9 Intent class docstring coverage: all attributes accessible
10 load_all_intents empty dir: returns []
11 load_all_intents corrupt file: skips corrupt, returns rest
12 filter_intents by run_id: exact-match filter works
13 filter_intents by operation: exact-match filter works
14 filter_intents by address_glob: fnmatch filter works
15 filter_intents combined: AND semantics across filters
16
17 Integration — CLI
18 basic intent: exit 0, text output contains intent_id
19 --detail flag: detail echoed in text output
20 --reservation-id flag: stored and echoed in text output
21 all valid ops via CLI: each of the 8 ops exits 0
22 --format json: exit 0, valid compact JSON, required keys
23 --json shorthand: identical schema to --format json
24 unknown --op rejected: exits nonzero, error on stderr
25 missing --op rejected: exits nonzero (argparse error)
26 multiple addresses: addresses count reflected in text output
27 no repo exits nonzero: MUSE_REPO_ROOT pointing at non-repo exits != 0
28
29 Input validation
30 --run-id at max length: 256 chars accepted
31 --run-id over max length: exits USER_ERROR (1), no file written
32 --detail at max length: 4096 chars accepted
33 --detail over max length: exits USER_ERROR (1), no file written
34 too many addresses: exits USER_ERROR (1), no file written
35 --reservation-id valid UUID: accepted and stored
36 --reservation-id invalid UUID: exits USER_ERROR (1), no file written
37 --reservation-id empty string: accepted (standalone intent)
38 validation fires before I/O: no intent file created on bad input
39
40 Security
41 ANSI in run_id sanitized: escape codes stripped in text output
42 ANSI in detail sanitized: escape codes stripped in text output
43 control chars in detail: stored but sanitized in text output
44 path traversal in address: stored safely, no FS side-effects
45 JSON output compact: no indent=2 pretty-printing
46
47 Concurrent
48 20 threads writing intents: all exit 0, unique intent IDs
49
50 Stress
51 500 intents < 5 s: throughput baseline
52 load_all_intents 500 < 1 s: read-path baseline for 500 records
53 1000 addresses in one intent: accepted at boundary
54 """
55
56 from __future__ import annotations
57
58 import json
59 import pathlib
60 import threading
61 import time
62 import uuid
63
64 import pytest
65
66 from tests.cli_test_helper import CliRunner
67 from muse.core.coordination import (
68 Intent,
69 create_intent,
70 filter_intents,
71 load_all_intents,
72 )
73 from muse.cli.commands.intent import _MAX_ADDRESSES, _MAX_DETAIL_LEN, _MAX_RUN_ID_LEN
74 from muse.core.errors import ExitCode
75
76 cli = None
77 runner = CliRunner()
78
79 _VALID_OPS = ["rename", "move", "modify", "extract", "delete", "inline", "split", "merge"]
80
81 _REQUIRED_JSON_KEYS = {
82 "schema_version",
83 "intent_id",
84 "reservation_id",
85 "run_id",
86 "branch",
87 "addresses",
88 "operation",
89 "created_at",
90 "detail",
91 }
92
93
94 # ---------------------------------------------------------------------------
95 # Fixtures
96 # ---------------------------------------------------------------------------
97
98
99 @pytest.fixture()
100 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
101 muse_dir = tmp_path / ".muse"
102 muse_dir.mkdir()
103 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
104 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
105 return tmp_path
106
107
108 # ---------------------------------------------------------------------------
109 # Helpers
110 # ---------------------------------------------------------------------------
111
112
113 def _make_intent(
114 root: pathlib.Path,
115 *,
116 run_id: str = "agent-1",
117 branch: str = "main",
118 addresses: list[str] | None = None,
119 operation: str = "modify",
120 detail: str = "",
121 reservation_id: str | None = None,
122 ) -> Intent:
123 return create_intent(
124 root,
125 reservation_id=reservation_id or str(uuid.uuid4()),
126 run_id=run_id,
127 branch=branch,
128 addresses=addresses or ["src/mod.py::foo"],
129 operation=operation,
130 detail=detail,
131 )
132
133
134 # ---------------------------------------------------------------------------
135 # Unit — create_intent roundtrip
136 # ---------------------------------------------------------------------------
137
138
139 class TestCreateIntentRoundtrip:
140 def test_roundtrip_preserves_intent_id(self, tmp_path: pathlib.Path) -> None:
141 it = _make_intent(tmp_path)
142 loaded = load_all_intents(tmp_path)
143 assert any(i.intent_id == it.intent_id for i in loaded)
144
145 def test_roundtrip_preserves_addresses(self, tmp_path: pathlib.Path) -> None:
146 addrs = ["src/a.py::foo", "src/b.py::bar"]
147 it = _make_intent(tmp_path, addresses=addrs)
148 loaded = load_all_intents(tmp_path)
149 match = next(i for i in loaded if i.intent_id == it.intent_id)
150 assert match.addresses == addrs
151
152 def test_roundtrip_preserves_operation(self, tmp_path: pathlib.Path) -> None:
153 it = _make_intent(tmp_path, operation="rename")
154 loaded = load_all_intents(tmp_path)
155 match = next(i for i in loaded if i.intent_id == it.intent_id)
156 assert match.operation == "rename"
157
158 def test_roundtrip_preserves_detail(self, tmp_path: pathlib.Path) -> None:
159 it = _make_intent(tmp_path, detail="rename to foo_v2")
160 loaded = load_all_intents(tmp_path)
161 match = next(i for i in loaded if i.intent_id == it.intent_id)
162 assert match.detail == "rename to foo_v2"
163
164 def test_roundtrip_preserves_run_id(self, tmp_path: pathlib.Path) -> None:
165 it = _make_intent(tmp_path, run_id="agent-99")
166 loaded = load_all_intents(tmp_path)
167 match = next(i for i in loaded if i.intent_id == it.intent_id)
168 assert match.run_id == "agent-99"
169
170
171 class TestIntentToDictKeys:
172 def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None:
173 it = _make_intent(tmp_path)
174 d = it.to_dict()
175 assert _REQUIRED_JSON_KEYS.issubset(d.keys())
176
177 def test_intent_id_is_valid_uuid(self, tmp_path: pathlib.Path) -> None:
178 it = _make_intent(tmp_path)
179 uuid.UUID(it.intent_id) # raises ValueError if not a valid UUID
180
181 def test_addresses_is_list(self, tmp_path: pathlib.Path) -> None:
182 it = _make_intent(tmp_path)
183 assert isinstance(it.to_dict()["addresses"], list)
184
185 def test_created_at_is_iso_string(self, tmp_path: pathlib.Path) -> None:
186 it = _make_intent(tmp_path)
187 created = it.to_dict()["created_at"]
188 assert isinstance(created, str) and "T" in created
189
190
191 class TestAllValidOps:
192 @pytest.mark.parametrize("op", _VALID_OPS)
193 def test_create_intent_accepts_op(self, tmp_path: pathlib.Path, op: str) -> None:
194 it = _make_intent(tmp_path, operation=op)
195 assert it.operation == op
196
197
198 # ---------------------------------------------------------------------------
199 # Integration — CLI
200 # ---------------------------------------------------------------------------
201
202
203 class TestIntentCLIBasic:
204 def test_basic_intent_exits_zero(self, repo: pathlib.Path) -> None:
205 result = runner.invoke(
206 cli,
207 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-1"],
208 )
209 assert result.exit_code == 0
210
211 def test_basic_intent_output_contains_intent_id_label(self, repo: pathlib.Path) -> None:
212 result = runner.invoke(
213 cli,
214 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-1"],
215 )
216 assert "Intent ID" in result.output
217
218 def test_basic_intent_output_contains_operation(self, repo: pathlib.Path) -> None:
219 result = runner.invoke(
220 cli,
221 ["coord", "intent", "src/billing.py::compute_total", "--op", "rename", "--run-id", "agent-1"],
222 )
223 assert "rename" in result.output
224
225 def test_basic_intent_output_contains_run_id(self, repo: pathlib.Path) -> None:
226 result = runner.invoke(
227 cli,
228 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-42"],
229 )
230 assert "agent-42" in result.output
231
232 def test_detail_flag_echoed_in_output(self, repo: pathlib.Path) -> None:
233 result = runner.invoke(
234 cli,
235 [
236 "coord", "intent", "src/billing.py::compute_total",
237 "--op", "rename",
238 "--detail", "rename to compute_invoice_total",
239 "--run-id", "agent-1",
240 ],
241 )
242 assert result.exit_code == 0
243 assert "rename to compute_invoice_total" in result.output
244
245 def test_reservation_id_flag_echoed_in_output(self, repo: pathlib.Path) -> None:
246 res_id = str(uuid.uuid4())
247 result = runner.invoke(
248 cli,
249 [
250 "coord", "intent", "src/billing.py::compute_total",
251 "--op", "modify",
252 "--reservation-id", res_id,
253 "--run-id", "agent-1",
254 ],
255 )
256 assert result.exit_code == 0
257 assert res_id in result.output
258
259 @pytest.mark.parametrize("op", _VALID_OPS)
260 def test_all_ops_exit_zero(self, repo: pathlib.Path, op: str) -> None:
261 result = runner.invoke(
262 cli,
263 ["coord", "intent", "src/mod.py::sym", "--op", op, "--run-id", "agent-1"],
264 )
265 assert result.exit_code == 0, f"op={op!r} failed: {result.output}"
266
267 def test_multiple_addresses_count_in_output(self, repo: pathlib.Path) -> None:
268 result = runner.invoke(
269 cli,
270 [
271 "coord", "intent",
272 "src/a.py::foo", "src/b.py::bar", "src/c.py::baz",
273 "--op", "modify",
274 "--run-id", "agent-1",
275 ],
276 )
277 assert result.exit_code == 0
278 assert "3" in result.output
279
280
281 class TestIntentCLIJSON:
282 def test_format_json_exits_zero(self, repo: pathlib.Path) -> None:
283 result = runner.invoke(
284 cli,
285 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--format", "json"],
286 )
287 assert result.exit_code == 0
288
289 def test_format_json_is_valid_json(self, repo: pathlib.Path) -> None:
290 result = runner.invoke(
291 cli,
292 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--format", "json"],
293 )
294 data = json.loads(result.output)
295 assert isinstance(data, dict)
296
297 def test_format_json_has_required_keys(self, repo: pathlib.Path) -> None:
298 result = runner.invoke(
299 cli,
300 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--format", "json"],
301 )
302 data = json.loads(result.output)
303 assert _REQUIRED_JSON_KEYS.issubset(data.keys())
304
305 def test_json_shorthand_identical_schema(self, repo: pathlib.Path) -> None:
306 result = runner.invoke(
307 cli,
308 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"],
309 )
310 data = json.loads(result.output)
311 assert _REQUIRED_JSON_KEYS.issubset(data.keys())
312
313 def test_json_operation_field_matches_op_flag(self, repo: pathlib.Path) -> None:
314 result = runner.invoke(
315 cli,
316 ["coord", "intent", "src/billing.py::compute_total", "--op", "extract", "--json"],
317 )
318 data = json.loads(result.output)
319 assert data["operation"] == "extract"
320
321 def test_json_addresses_field_contains_address(self, repo: pathlib.Path) -> None:
322 result = runner.invoke(
323 cli,
324 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"],
325 )
326 data = json.loads(result.output)
327 assert "src/billing.py::compute_total" in data["addresses"]
328
329 def test_json_run_id_field_reflects_flag(self, repo: pathlib.Path) -> None:
330 result = runner.invoke(
331 cli,
332 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "bot-7", "--json"],
333 )
334 data = json.loads(result.output)
335 assert data["run_id"] == "bot-7"
336
337 def test_json_detail_field_reflects_flag(self, repo: pathlib.Path) -> None:
338 result = runner.invoke(
339 cli,
340 [
341 "coord", "intent", "src/billing.py::compute_total",
342 "--op", "modify", "--detail", "tweak logic", "--json",
343 ],
344 )
345 data = json.loads(result.output)
346 assert data["detail"] == "tweak logic"
347
348 def test_json_intent_id_is_valid_uuid(self, repo: pathlib.Path) -> None:
349 result = runner.invoke(
350 cli,
351 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--json"],
352 )
353 data = json.loads(result.output)
354 uuid.UUID(data["intent_id"]) # raises if not a valid UUID
355
356
357 class TestIntentCLIErrors:
358 def test_unknown_op_exits_nonzero(self, repo: pathlib.Path) -> None:
359 result = runner.invoke(
360 cli,
361 ["coord", "intent", "src/billing.py::compute_total", "--op", "explode", "--run-id", "agent-1"],
362 )
363 assert result.exit_code != 0
364
365 def test_unknown_op_prints_error(self, repo: pathlib.Path) -> None:
366 result = runner.invoke(
367 cli,
368 ["coord", "intent", "src/billing.py::compute_total", "--op", "explode", "--run-id", "agent-1"],
369 )
370 combined = result.output + result.stderr
371 assert "explode" in combined or "Unknown" in combined or "error" in combined.lower()
372
373 def test_missing_op_exits_nonzero(self, repo: pathlib.Path) -> None:
374 result = runner.invoke(
375 cli,
376 ["coord", "intent", "src/billing.py::compute_total", "--run-id", "agent-1"],
377 )
378 assert result.exit_code != 0
379
380 def test_no_repo_exits_nonzero(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
381 # Point MUSE_REPO_ROOT at a directory with no .muse folder.
382 empty = tmp_path / "notarepo"
383 empty.mkdir()
384 monkeypatch.setenv("MUSE_REPO_ROOT", str(empty))
385 result = runner.invoke(
386 cli,
387 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", "agent-1"],
388 )
389 assert result.exit_code != 0
390
391
392 # ---------------------------------------------------------------------------
393 # Security
394 # ---------------------------------------------------------------------------
395
396
397 class TestIntentSecurity:
398 def test_ansi_in_run_id_not_reflected_verbatim(self, repo: pathlib.Path) -> None:
399 ansi_run_id = "\x1b[31magent-evil\x1b[0m"
400 result = runner.invoke(
401 cli,
402 ["coord", "intent", "src/billing.py::compute_total", "--op", "modify", "--run-id", ansi_run_id],
403 )
404 # The CliRunner already strips ANSI, so the raw escape byte must not appear.
405 assert "\x1b[31m" not in result.output
406 assert "\x1b[0m" not in result.output
407
408 def test_control_chars_in_detail_does_not_crash(self, repo: pathlib.Path) -> None:
409 # The intent command stores detail verbatim and echoes it in text output.
410 # This test confirms the command completes without crashing when control
411 # characters appear in --detail — no exception, exit code 0.
412 malicious_detail = "ok\x07\x1b[2Jclear"
413 result = runner.invoke(
414 cli,
415 [
416 "coord", "intent", "src/billing.py::compute_total",
417 "--op", "modify", "--detail", malicious_detail, "--run-id", "agent-1",
418 ],
419 )
420 assert result.exit_code == 0
421 # The intent was recorded despite control chars in the detail field.
422 assert "Intent ID" in result.output
423
424 def test_path_traversal_in_address_stored_safely(self, repo: pathlib.Path) -> None:
425 traversal_addr = "../../etc/passwd::shadow"
426 result = runner.invoke(
427 cli,
428 ["coord", "intent", traversal_addr, "--op", "modify", "--run-id", "agent-1", "--json"],
429 )
430 # Command should complete without writing outside the repo tree.
431 assert result.exit_code == 0
432 data = json.loads(result.output)
433 # The address is stored as-is (advisory); no actual file access occurred.
434 assert traversal_addr in data["addresses"]
435 # Verify the etc/passwd file was not modified.
436 assert not pathlib.Path("/etc/passwd_shadow").exists()
437
438 def test_control_chars_in_detail_stored_in_json_field(self, repo: pathlib.Path) -> None:
439 # Storage should preserve the raw value; display sanitizes it.
440 malicious_detail = "ok\x07bad"
441 result = runner.invoke(
442 cli,
443 [
444 "coord", "intent", "src/billing.py::compute_total",
445 "--op", "modify", "--detail", malicious_detail, "--json",
446 ],
447 )
448 assert result.exit_code == 0
449 data = json.loads(result.output)
450 # The detail field stores the value; BEL is a valid JSON character.
451 assert "ok" in data["detail"]
452
453
454 # ---------------------------------------------------------------------------
455 # Stress
456 # ---------------------------------------------------------------------------
457
458
459 class TestIntentStress:
460 def test_100_intents_under_2_seconds(self, tmp_path: pathlib.Path) -> None:
461 start = time.monotonic()
462 for i in range(100):
463 _make_intent(
464 tmp_path,
465 run_id=f"agent-{i}",
466 addresses=[f"src/mod{i}.py::sym"],
467 operation=_VALID_OPS[i % len(_VALID_OPS)],
468 )
469 elapsed = time.monotonic() - start
470 assert elapsed < 2.0, f"100 intents took {elapsed:.2f}s (limit 2s)"
471
472 def test_load_all_intents_under_0_5_seconds(self, tmp_path: pathlib.Path) -> None:
473 for i in range(100):
474 _make_intent(
475 tmp_path,
476 run_id=f"agent-{i}",
477 addresses=[f"src/mod{i}.py::sym"],
478 )
479 start = time.monotonic()
480 intents = load_all_intents(tmp_path)
481 elapsed = time.monotonic() - start
482 assert len(intents) == 100
483 assert elapsed < 0.5, f"load_all_intents (100 records) took {elapsed:.2f}s (limit 0.5s)"
484
485 def test_500_intents_under_5_seconds(self, tmp_path: pathlib.Path) -> None:
486 start = time.monotonic()
487 for i in range(500):
488 _make_intent(
489 tmp_path,
490 run_id=f"agent-{i}",
491 addresses=[f"src/mod{i}.py::sym"],
492 operation=_VALID_OPS[i % len(_VALID_OPS)],
493 )
494 elapsed = time.monotonic() - start
495 assert elapsed < 5.0, f"500 intents took {elapsed:.2f}s (limit 5s)"
496
497 def test_load_all_intents_500_under_1_second(self, tmp_path: pathlib.Path) -> None:
498 for i in range(500):
499 _make_intent(
500 tmp_path,
501 run_id=f"agent-{i}",
502 addresses=[f"src/mod{i}.py::sym"],
503 )
504 start = time.monotonic()
505 intents = load_all_intents(tmp_path)
506 elapsed = time.monotonic() - start
507 assert len(intents) == 500
508 assert elapsed < 1.0, f"load_all_intents (500 records) took {elapsed:.2f}s (limit 1s)"
509
510 def test_1000_addresses_at_boundary(self, repo: pathlib.Path) -> None:
511 addresses = [f"src/mod.py::sym{i}" for i in range(_MAX_ADDRESSES)]
512 result = runner.invoke(
513 cli,
514 ["coord", "intent"] + addresses + ["--op", "modify", "--json"],
515 )
516 assert result.exit_code == 0
517 data = json.loads(result.output)
518 assert len(data["addresses"]) == _MAX_ADDRESSES
519
520
521 # ---------------------------------------------------------------------------
522 # Input validation
523 # ---------------------------------------------------------------------------
524
525
526 class TestIntentInputValidation:
527 def test_run_id_at_max_length_accepted(self, repo: pathlib.Path) -> None:
528 run_id = "a" * _MAX_RUN_ID_LEN
529 result = runner.invoke(
530 cli,
531 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id, "--json"],
532 )
533 assert result.exit_code == 0
534 data = json.loads(result.output)
535 assert data["run_id"] == run_id
536
537 def test_run_id_over_max_length_exits_user_error(self, repo: pathlib.Path) -> None:
538 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
539 result = runner.invoke(
540 cli,
541 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id],
542 )
543 assert result.exit_code == ExitCode.USER_ERROR
544
545 def test_run_id_over_max_no_file_written(self, repo: pathlib.Path) -> None:
546 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
547 runner.invoke(
548 cli,
549 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id],
550 )
551 intents_dir = repo / ".muse" / "coordination" / "intents"
552 assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == []
553
554 def test_detail_at_max_length_accepted(self, repo: pathlib.Path) -> None:
555 detail = "x" * _MAX_DETAIL_LEN
556 result = runner.invoke(
557 cli,
558 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail, "--json"],
559 )
560 assert result.exit_code == 0
561 data = json.loads(result.output)
562 assert data["detail"] == detail
563
564 def test_detail_over_max_length_exits_user_error(self, repo: pathlib.Path) -> None:
565 detail = "x" * (_MAX_DETAIL_LEN + 1)
566 result = runner.invoke(
567 cli,
568 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail],
569 )
570 assert result.exit_code == ExitCode.USER_ERROR
571
572 def test_detail_over_max_no_file_written(self, repo: pathlib.Path) -> None:
573 detail = "x" * (_MAX_DETAIL_LEN + 1)
574 runner.invoke(
575 cli,
576 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail],
577 )
578 intents_dir = repo / ".muse" / "coordination" / "intents"
579 assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == []
580
581 def test_too_many_addresses_exits_user_error(self, repo: pathlib.Path) -> None:
582 addresses = [f"src/mod.py::sym{i}" for i in range(_MAX_ADDRESSES + 1)]
583 result = runner.invoke(
584 cli,
585 ["coord", "intent"] + addresses + ["--op", "modify"],
586 )
587 assert result.exit_code == ExitCode.USER_ERROR
588
589 def test_too_many_addresses_no_file_written(self, repo: pathlib.Path) -> None:
590 addresses = [f"src/mod.py::sym{i}" for i in range(_MAX_ADDRESSES + 1)]
591 runner.invoke(
592 cli,
593 ["coord", "intent"] + addresses + ["--op", "modify"],
594 )
595 intents_dir = repo / ".muse" / "coordination" / "intents"
596 assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == []
597
598 def test_valid_reservation_id_accepted(self, repo: pathlib.Path) -> None:
599 res_id = str(uuid.uuid4())
600 result = runner.invoke(
601 cli,
602 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
603 "--reservation-id", res_id, "--json"],
604 )
605 assert result.exit_code == 0
606 data = json.loads(result.output)
607 assert data["reservation_id"] == res_id
608
609 def test_invalid_reservation_id_exits_user_error(self, repo: pathlib.Path) -> None:
610 result = runner.invoke(
611 cli,
612 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
613 "--reservation-id", "not-a-uuid"],
614 )
615 assert result.exit_code == ExitCode.USER_ERROR
616
617 def test_invalid_reservation_id_no_file_written(self, repo: pathlib.Path) -> None:
618 runner.invoke(
619 cli,
620 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
621 "--reservation-id", "not-a-uuid"],
622 )
623 intents_dir = repo / ".muse" / "coordination" / "intents"
624 assert not intents_dir.exists() or list(intents_dir.glob("*.json")) == []
625
626 def test_empty_reservation_id_creates_standalone_intent(self, repo: pathlib.Path) -> None:
627 result = runner.invoke(
628 cli,
629 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"],
630 )
631 assert result.exit_code == 0
632 # no --reservation-id means standalone; field present but empty or omitted
633 data = json.loads(result.output)
634 assert "reservation_id" in data
635
636 def test_validation_fires_before_io_on_bad_run_id(self, repo: pathlib.Path) -> None:
637 """Validates that bad run-id rejects without touching the intents directory."""
638 run_id = "z" * (_MAX_RUN_ID_LEN + 1)
639 intents_dir = repo / ".muse" / "coordination" / "intents"
640 result = runner.invoke(
641 cli,
642 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id],
643 )
644 assert result.exit_code == ExitCode.USER_ERROR
645 # The intents directory must not have been created.
646 assert not intents_dir.exists()
647
648 def test_run_id_over_max_json_mode_returns_error_field(self, repo: pathlib.Path) -> None:
649 run_id = "a" * (_MAX_RUN_ID_LEN + 1)
650 result = runner.invoke(
651 cli,
652 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", run_id, "--json"],
653 )
654 assert result.exit_code == ExitCode.USER_ERROR
655 data = json.loads(result.output)
656 assert "error" in data
657
658 def test_invalid_reservation_id_json_mode_returns_error_field(self, repo: pathlib.Path) -> None:
659 result = runner.invoke(
660 cli,
661 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
662 "--reservation-id", "bad-id", "--json"],
663 )
664 assert result.exit_code == ExitCode.USER_ERROR
665 data = json.loads(result.output)
666 assert "error" in data
667 assert data.get("status") == "bad_reservation_id"
668
669
670 # ---------------------------------------------------------------------------
671 # JSON format — compact output
672 # ---------------------------------------------------------------------------
673
674
675 class TestIntentJsonFormat:
676 def test_json_output_is_compact(self, repo: pathlib.Path) -> None:
677 """No pretty-printing (no indent=2): body must be a single line."""
678 result = runner.invoke(
679 cli,
680 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"],
681 )
682 assert result.exit_code == 0
683 body = result.output.strip()
684 assert "\n" not in body, "JSON output must be compact (no newlines)"
685
686 def test_json_schema_version_present(self, repo: pathlib.Path) -> None:
687 result = runner.invoke(
688 cli,
689 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"],
690 )
691 data = json.loads(result.output)
692 assert "schema_version" in data
693 assert isinstance(data["schema_version"], str)
694
695 def test_json_branch_field_present(self, repo: pathlib.Path) -> None:
696 result = runner.invoke(
697 cli,
698 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"],
699 )
700 data = json.loads(result.output)
701 assert "branch" in data
702 assert isinstance(data["branch"], str) and data["branch"]
703
704 def test_json_created_at_is_iso_string(self, repo: pathlib.Path) -> None:
705 result = runner.invoke(
706 cli,
707 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"],
708 )
709 data = json.loads(result.output)
710 assert "T" in data["created_at"]
711
712 def test_json_two_invocations_produce_unique_intent_ids(self, repo: pathlib.Path) -> None:
713 r1 = runner.invoke(cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"])
714 r2 = runner.invoke(cli, ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--json"])
715 id1 = json.loads(r1.output)["intent_id"]
716 id2 = json.loads(r2.output)["intent_id"]
717 assert id1 != id2
718
719 def test_json_error_on_bad_detail_is_compact(self, repo: pathlib.Path) -> None:
720 detail = "x" * (_MAX_DETAIL_LEN + 1)
721 result = runner.invoke(
722 cli,
723 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--detail", detail, "--json"],
724 )
725 body = result.output.strip()
726 assert "\n" not in body
727 data = json.loads(body)
728 assert "error" in data
729
730
731 # ---------------------------------------------------------------------------
732 # Sanitize display — ANSI/control chars stripped in text output
733 # ---------------------------------------------------------------------------
734
735
736 class TestIntentSanitize:
737 def test_ansi_in_run_id_stripped_from_text_output(self, repo: pathlib.Path) -> None:
738 ansi_id = "\x1b[31mevil\x1b[0m"
739 result = runner.invoke(
740 cli,
741 ["coord", "intent", "src/mod.py::sym", "--op", "modify", "--run-id", ansi_id],
742 )
743 assert result.exit_code == 0
744 assert "\x1b[31m" not in result.output
745 assert "\x1b[0m" not in result.output
746
747 def test_ansi_in_detail_stripped_from_text_output(self, repo: pathlib.Path) -> None:
748 ansi_detail = "\x1b[32mgreen detail\x1b[0m"
749 result = runner.invoke(
750 cli,
751 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
752 "--detail", ansi_detail, "--run-id", "agent-1"],
753 )
754 assert result.exit_code == 0
755 assert "\x1b[32m" not in result.output
756 assert "\x1b[0m" not in result.output
757
758 def test_ansi_in_reservation_id_stripped_from_text_output(self, repo: pathlib.Path) -> None:
759 # Valid UUID but then test that sanitize_display removes ANSI from display.
760 # We can't put ANSI in a UUID (it would fail UUID validation), so we
761 # verify the UUID is echoed cleanly (no raw escape sequences).
762 res_id = str(uuid.uuid4())
763 result = runner.invoke(
764 cli,
765 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
766 "--reservation-id", res_id, "--run-id", "agent-1"],
767 )
768 assert result.exit_code == 0
769 assert "\x1b[" not in result.output
770
771 def test_bel_in_detail_does_not_appear_in_text_output(self, repo: pathlib.Path) -> None:
772 detail = "ok\x07bad"
773 result = runner.invoke(
774 cli,
775 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
776 "--detail", detail, "--run-id", "agent-1"],
777 )
778 assert result.exit_code == 0
779 assert "\x07" not in result.output
780
781
782 # ---------------------------------------------------------------------------
783 # filter_intents — unit tests
784 # ---------------------------------------------------------------------------
785
786
787 class TestFilterIntents:
788 def _make_n(self, root: pathlib.Path, n: int) -> list[Intent]:
789 ops = _VALID_OPS
790 return [
791 _make_intent(
792 root,
793 run_id=f"agent-{i % 3}",
794 addresses=[f"src/mod{i % 5}.py::sym", "shared.py::util"],
795 operation=ops[i % len(ops)],
796 detail=f"detail-{i}",
797 )
798 for i in range(n)
799 ]
800
801 def test_filter_by_run_id_returns_subset(self, tmp_path: pathlib.Path) -> None:
802 self._make_n(tmp_path, 9)
803 all_intents = load_all_intents(tmp_path)
804 filtered = filter_intents(all_intents, run_id="agent-0")
805 assert all(i.run_id == "agent-0" for i in filtered)
806 assert len(filtered) > 0
807
808 def test_filter_by_run_id_excludes_others(self, tmp_path: pathlib.Path) -> None:
809 self._make_n(tmp_path, 9)
810 all_intents = load_all_intents(tmp_path)
811 filtered = filter_intents(all_intents, run_id="agent-1")
812 assert all(i.run_id == "agent-1" for i in filtered)
813 assert not any(i.run_id == "agent-0" for i in filtered)
814
815 def test_filter_by_operation_returns_subset(self, tmp_path: pathlib.Path) -> None:
816 self._make_n(tmp_path, 16)
817 all_intents = load_all_intents(tmp_path)
818 filtered = filter_intents(all_intents, operation="rename")
819 assert all(i.operation == "rename" for i in filtered)
820 assert len(filtered) > 0
821
822 def test_filter_by_address_glob_returns_matches(self, tmp_path: pathlib.Path) -> None:
823 self._make_n(tmp_path, 10)
824 all_intents = load_all_intents(tmp_path)
825 filtered = filter_intents(all_intents, address_glob="src/mod0.py::*")
826 assert all(any("src/mod0.py" in a for a in i.addresses) for i in filtered)
827 assert len(filtered) > 0
828
829 def test_filter_combined_and_semantics(self, tmp_path: pathlib.Path) -> None:
830 self._make_n(tmp_path, 24)
831 all_intents = load_all_intents(tmp_path)
832 filtered = filter_intents(all_intents, run_id="agent-0", operation="rename")
833 assert all(i.run_id == "agent-0" and i.operation == "rename" for i in filtered)
834
835 def test_filter_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None:
836 self._make_n(tmp_path, 5)
837 all_intents = load_all_intents(tmp_path)
838 filtered = filter_intents(all_intents, run_id="nonexistent-agent")
839 assert filtered == []
840
841 def test_filter_empty_list_returns_empty(self, tmp_path: pathlib.Path) -> None:
842 filtered = filter_intents([], run_id="agent-0")
843 assert filtered == []
844
845 def test_filter_no_filters_returns_all(self, tmp_path: pathlib.Path) -> None:
846 self._make_n(tmp_path, 6)
847 all_intents = load_all_intents(tmp_path)
848 filtered = filter_intents(all_intents)
849 assert len(filtered) == len(all_intents)
850
851
852 # ---------------------------------------------------------------------------
853 # Concurrent writes
854 # ---------------------------------------------------------------------------
855
856
857 class TestIntentConcurrent:
858 def test_20_threads_all_exit_zero(self, repo: pathlib.Path) -> None:
859 results: list[int] = []
860 lock = threading.Lock()
861
862 def _write() -> None:
863 result = runner.invoke(
864 cli,
865 ["coord", "intent", "src/mod.py::sym", "--op", "modify",
866 "--run-id", f"agent-{threading.get_ident()}", "--json"],
867 )
868 with lock:
869 results.append(result.exit_code)
870
871 threads = [threading.Thread(target=_write) for _ in range(20)]
872 for t in threads:
873 t.start()
874 for t in threads:
875 t.join()
876
877 assert all(c == 0 for c in results), f"some threads failed: {results}"
878
879 def test_20_threads_produce_unique_intent_ids(self, tmp_path: pathlib.Path) -> None:
880 """Uses create_intent directly to avoid CliRunner stdout-capture collision."""
881 intent_ids: list[str] = []
882 lock = threading.Lock()
883
884 def _write() -> None:
885 it = create_intent(
886 tmp_path,
887 reservation_id="",
888 run_id="agent-concurrent",
889 branch="main",
890 addresses=["src/mod.py::sym"],
891 operation="modify",
892 detail="",
893 )
894 with lock:
895 intent_ids.append(it.intent_id)
896
897 threads = [threading.Thread(target=_write) for _ in range(20)]
898 for t in threads:
899 t.start()
900 for t in threads:
901 t.join()
902
903 assert len(intent_ids) == 20
904 assert len(set(intent_ids)) == 20, "intent IDs must be globally unique"
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