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