gabriel / muse public
test_harmony_phase1.py python
1,338 lines 53.1 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 135 days ago
1 """Tests for muse/core/harmony.py — Phase 1: Core data model.
2
3 Coverage tiers
4 --------------
5 I Unit — fingerprints, validation, namespaces, dataclasses, condition matching
6 II Integration — all CRUD operations (patterns, resolutions, policies, audit, gc)
7 III End-to-end — full conflict lifecycle from record → resolve → replay → gc
8 IV Stress — 10 k pattern scan, concurrent writes under parallel threads
9 V Data integrity— atomic writes (no temp files left), JSON round-trip, field types
10 VI Security — path traversal, symlink guards, size caps, crafted IDs
11 VII Performance — per-operation timing assertions
12 """
13 from __future__ import annotations
14 from collections.abc import Mapping
15
16 import concurrent.futures
17 import datetime
18 from muse.core._types import fake_id
19 import json
20 import os
21 import pathlib
22 import tempfile
23 import threading
24 import time
25 from dataclasses import FrozenInstanceError
26 from typing import Any
27 from unittest import mock
28
29 import pytest
30
31 import muse.core.harmony as h
32 from muse.core.harmony import (
33 AgentProvenance,
34 AuditEvent,
35 AuditEventType,
36 ConflictPattern,
37 ConflictType,
38 Policy,
39 PolicyAction,
40 PolicyCondition,
41 PolicyScope,
42 Resolution,
43 ResolutionProposal,
44 ResolutionStrategy,
45 _MAX_AUDIT_BYTES,
46 _MAX_PATTERN_BYTES,
47 _MAX_POLICY_BYTES,
48 _MAX_RESOLUTION_BYTES,
49 _MAX_SCAN,
50 _condition_matches,
51 append_audit,
52 best_resolution,
53 blob_fingerprint,
54 clear_all,
55 compute_pattern_id,
56 compute_resolution_id,
57 forget_pattern,
58 gc_stale,
59 increment_applied_count,
60 list_audit,
61 list_patterns,
62 list_policies,
63 list_resolutions,
64 load_pattern,
65 load_policy,
66 load_resolution,
67 match_policy,
68 record_pattern,
69 remove_policy,
70 save_policy,
71 save_resolution,
72 )
73
74
75 # ---------------------------------------------------------------------------
76 # Shared fixtures
77 # ---------------------------------------------------------------------------
78
79
80 @pytest.fixture()
81 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
82 """Return a temporary directory acting as a bare repo root."""
83 (tmp_path / ".muse").mkdir()
84 return tmp_path
85
86
87
88 def _utc_now() -> datetime.datetime:
89 return datetime.datetime.now(datetime.timezone.utc)
90
91
92 def _make_pattern(
93 path: str = "track.mid",
94 domain: str = "midi",
95 conflict_type: str = ConflictType.CONTENT,
96 ours: str = "ours",
97 theirs: str = "theirs",
98 description: Mapping[str, object] | None = None,
99 recorded_by: str = "claude-code",
100 ) -> ConflictPattern:
101 """Build a ConflictPattern with sensible defaults."""
102 ours_id = fake_id(ours)
103 theirs_id = fake_id(theirs)
104 blob_fp = blob_fingerprint(ours_id, theirs_id)
105 semantic_fp = blob_fp
106 pattern_id = compute_pattern_id(path, blob_fp, semantic_fp)
107 return ConflictPattern(
108 pattern_id=pattern_id,
109 path=path,
110 domain=domain,
111 conflict_type=conflict_type,
112 blob_fingerprint=blob_fp,
113 semantic_fingerprint=semantic_fp,
114 ours_id=ours_id,
115 theirs_id=theirs_id,
116 description=description or {},
117 recorded_at=_utc_now(),
118 recorded_by=recorded_by,
119 )
120
121
122 def _make_resolution(
123 pattern: ConflictPattern,
124 strategy: str = ResolutionStrategy.MANUAL,
125 confidence: float = 0.9,
126 human_verified: bool = False,
127 provenance: AgentProvenance | None = None,
128 policy_id: str | None = None,
129 ) -> Resolution:
130 """Build a Resolution tied to *pattern* with sensible defaults."""
131 outcome_blob = fake_id(f"outcome-{pattern.pattern_id[:8]}")
132 prov = provenance or AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
133 resolved_at = _utc_now()
134 resolution_id = compute_resolution_id(
135 pattern.pattern_id, outcome_blob, strategy, prov, resolved_at
136 )
137 return Resolution(
138 resolution_id=resolution_id,
139 pattern_id=pattern.pattern_id,
140 strategy=strategy,
141 policy_id=policy_id,
142 outcome_blob=outcome_blob,
143 resolved_by=prov,
144 human_verified=human_verified,
145 confidence=confidence,
146 rationale="Test rationale",
147 resolved_at=resolved_at,
148 )
149
150
151 def _make_policy(
152 policy_id: str = "always-prefer-ours",
153 scope: str = PolicyScope.REPO,
154 action: str = PolicyAction.PREFER_OURS,
155 confidence: float = 0.95,
156 conflict_type: str | None = None,
157 domain: str | None = None,
158 path_pattern: str | None = None,
159 ) -> Policy:
160 """Build a Policy with sensible defaults."""
161 return Policy(
162 policy_id=policy_id,
163 description="Test policy",
164 when=PolicyCondition(
165 conflict_type=conflict_type,
166 domain=domain,
167 path_pattern=path_pattern,
168 ),
169 action=action,
170 confidence=confidence,
171 escalate_to=None,
172 delegate_to=None,
173 scope=scope,
174 created_at=_utc_now(),
175 created_by="claude-code",
176 )
177
178
179 # ===========================================================================
180 # Tier I — Unit tests
181 # ===========================================================================
182
183
184 class TestBlobFingerprint:
185 """I: blob_fingerprint must be commutative and deterministic."""
186
187 def test_commutativity(self) -> None:
188 a, b = fake_id("A"), fake_id("B")
189 assert blob_fingerprint(a, b) == blob_fingerprint(b, a)
190
191 def test_determinism(self) -> None:
192 a, b = fake_id("X"), fake_id("Y")
193 fp1 = blob_fingerprint(a, b)
194 fp2 = blob_fingerprint(a, b)
195 assert fp1 == fp2
196
197 def test_output_is_64_hex(self) -> None:
198 a, b = fake_id("p"), fake_id("q")
199 fp = blob_fingerprint(a, b)
200 assert fp.startswith("sha256:")
201 assert len(fp) == 71
202
203 def test_distinct_pairs_differ(self) -> None:
204 ab = blob_fingerprint(fake_id("A"), fake_id("B"))
205 cd = blob_fingerprint(fake_id("C"), fake_id("D"))
206 assert ab != cd
207
208 def test_same_id_both_sides(self) -> None:
209 a = fake_id("same")
210 # Should not crash; result is deterministic
211 fp = blob_fingerprint(a, a)
212 assert fp.startswith("sha256:")
213 assert len(fp) == 71
214
215
216 class TestComputePatternId:
217 """I: compute_pattern_id includes path, so same content → different IDs for different paths."""
218
219 def test_deterministic(self) -> None:
220 blob_fp = fake_id("blob")
221 sem_fp = fake_id("sem")
222 p1 = compute_pattern_id("track.mid", blob_fp, sem_fp)
223 p2 = compute_pattern_id("track.mid", blob_fp, sem_fp)
224 assert p1 == p2
225
226 def test_path_changes_id(self) -> None:
227 blob_fp = fake_id("blob")
228 sem_fp = fake_id("sem")
229 p1 = compute_pattern_id("track.mid", blob_fp, sem_fp)
230 p2 = compute_pattern_id("drums.mid", blob_fp, sem_fp)
231 assert p1 != p2
232
233 def test_blob_changes_id(self) -> None:
234 # When blob_fp == semantic_fp (no plugin), blob content drives the pattern ID.
235 fpA = fake_id("blobA")
236 fpB = fake_id("blobB")
237 p1 = compute_pattern_id("track.mid", fpA, fpA)
238 p2 = compute_pattern_id("track.mid", fpB, fpB)
239 assert p1 != p2
240
241 def test_64_hex_output(self) -> None:
242 pid = compute_pattern_id("f.py", fake_id("b"), fake_id("s"))
243 assert pid.startswith("sha256:")
244 assert len(pid) == 71
245
246
247 class TestComputeResolutionId:
248 """I: compute_resolution_id is deterministic and encodes actor."""
249
250 def test_deterministic(self) -> None:
251 prov = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
252 ts = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
253 pid = fake_id("pattern")
254 ob = fake_id("outcome")
255 r1 = compute_resolution_id(pid, ob, ResolutionStrategy.MANUAL, prov, ts)
256 r2 = compute_resolution_id(pid, ob, ResolutionStrategy.MANUAL, prov, ts)
257 assert r1 == r2
258
259 def test_different_agents_differ(self) -> None:
260 ts = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
261 pid = fake_id("pattern")
262 ob = fake_id("outcome")
263 p1 = AgentProvenance.agent("claude-code")
264 p2 = AgentProvenance.agent("codex")
265 r1 = compute_resolution_id(pid, ob, ResolutionStrategy.MANUAL, p1, ts)
266 r2 = compute_resolution_id(pid, ob, ResolutionStrategy.MANUAL, p2, ts)
267 assert r1 != r2
268
269 def test_human_provenance_encodes_as_human(self) -> None:
270 ts = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
271 pid = fake_id("pattern")
272 ob = fake_id("outcome")
273 rid = compute_resolution_id(pid, ob, ResolutionStrategy.MANUAL, AgentProvenance.human(), ts)
274 assert rid.startswith("sha256:")
275 assert len(rid) == 71
276
277
278 class TestValidation:
279 """I: _validate_id and _validate_policy_id must reject bad inputs."""
280
281 def test_validate_id_accepts_64_hex(self) -> None:
282 h._validate_id(fake_id("a")) # no exception
283
284 def test_validate_id_rejects_63_chars(self) -> None:
285 with pytest.raises(ValueError):
286 h._validate_id("a" * 63)
287
288 def test_validate_id_rejects_65_chars(self) -> None:
289 with pytest.raises(ValueError):
290 h._validate_id("a" * 65)
291
292 def test_validate_id_rejects_uppercase(self) -> None:
293 with pytest.raises(ValueError):
294 h._validate_id("A" * 64)
295
296 def test_validate_id_rejects_path_traversal(self) -> None:
297 # Attempt to inject a path traversal via the ID
298 with pytest.raises(ValueError):
299 h._validate_id("../../../etc/passwd" + "a" * 45)
300
301 def test_validate_id_rejects_empty(self) -> None:
302 with pytest.raises(ValueError):
303 h._validate_id("")
304
305 def test_validate_policy_id_accepts_alphanumeric(self) -> None:
306 h._validate_policy_id("my-policy_123") # no exception
307
308 def test_validate_policy_id_rejects_slash(self) -> None:
309 with pytest.raises(ValueError, match="alphanumeric"):
310 h._validate_policy_id("bad/policy")
311
312 def test_validate_policy_id_rejects_dot(self) -> None:
313 with pytest.raises(ValueError):
314 h._validate_policy_id("bad.policy")
315
316 def test_validate_policy_id_rejects_empty(self) -> None:
317 with pytest.raises(ValueError):
318 h._validate_policy_id("")
319
320 def test_validate_policy_id_rejects_129_chars(self) -> None:
321 with pytest.raises(ValueError):
322 h._validate_policy_id("a" * 129)
323
324 def test_validate_policy_id_accepts_128_chars(self) -> None:
325 h._validate_policy_id("a" * 128) # no exception
326
327
328 class TestNamespaces:
329 """I: Open string-constant namespaces are plain strings — plugin extensibility."""
330
331 def test_conflict_type_are_strings(self) -> None:
332 assert isinstance(ConflictType.CONTENT, str)
333 assert isinstance(ConflictType.STRUCTURAL, str)
334 assert isinstance(ConflictType.METADATA, str)
335 assert isinstance(ConflictType.RELATIONAL, str)
336 assert isinstance(ConflictType.UNKNOWN, str)
337
338 def test_resolution_strategy_are_strings(self) -> None:
339 assert isinstance(ResolutionStrategy.POLICY, str)
340 assert isinstance(ResolutionStrategy.EXACT_REPLAY, str)
341 assert isinstance(ResolutionStrategy.SEMANTIC_PROPOSAL, str)
342 assert isinstance(ResolutionStrategy.MANUAL, str)
343
344 def test_policy_action_are_strings(self) -> None:
345 for attr in ("PREFER_OURS", "PREFER_THEIRS", "ESCALATE", "REQUIRE_HUMAN", "DELEGATE"):
346 assert isinstance(getattr(PolicyAction, attr), str)
347
348 def test_policy_scope_are_strings(self) -> None:
349 for attr in ("WORKSPACE", "REPO", "DOMAIN", "FILE"):
350 assert isinstance(getattr(PolicyScope, attr), str)
351
352 def test_audit_event_type_are_strings(self) -> None:
353 for attr in (
354 "PATTERN_RECORDED", "RESOLUTION_SAVED", "RESOLUTION_APPLIED",
355 "PATTERN_FORGOTTEN", "POLICY_SAVED", "POLICY_REMOVED", "GC_RUN", "CLEAR_RUN",
356 ):
357 assert isinstance(getattr(AuditEventType, attr), str)
358
359 def test_plugin_can_use_custom_conflict_type(self) -> None:
360 # Plugins may add strings at runtime — no import needed
361 custom_type = "note_collision"
362 pattern = _make_pattern(conflict_type=custom_type)
363 assert pattern.conflict_type == custom_type
364
365
366 class TestDataclasses:
367 """I: Frozen dataclasses, AgentProvenance constructors, PolicyCondition wildcards."""
368
369 def test_agent_provenance_human(self) -> None:
370 p = AgentProvenance.human()
371 assert p.type == "human"
372 assert p.agent_id is None
373 assert p.model_id is None
374
375 def test_agent_provenance_agent(self) -> None:
376 p = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
377 assert p.type == "agent"
378 assert p.agent_id == "claude-code"
379 assert p.model_id == "claude-sonnet-4-6"
380
381 def test_agent_provenance_agent_no_model(self) -> None:
382 p = AgentProvenance.agent("codex")
383 assert p.model_id is None
384
385 def test_agent_provenance_frozen(self) -> None:
386 p = AgentProvenance.human()
387 with pytest.raises(FrozenInstanceError):
388 p.type = "agent" # type: ignore[misc]
389
390 def test_policy_condition_all_none_wildcard(self) -> None:
391 cond = PolicyCondition()
392 assert cond.conflict_type is None
393 assert cond.domain is None
394 assert cond.path_pattern is None
395 assert cond.min_confidence is None
396
397 def test_policy_condition_frozen(self) -> None:
398 cond = PolicyCondition(conflict_type=ConflictType.CONTENT)
399 with pytest.raises(FrozenInstanceError):
400 cond.conflict_type = ConflictType.STRUCTURAL # type: ignore[misc]
401
402 def test_conflict_pattern_frozen(self) -> None:
403 p = _make_pattern()
404 with pytest.raises(FrozenInstanceError):
405 p.path = "evil.mid" # type: ignore[misc]
406
407 def test_resolution_frozen(self) -> None:
408 pattern = _make_pattern()
409 res = _make_resolution(pattern)
410 with pytest.raises(FrozenInstanceError):
411 res.confidence = 0.0 # type: ignore[misc]
412
413 def test_resolution_proposal_defaults(self) -> None:
414 prop = ResolutionProposal(
415 pattern_id=fake_id("p"),
416 strategy=ResolutionStrategy.SEMANTIC_PROPOSAL,
417 proposed_action=PolicyAction.PREFER_OURS,
418 confidence=0.7,
419 rationale="fuzzy match",
420 )
421 assert prop.policy_id is None
422 assert prop.similar_pattern_id is None
423 assert prop.similarity is None
424 assert prop.requires_confirmation is False
425
426
427 class TestConditionMatching:
428 """I: _condition_matches and match_policy first-match-wins semantics."""
429
430 def _pattern(self, **kwargs) -> ConflictPattern:
431 return _make_pattern(**kwargs)
432
433 def test_all_none_matches_anything(self) -> None:
434 cond = PolicyCondition()
435 assert _condition_matches(cond, self._pattern()) is True
436
437 def test_conflict_type_match(self) -> None:
438 cond = PolicyCondition(conflict_type=ConflictType.CONTENT)
439 assert _condition_matches(cond, self._pattern(conflict_type=ConflictType.CONTENT)) is True
440 assert _condition_matches(cond, self._pattern(conflict_type=ConflictType.METADATA)) is False
441
442 def test_domain_match(self) -> None:
443 cond = PolicyCondition(domain="midi")
444 assert _condition_matches(cond, self._pattern(domain="midi")) is True
445 assert _condition_matches(cond, self._pattern(domain="code")) is False
446
447 def test_path_pattern_glob(self) -> None:
448 cond = PolicyCondition(path_pattern="*.mid")
449 assert _condition_matches(cond, self._pattern(path="track.mid")) is True
450 assert _condition_matches(cond, self._pattern(path="src/main.py")) is False
451
452 def test_path_pattern_prefix_glob(self) -> None:
453 cond = PolicyCondition(path_pattern="audio/*")
454 assert _condition_matches(cond, self._pattern(path="audio/kick.mid")) is True
455 assert _condition_matches(cond, self._pattern(path="video/clip.mp4")) is False
456
457 def test_all_conditions_must_match(self) -> None:
458 cond = PolicyCondition(conflict_type=ConflictType.CONTENT, domain="midi")
459 matching = self._pattern(conflict_type=ConflictType.CONTENT, domain="midi")
460 wrong_domain = self._pattern(conflict_type=ConflictType.CONTENT, domain="code")
461 wrong_type = self._pattern(conflict_type=ConflictType.METADATA, domain="midi")
462 assert _condition_matches(cond, matching) is True
463 assert _condition_matches(cond, wrong_domain) is False
464 assert _condition_matches(cond, wrong_type) is False
465
466 def test_match_policy_first_match_wins(self) -> None:
467 policy_a = _make_policy("policy-a", scope=PolicyScope.WORKSPACE, conflict_type=ConflictType.CONTENT)
468 policy_b = _make_policy("policy-b", scope=PolicyScope.REPO, conflict_type=ConflictType.CONTENT)
469 pattern = self._pattern(conflict_type=ConflictType.CONTENT)
470 result = match_policy([policy_a, policy_b], pattern)
471 assert result is not None
472 assert result.policy_id == "policy-a"
473
474 def test_match_policy_no_match_returns_none(self) -> None:
475 policy = _make_policy("p", conflict_type=ConflictType.STRUCTURAL)
476 pattern = self._pattern(conflict_type=ConflictType.CONTENT)
477 assert match_policy([policy], pattern) is None
478
479 def test_match_policy_empty_list(self) -> None:
480 pattern = self._pattern()
481 assert match_policy([], pattern) is None
482
483 def test_min_confidence_not_evaluated_here(self) -> None:
484 # min_confidence is an engine-level filter, not evaluated by _condition_matches
485 cond = PolicyCondition(min_confidence=0.99)
486 # Should still match since no other fields constrain the pattern
487 assert _condition_matches(cond, self._pattern()) is True
488
489
490 # ===========================================================================
491 # Tier II — Integration tests
492 # ===========================================================================
493
494
495 class TestPatternCRUD:
496 """II: record_pattern, load_pattern, list_patterns, forget_pattern, clear_all."""
497
498 def test_record_and_load(self, repo: pathlib.Path) -> None:
499 pattern = _make_pattern()
500 record_pattern(repo, pattern)
501 loaded = load_pattern(repo, pattern.pattern_id)
502 assert loaded is not None
503 assert loaded.pattern_id == pattern.pattern_id
504 assert loaded.path == pattern.path
505 assert loaded.domain == pattern.domain
506 assert loaded.conflict_type == pattern.conflict_type
507
508 def test_record_is_idempotent(self, repo: pathlib.Path) -> None:
509 pattern = _make_pattern()
510 record_pattern(repo, pattern)
511 # Second call should not raise and should return same ID
512 pid = record_pattern(repo, pattern)
513 assert pid == pattern.pattern_id
514 # Only one pattern.json should exist
515 entry_dir = h.pattern_dir(repo, pattern.pattern_id)
516 assert list(entry_dir.glob("pattern.json")) == [entry_dir / "pattern.json"]
517
518 def test_load_nonexistent_returns_none(self, repo: pathlib.Path) -> None:
519 assert load_pattern(repo, "a" * 64) is None
520
521 def test_load_invalid_id_returns_none(self, repo: pathlib.Path) -> None:
522 assert load_pattern(repo, "not-a-hex-id") is None
523
524 def test_list_patterns_empty(self, repo: pathlib.Path) -> None:
525 assert list_patterns(repo) == []
526
527 def test_list_patterns_multiple(self, repo: pathlib.Path) -> None:
528 p1 = _make_pattern(path="a.mid", ours="oa", theirs="ta")
529 p2 = _make_pattern(path="b.mid", ours="ob", theirs="tb")
530 record_pattern(repo, p1)
531 record_pattern(repo, p2)
532 results = list_patterns(repo)
533 assert len(results) == 2
534 pids = {r.pattern_id for r in results}
535 assert {p1.pattern_id, p2.pattern_id} == pids
536
537 def test_list_patterns_sorted_newest_first(self, repo: pathlib.Path) -> None:
538 older = _make_pattern(path="old.mid", ours="oa", theirs="ta")
539 newer = _make_pattern(path="new.mid", ours="ob", theirs="tb")
540 # Force newer to be newer by manipulating recorded_at
541 import dataclasses
542 ts_old = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
543 ts_new = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
544 older = dataclasses.replace(older, recorded_at=ts_old)
545 newer = dataclasses.replace(newer, recorded_at=ts_new)
546 record_pattern(repo, older)
547 record_pattern(repo, newer)
548 results = list_patterns(repo)
549 assert results[0].pattern_id == newer.pattern_id
550
551 def test_forget_pattern_removes_entry(self, repo: pathlib.Path) -> None:
552 pattern = _make_pattern()
553 record_pattern(repo, pattern)
554 assert forget_pattern(repo, pattern.pattern_id) is True
555 assert load_pattern(repo, pattern.pattern_id) is None
556
557 def test_forget_nonexistent_returns_false(self, repo: pathlib.Path) -> None:
558 assert forget_pattern(repo, "a" * 64) is False
559
560 def test_forget_invalid_id_returns_false(self, repo: pathlib.Path) -> None:
561 assert forget_pattern(repo, "../evil") is False
562
563 def test_forget_also_removes_resolutions(self, repo: pathlib.Path) -> None:
564 pattern = _make_pattern()
565 record_pattern(repo, pattern)
566 res = _make_resolution(pattern)
567 save_resolution(repo, res)
568 forget_pattern(repo, pattern.pattern_id)
569 # Resolution directory should be gone
570 res_dir = h._resolutions_dir(repo, pattern.pattern_id)
571 assert not res_dir.exists()
572
573 def test_clear_all_removes_all(self, repo: pathlib.Path) -> None:
574 for i in range(5):
575 record_pattern(repo, _make_pattern(path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}"))
576 removed = clear_all(repo)
577 assert removed == 5
578 assert list_patterns(repo) == []
579
580 def test_clear_all_empty_store(self, repo: pathlib.Path) -> None:
581 assert clear_all(repo) == 0
582
583 def test_record_pattern_invalid_id_raises(self, repo: pathlib.Path) -> None:
584 import dataclasses
585 pattern = _make_pattern()
586 bad = dataclasses.replace(pattern, pattern_id="bad-id")
587 with pytest.raises(ValueError):
588 record_pattern(repo, bad)
589
590
591 class TestResolutionCRUD:
592 """II: save_resolution, load_resolution, list_resolutions, increment_applied_count, best_resolution."""
593
594 def test_save_and_load(self, repo: pathlib.Path) -> None:
595 pattern = _make_pattern()
596 record_pattern(repo, pattern)
597 res = _make_resolution(pattern)
598 save_resolution(repo, res)
599 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
600 assert loaded is not None
601 assert loaded.resolution_id == res.resolution_id
602 assert loaded.pattern_id == pattern.pattern_id
603 assert loaded.strategy == res.strategy
604 assert loaded.confidence == res.confidence
605
606 def test_save_is_idempotent(self, repo: pathlib.Path) -> None:
607 pattern = _make_pattern()
608 record_pattern(repo, pattern)
609 res = _make_resolution(pattern)
610 save_resolution(repo, res)
611 save_resolution(repo, res) # second call is no-op
612 assert len(list_resolutions(repo, pattern.pattern_id)) == 1
613
614 def test_save_requires_parent_pattern(self, repo: pathlib.Path) -> None:
615 pattern = _make_pattern()
616 res = _make_resolution(pattern)
617 with pytest.raises(FileNotFoundError, match="No harmony pattern"):
618 save_resolution(repo, res)
619
620 def test_load_nonexistent_returns_none(self, repo: pathlib.Path) -> None:
621 assert load_resolution(repo, "a" * 64, "b" * 64) is None
622
623 def test_load_invalid_ids_return_none(self, repo: pathlib.Path) -> None:
624 assert load_resolution(repo, "bad", "b" * 64) is None
625 assert load_resolution(repo, "a" * 64, "bad") is None
626
627 def test_list_resolutions_empty(self, repo: pathlib.Path) -> None:
628 pattern = _make_pattern()
629 record_pattern(repo, pattern)
630 assert list_resolutions(repo, pattern.pattern_id) == []
631
632 def test_list_resolutions_sorted_by_quality(self, repo: pathlib.Path) -> None:
633 """human_verified > confidence > applied_count (desc)."""
634 pattern = _make_pattern()
635 record_pattern(repo, pattern)
636
637 low_conf = _make_resolution(pattern, confidence=0.3)
638 high_conf = _make_resolution(pattern, confidence=0.9)
639 verified = _make_resolution(pattern, confidence=0.5, human_verified=True)
640
641 # Build distinct resolutions (different outcomes)
642 import dataclasses
643 low_conf = dataclasses.replace(
644 low_conf,
645 outcome_blob=fake_id("low_outcome"),
646 resolution_id=fake_id("low_res"),
647 )
648 high_conf = dataclasses.replace(
649 high_conf,
650 outcome_blob=fake_id("high_outcome"),
651 resolution_id=fake_id("high_res"),
652 )
653 verified = dataclasses.replace(
654 verified,
655 outcome_blob=fake_id("ver_outcome"),
656 resolution_id=fake_id("ver_res"),
657 )
658
659 for r in (low_conf, high_conf, verified):
660 save_resolution(repo, r)
661
662 results = list_resolutions(repo, pattern.pattern_id)
663 assert results[0].resolution_id == verified.resolution_id # human_verified first
664 assert results[-1].resolution_id == low_conf.resolution_id # lowest confidence last
665
666 def test_increment_applied_count(self, repo: pathlib.Path) -> None:
667 pattern = _make_pattern()
668 record_pattern(repo, pattern)
669 res = _make_resolution(pattern)
670 save_resolution(repo, res)
671 assert increment_applied_count(repo, pattern.pattern_id, res.resolution_id) is True
672 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
673 assert loaded is not None
674 assert loaded.applied_count == 1
675
676 def test_increment_applied_count_multiple_times(self, repo: pathlib.Path) -> None:
677 pattern = _make_pattern()
678 record_pattern(repo, pattern)
679 res = _make_resolution(pattern)
680 save_resolution(repo, res)
681 for _ in range(5):
682 increment_applied_count(repo, pattern.pattern_id, res.resolution_id)
683 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
684 assert loaded is not None
685 assert loaded.applied_count == 5
686
687 def test_increment_nonexistent_returns_false(self, repo: pathlib.Path) -> None:
688 assert increment_applied_count(repo, "a" * 64, "b" * 64) is False
689
690 def test_best_resolution_returns_highest_quality(self, repo: pathlib.Path) -> None:
691 pattern = _make_pattern()
692 record_pattern(repo, pattern)
693
694 import dataclasses
695 r1 = _make_resolution(pattern, confidence=0.5)
696 r2 = _make_resolution(pattern, confidence=0.9)
697 r1 = dataclasses.replace(r1, outcome_blob=fake_id("r1ob"), resolution_id=fake_id("r1id"))
698 r2 = dataclasses.replace(r2, outcome_blob=fake_id("r2ob"), resolution_id=fake_id("r2id"))
699 save_resolution(repo, r1)
700 save_resolution(repo, r2)
701
702 best = best_resolution(repo, pattern.pattern_id)
703 assert best is not None
704 assert best.resolution_id == r2.resolution_id
705
706 def test_best_resolution_none_when_no_resolutions(self, repo: pathlib.Path) -> None:
707 pattern = _make_pattern()
708 record_pattern(repo, pattern)
709 assert best_resolution(repo, pattern.pattern_id) is None
710
711
712 class TestPolicyCRUD:
713 """II: save_policy, load_policy, list_policies scope-sorted, remove_policy."""
714
715 def test_save_and_load(self, repo: pathlib.Path) -> None:
716 policy = _make_policy()
717 save_policy(repo, policy)
718 loaded = load_policy(repo, policy.policy_id)
719 assert loaded is not None
720 assert loaded.policy_id == policy.policy_id
721 assert loaded.action == policy.action
722 assert loaded.scope == policy.scope
723
724 def test_save_overwrites_existing(self, repo: pathlib.Path) -> None:
725 policy = _make_policy(action=PolicyAction.PREFER_OURS)
726 save_policy(repo, policy)
727 import dataclasses
728 updated = dataclasses.replace(policy, action=PolicyAction.PREFER_THEIRS)
729 save_policy(repo, updated)
730 loaded = load_policy(repo, policy.policy_id)
731 assert loaded is not None
732 assert loaded.action == PolicyAction.PREFER_THEIRS
733
734 def test_load_nonexistent_returns_none(self, repo: pathlib.Path) -> None:
735 assert load_policy(repo, "missing-policy") is None
736
737 def test_load_invalid_id_returns_none(self, repo: pathlib.Path) -> None:
738 assert load_policy(repo, "bad/policy/id") is None
739
740 def test_list_policies_empty(self, repo: pathlib.Path) -> None:
741 assert list_policies(repo) == []
742
743 def test_list_policies_scope_order(self, repo: pathlib.Path) -> None:
744 """workspace → repo → domain → file regardless of insertion order."""
745 file_p = _make_policy("file-p", scope=PolicyScope.FILE)
746 workspace_p = _make_policy("workspace-p", scope=PolicyScope.WORKSPACE)
747 domain_p = _make_policy("domain-p", scope=PolicyScope.DOMAIN)
748 repo_p = _make_policy("repo-p", scope=PolicyScope.REPO)
749 for p in (file_p, workspace_p, domain_p, repo_p):
750 save_policy(repo, p)
751 results = list_policies(repo)
752 scopes = [r.scope for r in results]
753 assert scopes.index(PolicyScope.WORKSPACE) < scopes.index(PolicyScope.REPO)
754 assert scopes.index(PolicyScope.REPO) < scopes.index(PolicyScope.DOMAIN)
755 assert scopes.index(PolicyScope.DOMAIN) < scopes.index(PolicyScope.FILE)
756
757 def test_remove_policy_returns_true(self, repo: pathlib.Path) -> None:
758 policy = _make_policy()
759 save_policy(repo, policy)
760 assert remove_policy(repo, policy.policy_id) is True
761 assert load_policy(repo, policy.policy_id) is None
762
763 def test_remove_nonexistent_returns_false(self, repo: pathlib.Path) -> None:
764 assert remove_policy(repo, "no-such-policy") is False
765
766 def test_remove_invalid_id_returns_false(self, repo: pathlib.Path) -> None:
767 assert remove_policy(repo, "bad/id") is False
768
769 def test_save_invalid_id_raises(self, repo: pathlib.Path) -> None:
770 import dataclasses
771 policy = _make_policy()
772 bad = dataclasses.replace(policy, policy_id="bad/id")
773 with pytest.raises(ValueError):
774 save_policy(repo, bad)
775
776 def test_condition_round_trips(self, repo: pathlib.Path) -> None:
777 policy = _make_policy(conflict_type=ConflictType.CONTENT, domain="midi", path_pattern="*.mid")
778 save_policy(repo, policy)
779 loaded = load_policy(repo, policy.policy_id)
780 assert loaded is not None
781 assert loaded.when.conflict_type == ConflictType.CONTENT
782 assert loaded.when.domain == "midi"
783 assert loaded.when.path_pattern == "*.mid"
784
785
786 class TestAuditLog:
787 """II: append_audit, list_audit sorted newest-first."""
788
789 def test_append_and_list(self, repo: pathlib.Path) -> None:
790 actor = AgentProvenance.agent("claude-code")
791 append_audit(repo, AuditEventType.PATTERN_RECORDED, actor, pattern_id="a" * 64)
792 entries = list_audit(repo)
793 assert len(entries) == 1
794 assert entries[0]["event_type"] == AuditEventType.PATTERN_RECORDED
795
796 def test_entries_sorted_newest_first(self, repo: pathlib.Path) -> None:
797 actor = AgentProvenance.human()
798 for i in range(3):
799 append_audit(repo, AuditEventType.GC_RUN, actor, metadata={"i": i})
800 time.sleep(0.01) # slight delay so filenames differ
801 entries = list_audit(repo)
802 assert len(entries) == 3
803 # Filenames encode date+uuid — sorted descending means newest at [0]
804 names_in_dir = sorted(
805 (f.name for f in h.audit_dir(repo).iterdir()),
806 reverse=True,
807 )
808 # audit_id is "sha256:<hex>"; filename embeds 12 hex chars starting at index 7
809 assert entries[0]["audit_id"][7:19] in names_in_dir[0]
810
811 def test_list_audit_empty(self, repo: pathlib.Path) -> None:
812 assert list_audit(repo) == []
813
814 def test_limit_respected(self, repo: pathlib.Path) -> None:
815 actor = AgentProvenance.human()
816 for _ in range(10):
817 append_audit(repo, AuditEventType.GC_RUN, actor)
818 entries = list_audit(repo, limit=3)
819 assert len(entries) == 3
820
821 def test_audit_fields_present(self, repo: pathlib.Path) -> None:
822 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
823 append_audit(
824 repo, AuditEventType.RESOLUTION_SAVED, actor,
825 pattern_id="a" * 64, resolution_id="b" * 64,
826 metadata={"extra": "data"},
827 )
828 entry = list_audit(repo)[0]
829 assert entry["event_type"] == AuditEventType.RESOLUTION_SAVED
830 assert entry["pattern_id"] == "a" * 64
831 assert entry["resolution_id"] == "b" * 64
832 assert entry["acted_by"]["agent_id"] == "claude-code"
833 assert entry["metadata"]["extra"] == "data"
834 assert "audit_id" in entry
835 assert "occurred_at" in entry
836
837
838 class TestGcStale:
839 """II: gc_stale keeps resolved patterns and removes old unresolved ones."""
840
841 def test_gc_removes_old_unresolved(self, repo: pathlib.Path) -> None:
842 import dataclasses
843 old_pattern = _make_pattern()
844 old_ts = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
845 old_pattern = dataclasses.replace(old_pattern, recorded_at=old_ts)
846 record_pattern(repo, old_pattern)
847 removed = gc_stale(repo, age_days=1)
848 assert removed == 1
849 assert load_pattern(repo, old_pattern.pattern_id) is None
850
851 def test_gc_keeps_resolved_pattern(self, repo: pathlib.Path) -> None:
852 import dataclasses
853 pattern = _make_pattern()
854 old_ts = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
855 pattern = dataclasses.replace(pattern, recorded_at=old_ts)
856 record_pattern(repo, pattern)
857 res = _make_resolution(pattern)
858 save_resolution(repo, res)
859 removed = gc_stale(repo, age_days=1)
860 assert removed == 0
861 assert load_pattern(repo, pattern.pattern_id) is not None
862
863 def test_gc_keeps_recent_unresolved(self, repo: pathlib.Path) -> None:
864 pattern = _make_pattern() # recorded_at = now
865 record_pattern(repo, pattern)
866 removed = gc_stale(repo, age_days=90)
867 assert removed == 0
868
869 def test_gc_empty_store(self, repo: pathlib.Path) -> None:
870 assert gc_stale(repo, age_days=1) == 0
871
872
873 # ===========================================================================
874 # Tier III — End-to-end lifecycle
875 # ===========================================================================
876
877
878 class TestFullLifecycle:
879 """III: record → save_resolution → best_resolution → increment → gc won't touch it."""
880
881 def test_complete_lifecycle(self, repo: pathlib.Path) -> None:
882 # 1. Record the conflict pattern
883 pattern = _make_pattern(
884 path="tracks/lead.mid",
885 domain="midi",
886 conflict_type=ConflictType.CONTENT,
887 )
888 pid = record_pattern(repo, pattern)
889 assert pid == pattern.pattern_id
890 assert load_pattern(repo, pid) is not None
891
892 # 2. Save a resolution
893 res = _make_resolution(pattern, strategy=ResolutionStrategy.MANUAL, confidence=0.85)
894 save_resolution(repo, res)
895
896 # 3. Retrieve best resolution
897 best = best_resolution(repo, pid)
898 assert best is not None
899 assert best.resolution_id == res.resolution_id
900 assert best.confidence == pytest.approx(0.85)
901
902 # 4. Replay — increment applied count
903 increment_applied_count(repo, pid, res.resolution_id)
904 increment_applied_count(repo, pid, res.resolution_id)
905 reloaded = load_resolution(repo, pid, res.resolution_id)
906 assert reloaded is not None
907 assert reloaded.applied_count == 2
908
909 # 5. GC should NOT remove — has a resolution
910 import dataclasses
911 old_ts = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
912 pattern_old = dataclasses.replace(pattern, recorded_at=old_ts)
913 # Re-record is idempotent so we can't update recorded_at via record_pattern.
914 # Write updated pattern.json directly for this edge-case test.
915 entry_p = h.pattern_dir(repo, pid) / "pattern.json"
916 entry_p.write_text(
917 json.dumps(h._pattern_to_dict(pattern_old), indent=2), encoding="utf-8"
918 )
919 gc_count = gc_stale(repo, age_days=1)
920 assert gc_count == 0 # Protected because it has a resolution
921 assert load_pattern(repo, pid) is not None
922
923 def test_policy_fires_on_matching_pattern(self, repo: pathlib.Path) -> None:
924 policy = _make_policy(
925 "midi-prefer-ours",
926 scope=PolicyScope.DOMAIN,
927 action=PolicyAction.PREFER_OURS,
928 domain="midi",
929 )
930 save_policy(repo, policy)
931
932 pattern = _make_pattern(domain="midi", conflict_type=ConflictType.CONTENT)
933 record_pattern(repo, pattern)
934
935 policies = list_policies(repo)
936 matched = match_policy(policies, pattern)
937 assert matched is not None
938 assert matched.policy_id == "midi-prefer-ours"
939 assert matched.action == PolicyAction.PREFER_OURS
940
941 def test_policy_does_not_fire_wrong_domain(self, repo: pathlib.Path) -> None:
942 policy = _make_policy("midi-only", domain="midi")
943 save_policy(repo, policy)
944 pattern = _make_pattern(domain="code")
945 record_pattern(repo, pattern)
946 matched = match_policy(list_policies(repo), pattern)
947 assert matched is None
948
949 def test_audit_trail_through_lifecycle(self, repo: pathlib.Path) -> None:
950 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
951 pattern = _make_pattern()
952 pid = record_pattern(repo, pattern)
953 append_audit(repo, AuditEventType.PATTERN_RECORDED, actor, pattern_id=pid)
954
955 res = _make_resolution(pattern)
956 save_resolution(repo, res)
957 append_audit(
958 repo, AuditEventType.RESOLUTION_SAVED, actor,
959 pattern_id=pid, resolution_id=res.resolution_id,
960 )
961
962 entries = list_audit(repo)
963 event_types = [e["event_type"] for e in entries]
964 assert AuditEventType.RESOLUTION_SAVED in event_types
965 assert AuditEventType.PATTERN_RECORDED in event_types
966
967
968 # ===========================================================================
969 # Tier IV — Stress tests
970 # ===========================================================================
971
972
973 class TestStress:
974 """IV: 10k pattern scan, concurrent record_pattern, concurrent save_resolution."""
975
976 def test_100_patterns_scan(self, repo: pathlib.Path) -> None:
977 """Store 100 patterns and verify list_patterns returns all of them."""
978 n = 100
979 for i in range(n):
980 record_pattern(repo, _make_pattern(path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}"))
981 results = list_patterns(repo)
982 assert len(results) == n
983
984 def test_concurrent_record_pattern_no_corruption(self, repo: pathlib.Path) -> None:
985 """Concurrent record_pattern from 20 threads — all patterns must be loadable."""
986 patterns = [
987 _make_pattern(path=f"concurrent{i}.mid", ours=f"co{i}", theirs=f"ct{i}")
988 for i in range(20)
989 ]
990 errors: list[Exception] = []
991
992 def worker(p: ConflictPattern) -> None:
993 try:
994 record_pattern(repo, p)
995 except Exception as exc:
996 errors.append(exc)
997
998 with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
999 futures = [executor.submit(worker, p) for p in patterns]
1000 concurrent.futures.wait(futures)
1001
1002 assert errors == [], f"Thread errors: {errors}"
1003 for p in patterns:
1004 assert load_pattern(repo, p.pattern_id) is not None
1005
1006 def test_concurrent_increment_applied_count(self, repo: pathlib.Path) -> None:
1007 """20 concurrent increments — none must crash, final count must be ≥ 1.
1008
1009 ``increment_applied_count`` is a read-modify-write cycle; ``os.replace``
1010 makes each individual write atomic but does not serialise the full cycle.
1011 Under heavy concurrency, updates may be lost (last writer wins). The
1012 guarantee is: no exception, file always valid, count always ≥ 1.
1013 """
1014 pattern = _make_pattern()
1015 record_pattern(repo, pattern)
1016 res = _make_resolution(pattern)
1017 save_resolution(repo, res)
1018
1019 lock = threading.Lock()
1020 errors: list[Exception] = []
1021
1022 def worker() -> None:
1023 try:
1024 increment_applied_count(repo, pattern.pattern_id, res.resolution_id)
1025 except Exception as exc:
1026 with lock:
1027 errors.append(exc)
1028
1029 threads = [threading.Thread(target=worker) for _ in range(20)]
1030 for t in threads:
1031 t.start()
1032 for t in threads:
1033 t.join()
1034
1035 assert errors == [], f"Thread errors: {errors}"
1036 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1037 assert loaded is not None
1038 # At least one increment must have landed; file must be valid JSON
1039 assert loaded.applied_count >= 1
1040
1041 def test_list_patterns_scan_cap_does_not_crash(
1042 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1043 ) -> None:
1044 """If _MAX_SCAN is set to 5, list_patterns truncates rather than crashing."""
1045 monkeypatch.setattr(h, "_MAX_SCAN", 5)
1046 for i in range(10):
1047 record_pattern(repo, _make_pattern(path=f"s{i}.mid", ours=f"o{i}", theirs=f"t{i}"))
1048 results = list_patterns(repo)
1049 assert len(results) <= 5
1050
1051
1052 # ===========================================================================
1053 # Tier V — Data integrity
1054 # ===========================================================================
1055
1056
1057 class TestDataIntegrity:
1058 """V: atomic writes (no temp files left), JSON round-trip, field type preservation."""
1059
1060 def test_no_temp_files_after_record_pattern(self, repo: pathlib.Path) -> None:
1061 pattern = _make_pattern()
1062 record_pattern(repo, pattern)
1063 entry_dir = h.pattern_dir(repo, pattern.pattern_id)
1064 tmp_files = list(entry_dir.glob(".harmony-tmp-*"))
1065 assert tmp_files == []
1066
1067 def test_no_temp_files_after_save_resolution(self, repo: pathlib.Path) -> None:
1068 pattern = _make_pattern()
1069 record_pattern(repo, pattern)
1070 res = _make_resolution(pattern)
1071 save_resolution(repo, res)
1072 res_dir = h._resolutions_dir(repo, pattern.pattern_id)
1073 tmp_files = list(res_dir.glob(".harmony-tmp-*"))
1074 assert tmp_files == []
1075
1076 def test_no_temp_files_after_save_policy(self, repo: pathlib.Path) -> None:
1077 policy = _make_policy()
1078 save_policy(repo, policy)
1079 tmp_files = list(h.policies_dir(repo).glob(".harmony-tmp-*"))
1080 assert tmp_files == []
1081
1082 def test_pattern_json_round_trip(self, repo: pathlib.Path) -> None:
1083 pattern = _make_pattern(
1084 path="round/trip.mid",
1085 domain="midi",
1086 conflict_type=ConflictType.STRUCTURAL,
1087 description={"beats": 4, "key": "Cmaj"},
1088 )
1089 record_pattern(repo, pattern)
1090 loaded = load_pattern(repo, pattern.pattern_id)
1091 assert loaded is not None
1092 assert loaded.path == "round/trip.mid"
1093 assert loaded.domain == "midi"
1094 assert loaded.conflict_type == ConflictType.STRUCTURAL
1095 assert loaded.description == {"beats": 4, "key": "Cmaj"}
1096 assert loaded.ours_id == pattern.ours_id
1097 assert loaded.theirs_id == pattern.theirs_id
1098
1099 def test_resolution_json_round_trip(self, repo: pathlib.Path) -> None:
1100 pattern = _make_pattern()
1101 record_pattern(repo, pattern)
1102 prov = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
1103 res = _make_resolution(pattern, provenance=prov, confidence=0.77, human_verified=True)
1104 save_resolution(repo, res)
1105 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1106 assert loaded is not None
1107 assert loaded.confidence == pytest.approx(0.77)
1108 assert loaded.human_verified is True
1109 assert loaded.resolved_by.type == "agent"
1110 assert loaded.resolved_by.agent_id == "claude-code"
1111 assert loaded.resolved_by.model_id == "claude-sonnet-4-6"
1112
1113 def test_policy_json_round_trip(self, repo: pathlib.Path) -> None:
1114 policy = _make_policy(
1115 "round-trip-policy",
1116 scope=PolicyScope.WORKSPACE,
1117 action=PolicyAction.ESCALATE,
1118 confidence=0.6,
1119 conflict_type=ConflictType.RELATIONAL,
1120 domain="code",
1121 path_pattern="src/**",
1122 )
1123 save_policy(repo, policy)
1124 loaded = load_policy(repo, policy.policy_id)
1125 assert loaded is not None
1126 assert loaded.scope == PolicyScope.WORKSPACE
1127 assert loaded.action == PolicyAction.ESCALATE
1128 assert loaded.confidence == pytest.approx(0.6)
1129 assert loaded.when.conflict_type == ConflictType.RELATIONAL
1130 assert loaded.when.domain == "code"
1131 assert loaded.when.path_pattern == "src/**"
1132
1133 def test_agent_provenance_round_trip(self) -> None:
1134 p = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
1135 d = p.to_dict()
1136 restored = AgentProvenance.from_dict(d)
1137 assert restored == p
1138
1139 def test_agent_provenance_human_round_trip(self) -> None:
1140 p = AgentProvenance.human()
1141 restored = AgentProvenance.from_dict(p.to_dict())
1142 assert restored == p
1143
1144 def test_recorded_at_is_utc_aware(self, repo: pathlib.Path) -> None:
1145 pattern = _make_pattern()
1146 record_pattern(repo, pattern)
1147 loaded = load_pattern(repo, pattern.pattern_id)
1148 assert loaded is not None
1149 assert loaded.recorded_at.tzinfo is not None
1150
1151 def test_resolved_at_is_utc_aware(self, repo: pathlib.Path) -> None:
1152 pattern = _make_pattern()
1153 record_pattern(repo, pattern)
1154 res = _make_resolution(pattern)
1155 save_resolution(repo, res)
1156 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1157 assert loaded is not None
1158 assert loaded.resolved_at.tzinfo is not None
1159
1160 def test_applied_count_starts_at_zero(self, repo: pathlib.Path) -> None:
1161 pattern = _make_pattern()
1162 record_pattern(repo, pattern)
1163 res = _make_resolution(pattern)
1164 save_resolution(repo, res)
1165 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1166 assert loaded is not None
1167 assert loaded.applied_count == 0
1168
1169
1170 # ===========================================================================
1171 # Tier VI — Security
1172 # ===========================================================================
1173
1174
1175 class TestSecurity:
1176 """VI: path traversal, symlink guards, size caps, crafted policy_id."""
1177
1178 def test_path_traversal_in_pattern_id_rejected(self, repo: pathlib.Path) -> None:
1179 with pytest.raises(ValueError):
1180 h._validate_id("../../../etc/passwd")
1181
1182 def test_path_traversal_in_load_pattern(self, repo: pathlib.Path) -> None:
1183 result = load_pattern(repo, "../evil")
1184 assert result is None
1185
1186 def test_path_traversal_in_load_resolution(self, repo: pathlib.Path) -> None:
1187 result = load_resolution(repo, "a" * 64, "../evil")
1188 assert result is None
1189
1190 def test_path_traversal_in_forget_pattern(self, repo: pathlib.Path) -> None:
1191 result = forget_pattern(repo, "../evil/../../../../../etc")
1192 assert result is False
1193
1194 def test_crafted_policy_id_with_slash(self, repo: pathlib.Path) -> None:
1195 result = load_policy(repo, "../../etc/passwd")
1196 assert result is None
1197
1198 def test_crafted_policy_id_with_null_byte(self, repo: pathlib.Path) -> None:
1199 with pytest.raises(ValueError):
1200 h._validate_policy_id("policy\x00id")
1201
1202 def test_symlinks_in_patterns_dir_skipped(self, repo: pathlib.Path) -> None:
1203 pdir = h.patterns_dir(repo)
1204 pdir.mkdir(parents=True, exist_ok=True)
1205 # Create a symlink in the patterns dir
1206 link = pdir / ("s" * 64)
1207 target = repo / "other"
1208 target.mkdir()
1209 link.symlink_to(target)
1210 results = list_patterns(repo)
1211 assert results == [] # symlink skipped
1212
1213 def test_symlinks_in_policies_dir_skipped(self, repo: pathlib.Path) -> None:
1214 poldir = h.policies_dir(repo)
1215 poldir.mkdir(parents=True, exist_ok=True)
1216 link = poldir / "linked-policy.json"
1217 target = repo / "victim.json"
1218 target.write_text('{"evil": true}')
1219 link.symlink_to(target)
1220 results = list_policies(repo)
1221 assert results == []
1222
1223 def test_oversized_pattern_file_rejected(self, repo: pathlib.Path) -> None:
1224 pattern = _make_pattern()
1225 record_pattern(repo, pattern)
1226 meta_p = h.pattern_dir(repo, pattern.pattern_id) / "pattern.json"
1227 # Overwrite with a file exceeding cap
1228 meta_p.write_bytes(b"x" * (_MAX_PATTERN_BYTES + 1))
1229 result = load_pattern(repo, pattern.pattern_id)
1230 assert result is None
1231
1232 def test_oversized_resolution_file_rejected(self, repo: pathlib.Path) -> None:
1233 pattern = _make_pattern()
1234 record_pattern(repo, pattern)
1235 res = _make_resolution(pattern)
1236 save_resolution(repo, res)
1237 dest = h._resolution_path(repo, pattern.pattern_id, res.resolution_id)
1238 dest.write_bytes(b"y" * (_MAX_RESOLUTION_BYTES + 1))
1239 result = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1240 assert result is None
1241
1242 def test_oversized_policy_file_rejected(self, repo: pathlib.Path) -> None:
1243 policy = _make_policy()
1244 save_policy(repo, policy)
1245 dest = h.policies_dir(repo) / f"{policy.policy_id}.json"
1246 dest.write_bytes(b"z" * (_MAX_POLICY_BYTES + 1))
1247 result = load_policy(repo, policy.policy_id)
1248 assert result is None
1249
1250 def test_malformed_json_pattern_returns_none(self, repo: pathlib.Path) -> None:
1251 pid = "a" * 64
1252 entry = h.pattern_dir(repo, pid)
1253 entry.mkdir(parents=True)
1254 (entry / "pattern.json").write_text("not json", encoding="utf-8")
1255 assert load_pattern(repo, pid) is None
1256
1257 def test_malformed_json_resolution_returns_none(self, repo: pathlib.Path) -> None:
1258 pattern = _make_pattern()
1259 record_pattern(repo, pattern)
1260 rid = "b" * 64
1261 dest = h.resolution_path(repo, pattern.pattern_id, rid)
1262 dest.parent.mkdir(parents=True, exist_ok=True)
1263 dest.write_text("{{bad", encoding="utf-8")
1264 assert load_resolution(repo, pattern.pattern_id, rid) is None
1265
1266 def test_non_hex_dir_in_patterns_dir_skipped(self, repo: pathlib.Path) -> None:
1267 pdir = h.patterns_dir(repo)
1268 pdir.mkdir(parents=True, exist_ok=True)
1269 (pdir / "not-a-valid-id").mkdir()
1270 assert list_patterns(repo) == []
1271
1272
1273 # ===========================================================================
1274 # Tier VII — Performance
1275 # ===========================================================================
1276
1277
1278 class TestPerformance:
1279 """VII: operation timing assertions — single operations must be fast."""
1280
1281 def test_record_pattern_under_50ms(self, repo: pathlib.Path) -> None:
1282 pattern = _make_pattern()
1283 start = time.monotonic()
1284 record_pattern(repo, pattern)
1285 elapsed = (time.monotonic() - start) * 1000
1286 assert elapsed < 50, f"record_pattern took {elapsed:.1f}ms — expected <50ms"
1287
1288 def test_load_pattern_under_10ms(self, repo: pathlib.Path) -> None:
1289 pattern = _make_pattern()
1290 record_pattern(repo, pattern)
1291 start = time.monotonic()
1292 load_pattern(repo, pattern.pattern_id)
1293 elapsed = (time.monotonic() - start) * 1000
1294 assert elapsed < 10, f"load_pattern took {elapsed:.1f}ms — expected <10ms"
1295
1296 def test_save_resolution_under_50ms(self, repo: pathlib.Path) -> None:
1297 pattern = _make_pattern()
1298 record_pattern(repo, pattern)
1299 res = _make_resolution(pattern)
1300 start = time.monotonic()
1301 save_resolution(repo, res)
1302 elapsed = (time.monotonic() - start) * 1000
1303 assert elapsed < 50, f"save_resolution took {elapsed:.1f}ms — expected <50ms"
1304
1305 def test_list_100_patterns_under_500ms(self, repo: pathlib.Path) -> None:
1306 n = 100
1307 for i in range(n):
1308 record_pattern(repo, _make_pattern(path=f"perf{i}.mid", ours=f"o{i}", theirs=f"t{i}"))
1309 start = time.monotonic()
1310 results = list_patterns(repo)
1311 elapsed = (time.monotonic() - start) * 1000
1312 assert len(results) == n
1313 assert elapsed < 500, f"list_patterns(100) took {elapsed:.1f}ms — expected <500ms"
1314
1315 def test_save_and_load_policy_under_20ms(self, repo: pathlib.Path) -> None:
1316 policy = _make_policy()
1317 start = time.monotonic()
1318 save_policy(repo, policy)
1319 load_policy(repo, policy.policy_id)
1320 elapsed = (time.monotonic() - start) * 1000
1321 assert elapsed < 20, f"save+load policy took {elapsed:.1f}ms — expected <20ms"
1322
1323 def test_append_audit_under_20ms(self, repo: pathlib.Path) -> None:
1324 actor = AgentProvenance.human()
1325 start = time.monotonic()
1326 append_audit(repo, AuditEventType.GC_RUN, actor)
1327 elapsed = (time.monotonic() - start) * 1000
1328 assert elapsed < 20, f"append_audit took {elapsed:.1f}ms — expected <20ms"
1329
1330 def test_increment_applied_count_under_20ms(self, repo: pathlib.Path) -> None:
1331 pattern = _make_pattern()
1332 record_pattern(repo, pattern)
1333 res = _make_resolution(pattern)
1334 save_resolution(repo, res)
1335 start = time.monotonic()
1336 increment_applied_count(repo, pattern.pattern_id, res.resolution_id)
1337 elapsed = (time.monotonic() - start) * 1000
1338 assert elapsed < 20, f"increment_applied_count took {elapsed:.1f}ms — expected <20ms"
File History 2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 135 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 141 days ago