gabriel / muse public
test_harmony_phase1.py python
1,338 lines 53.2 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 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
15 import concurrent.futures
16 import datetime
17 import hashlib
18 import json
19 import os
20 import pathlib
21 import tempfile
22 import threading
23 import time
24 from dataclasses import FrozenInstanceError
25 from typing import Any
26 from unittest import mock
27
28 import pytest
29
30 import muse.core.harmony as h
31 from muse.core.harmony import (
32 AgentProvenance,
33 AuditEvent,
34 AuditEventType,
35 ConflictPattern,
36 ConflictType,
37 Policy,
38 PolicyAction,
39 PolicyCondition,
40 PolicyScope,
41 Resolution,
42 ResolutionProposal,
43 ResolutionStrategy,
44 _MAX_AUDIT_BYTES,
45 _MAX_PATTERN_BYTES,
46 _MAX_POLICY_BYTES,
47 _MAX_RESOLUTION_BYTES,
48 _MAX_SCAN,
49 _condition_matches,
50 append_audit,
51 best_resolution,
52 blob_fingerprint,
53 clear_all,
54 compute_pattern_id,
55 compute_resolution_id,
56 forget_pattern,
57 gc_stale,
58 increment_applied_count,
59 list_audit,
60 list_patterns,
61 list_policies,
62 list_resolutions,
63 load_pattern,
64 load_policy,
65 load_resolution,
66 match_policy,
67 record_pattern,
68 remove_policy,
69 save_policy,
70 save_resolution,
71 )
72
73
74 # ---------------------------------------------------------------------------
75 # Shared fixtures
76 # ---------------------------------------------------------------------------
77
78
79 @pytest.fixture()
80 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
81 """Return a temporary directory acting as a bare repo root."""
82 (tmp_path / ".muse").mkdir()
83 return tmp_path
84
85
86 def _fake_id(seed: str) -> str:
87 """Return a deterministic 64-char hex string from *seed*."""
88 return hashlib.sha256(seed.encode()).hexdigest()
89
90
91 def _utc_now() -> datetime.datetime:
92 return datetime.datetime.now(datetime.timezone.utc)
93
94
95 def _make_pattern(
96 path: str = "track.mid",
97 domain: str = "midi",
98 conflict_type: str = ConflictType.CONTENT,
99 ours: str = "ours",
100 theirs: str = "theirs",
101 description: dict[str, Any] | None = None,
102 recorded_by: str = "claude-code",
103 ) -> ConflictPattern:
104 """Build a ConflictPattern with sensible defaults."""
105 ours_id = _fake_id(ours)
106 theirs_id = _fake_id(theirs)
107 blob_fp = blob_fingerprint(ours_id, theirs_id)
108 semantic_fp = blob_fp
109 pattern_id = compute_pattern_id(path, blob_fp, semantic_fp)
110 return ConflictPattern(
111 pattern_id=pattern_id,
112 path=path,
113 domain=domain,
114 conflict_type=conflict_type,
115 blob_fingerprint=blob_fp,
116 semantic_fingerprint=semantic_fp,
117 ours_id=ours_id,
118 theirs_id=theirs_id,
119 description=description or {},
120 recorded_at=_utc_now(),
121 recorded_by=recorded_by,
122 )
123
124
125 def _make_resolution(
126 pattern: ConflictPattern,
127 strategy: str = ResolutionStrategy.MANUAL,
128 confidence: float = 0.9,
129 human_verified: bool = False,
130 provenance: AgentProvenance | None = None,
131 policy_id: str | None = None,
132 ) -> Resolution:
133 """Build a Resolution tied to *pattern* with sensible defaults."""
134 outcome_blob = _fake_id(f"outcome-{pattern.pattern_id[:8]}")
135 prov = provenance or AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
136 resolved_at = _utc_now()
137 resolution_id = compute_resolution_id(
138 pattern.pattern_id, outcome_blob, strategy, prov, resolved_at
139 )
140 return Resolution(
141 resolution_id=resolution_id,
142 pattern_id=pattern.pattern_id,
143 strategy=strategy,
144 policy_id=policy_id,
145 outcome_blob=outcome_blob,
146 resolved_by=prov,
147 human_verified=human_verified,
148 confidence=confidence,
149 rationale="Test rationale",
150 resolved_at=resolved_at,
151 )
152
153
154 def _make_policy(
155 policy_id: str = "always-prefer-ours",
156 scope: str = PolicyScope.REPO,
157 action: str = PolicyAction.PREFER_OURS,
158 confidence: float = 0.95,
159 conflict_type: str | None = None,
160 domain: str | None = None,
161 path_pattern: str | None = None,
162 ) -> Policy:
163 """Build a Policy with sensible defaults."""
164 return Policy(
165 policy_id=policy_id,
166 description="Test policy",
167 when=PolicyCondition(
168 conflict_type=conflict_type,
169 domain=domain,
170 path_pattern=path_pattern,
171 ),
172 action=action,
173 confidence=confidence,
174 escalate_to=None,
175 delegate_to=None,
176 scope=scope,
177 created_at=_utc_now(),
178 created_by="claude-code",
179 )
180
181
182 # ===========================================================================
183 # Tier I — Unit tests
184 # ===========================================================================
185
186
187 class TestBlobFingerprint:
188 """I: blob_fingerprint must be commutative and deterministic."""
189
190 def test_commutativity(self) -> None:
191 a, b = _fake_id("A"), _fake_id("B")
192 assert blob_fingerprint(a, b) == blob_fingerprint(b, a)
193
194 def test_determinism(self) -> None:
195 a, b = _fake_id("X"), _fake_id("Y")
196 fp1 = blob_fingerprint(a, b)
197 fp2 = blob_fingerprint(a, b)
198 assert fp1 == fp2
199
200 def test_output_is_64_hex(self) -> None:
201 a, b = _fake_id("p"), _fake_id("q")
202 fp = blob_fingerprint(a, b)
203 assert len(fp) == 64
204 assert all(c in "0123456789abcdef" for c in fp)
205
206 def test_distinct_pairs_differ(self) -> None:
207 ab = blob_fingerprint(_fake_id("A"), _fake_id("B"))
208 cd = blob_fingerprint(_fake_id("C"), _fake_id("D"))
209 assert ab != cd
210
211 def test_same_id_both_sides(self) -> None:
212 a = _fake_id("same")
213 # Should not crash; result is deterministic
214 fp = blob_fingerprint(a, a)
215 assert len(fp) == 64
216
217
218 class TestComputePatternId:
219 """I: compute_pattern_id includes path, so same content → different IDs for different paths."""
220
221 def test_deterministic(self) -> None:
222 blob_fp = _fake_id("blob")
223 sem_fp = _fake_id("sem")
224 p1 = compute_pattern_id("track.mid", blob_fp, sem_fp)
225 p2 = compute_pattern_id("track.mid", blob_fp, sem_fp)
226 assert p1 == p2
227
228 def test_path_changes_id(self) -> None:
229 blob_fp = _fake_id("blob")
230 sem_fp = _fake_id("sem")
231 p1 = compute_pattern_id("track.mid", blob_fp, sem_fp)
232 p2 = compute_pattern_id("drums.mid", blob_fp, sem_fp)
233 assert p1 != p2
234
235 def test_blob_changes_id(self) -> None:
236 sem_fp = _fake_id("sem")
237 p1 = compute_pattern_id("track.mid", _fake_id("blobA"), sem_fp)
238 p2 = compute_pattern_id("track.mid", _fake_id("blobB"), sem_fp)
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 len(pid) == 64
244 assert all(c in "0123456789abcdef" for c in pid)
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 # Should not crash and should produce 64 hex chars
274 rid = compute_resolution_id(pid, ob, ResolutionStrategy.MANUAL, AgentProvenance.human(), ts)
275 assert len(rid) == 64
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("a" * 64) # no exception
283
284 def test_validate_id_rejects_63_chars(self) -> None:
285 with pytest.raises(ValueError, match="64 lowercase hexadecimal"):
286 h._validate_id("a" * 63)
287
288 def test_validate_id_rejects_65_chars(self) -> None:
289 with pytest.raises(ValueError, match="64 lowercase hexadecimal"):
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: Any) -> 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.patterns_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 assert entries[0]["audit_id"] in names_in_dir[0]
809
810 def test_list_audit_empty(self, repo: pathlib.Path) -> None:
811 assert list_audit(repo) == []
812
813 def test_limit_respected(self, repo: pathlib.Path) -> None:
814 actor = AgentProvenance.human()
815 for _ in range(10):
816 append_audit(repo, AuditEventType.GC_RUN, actor)
817 entries = list_audit(repo, limit=3)
818 assert len(entries) == 3
819
820 def test_audit_fields_present(self, repo: pathlib.Path) -> None:
821 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
822 append_audit(
823 repo, AuditEventType.RESOLUTION_SAVED, actor,
824 pattern_id="a" * 64, resolution_id="b" * 64,
825 metadata={"extra": "data"},
826 )
827 entry = list_audit(repo)[0]
828 assert entry["event_type"] == AuditEventType.RESOLUTION_SAVED
829 assert entry["pattern_id"] == "a" * 64
830 assert entry["resolution_id"] == "b" * 64
831 assert entry["acted_by"]["agent_id"] == "claude-code"
832 assert entry["metadata"]["extra"] == "data"
833 assert "audit_id" in entry
834 assert "occurred_at" in entry
835
836
837 class TestGcStale:
838 """II: gc_stale keeps resolved patterns and removes old unresolved ones."""
839
840 def test_gc_removes_old_unresolved(self, repo: pathlib.Path) -> None:
841 import dataclasses
842 old_pattern = _make_pattern()
843 old_ts = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
844 old_pattern = dataclasses.replace(old_pattern, recorded_at=old_ts)
845 record_pattern(repo, old_pattern)
846 removed = gc_stale(repo, age_days=1)
847 assert removed == 1
848 assert load_pattern(repo, old_pattern.pattern_id) is None
849
850 def test_gc_keeps_resolved_pattern(self, repo: pathlib.Path) -> None:
851 import dataclasses
852 pattern = _make_pattern()
853 old_ts = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
854 pattern = dataclasses.replace(pattern, recorded_at=old_ts)
855 record_pattern(repo, pattern)
856 res = _make_resolution(pattern)
857 save_resolution(repo, res)
858 removed = gc_stale(repo, age_days=1)
859 assert removed == 0
860 assert load_pattern(repo, pattern.pattern_id) is not None
861
862 def test_gc_keeps_recent_unresolved(self, repo: pathlib.Path) -> None:
863 pattern = _make_pattern() # recorded_at = now
864 record_pattern(repo, pattern)
865 removed = gc_stale(repo, age_days=90)
866 assert removed == 0
867
868 def test_gc_empty_store(self, repo: pathlib.Path) -> None:
869 assert gc_stale(repo, age_days=1) == 0
870
871
872 # ===========================================================================
873 # Tier III — End-to-end lifecycle
874 # ===========================================================================
875
876
877 class TestFullLifecycle:
878 """III: record → save_resolution → best_resolution → increment → gc won't touch it."""
879
880 def test_complete_lifecycle(self, repo: pathlib.Path) -> None:
881 # 1. Record the conflict pattern
882 pattern = _make_pattern(
883 path="tracks/lead.mid",
884 domain="midi",
885 conflict_type=ConflictType.CONTENT,
886 )
887 pid = record_pattern(repo, pattern)
888 assert pid == pattern.pattern_id
889 assert load_pattern(repo, pid) is not None
890
891 # 2. Save a resolution
892 res = _make_resolution(pattern, strategy=ResolutionStrategy.MANUAL, confidence=0.85)
893 save_resolution(repo, res)
894
895 # 3. Retrieve best resolution
896 best = best_resolution(repo, pid)
897 assert best is not None
898 assert best.resolution_id == res.resolution_id
899 assert best.confidence == pytest.approx(0.85)
900
901 # 4. Replay — increment applied count
902 increment_applied_count(repo, pid, res.resolution_id)
903 increment_applied_count(repo, pid, res.resolution_id)
904 reloaded = load_resolution(repo, pid, res.resolution_id)
905 assert reloaded is not None
906 assert reloaded.applied_count == 2
907
908 # 5. GC should NOT remove — has a resolution
909 import dataclasses
910 old_ts = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
911 pattern_old = dataclasses.replace(pattern, recorded_at=old_ts)
912 # Re-record is idempotent so we can't update recorded_at via record_pattern.
913 # Write updated pattern.json directly for this edge-case test.
914 entry_p = h.patterns_dir(repo) / pid / "pattern.json"
915 entry_p.write_text(
916 json.dumps(h._pattern_to_dict(pattern_old), indent=2), encoding="utf-8"
917 )
918 gc_count = gc_stale(repo, age_days=1)
919 assert gc_count == 0 # Protected because it has a resolution
920 assert load_pattern(repo, pid) is not None
921
922 def test_policy_fires_on_matching_pattern(self, repo: pathlib.Path) -> None:
923 policy = _make_policy(
924 "midi-prefer-ours",
925 scope=PolicyScope.DOMAIN,
926 action=PolicyAction.PREFER_OURS,
927 domain="midi",
928 )
929 save_policy(repo, policy)
930
931 pattern = _make_pattern(domain="midi", conflict_type=ConflictType.CONTENT)
932 record_pattern(repo, pattern)
933
934 policies = list_policies(repo)
935 matched = match_policy(policies, pattern)
936 assert matched is not None
937 assert matched.policy_id == "midi-prefer-ours"
938 assert matched.action == PolicyAction.PREFER_OURS
939
940 def test_policy_does_not_fire_wrong_domain(self, repo: pathlib.Path) -> None:
941 policy = _make_policy("midi-only", domain="midi")
942 save_policy(repo, policy)
943 pattern = _make_pattern(domain="code")
944 record_pattern(repo, pattern)
945 matched = match_policy(list_policies(repo), pattern)
946 assert matched is None
947
948 def test_audit_trail_through_lifecycle(self, repo: pathlib.Path) -> None:
949 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
950 pattern = _make_pattern()
951 pid = record_pattern(repo, pattern)
952 append_audit(repo, AuditEventType.PATTERN_RECORDED, actor, pattern_id=pid)
953
954 res = _make_resolution(pattern)
955 save_resolution(repo, res)
956 append_audit(
957 repo, AuditEventType.RESOLUTION_SAVED, actor,
958 pattern_id=pid, resolution_id=res.resolution_id,
959 )
960
961 entries = list_audit(repo)
962 event_types = [e["event_type"] for e in entries]
963 assert AuditEventType.RESOLUTION_SAVED in event_types
964 assert AuditEventType.PATTERN_RECORDED in event_types
965
966
967 # ===========================================================================
968 # Tier IV — Stress tests
969 # ===========================================================================
970
971
972 class TestStress:
973 """IV: 10k pattern scan, concurrent record_pattern, concurrent save_resolution."""
974
975 def test_100_patterns_scan(self, repo: pathlib.Path) -> None:
976 """Store 100 patterns and verify list_patterns returns all of them."""
977 n = 100
978 for i in range(n):
979 record_pattern(repo, _make_pattern(path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}"))
980 results = list_patterns(repo)
981 assert len(results) == n
982
983 def test_concurrent_record_pattern_no_corruption(self, repo: pathlib.Path) -> None:
984 """Concurrent record_pattern from 20 threads — all patterns must be loadable."""
985 patterns = [
986 _make_pattern(path=f"concurrent{i}.mid", ours=f"co{i}", theirs=f"ct{i}")
987 for i in range(20)
988 ]
989 errors: list[Exception] = []
990
991 def worker(p: ConflictPattern) -> None:
992 try:
993 record_pattern(repo, p)
994 except Exception as exc:
995 errors.append(exc)
996
997 with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
998 futures = [executor.submit(worker, p) for p in patterns]
999 concurrent.futures.wait(futures)
1000
1001 assert errors == [], f"Thread errors: {errors}"
1002 for p in patterns:
1003 assert load_pattern(repo, p.pattern_id) is not None
1004
1005 def test_concurrent_increment_applied_count(self, repo: pathlib.Path) -> None:
1006 """20 concurrent increments — none must crash, final count must be ≥ 1.
1007
1008 ``increment_applied_count`` is a read-modify-write cycle; ``os.replace``
1009 makes each individual write atomic but does not serialise the full cycle.
1010 Under heavy concurrency, updates may be lost (last writer wins). The
1011 guarantee is: no exception, file always valid, count always ≥ 1.
1012 """
1013 pattern = _make_pattern()
1014 record_pattern(repo, pattern)
1015 res = _make_resolution(pattern)
1016 save_resolution(repo, res)
1017
1018 lock = threading.Lock()
1019 errors: list[Exception] = []
1020
1021 def worker() -> None:
1022 try:
1023 increment_applied_count(repo, pattern.pattern_id, res.resolution_id)
1024 except Exception as exc:
1025 with lock:
1026 errors.append(exc)
1027
1028 threads = [threading.Thread(target=worker) for _ in range(20)]
1029 for t in threads:
1030 t.start()
1031 for t in threads:
1032 t.join()
1033
1034 assert errors == [], f"Thread errors: {errors}"
1035 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1036 assert loaded is not None
1037 # At least one increment must have landed; file must be valid JSON
1038 assert loaded.applied_count >= 1
1039
1040 def test_list_patterns_scan_cap_does_not_crash(
1041 self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1042 ) -> None:
1043 """If _MAX_SCAN is set to 5, list_patterns truncates rather than crashing."""
1044 monkeypatch.setattr(h, "_MAX_SCAN", 5)
1045 for i in range(10):
1046 record_pattern(repo, _make_pattern(path=f"s{i}.mid", ours=f"o{i}", theirs=f"t{i}"))
1047 results = list_patterns(repo)
1048 assert len(results) <= 5
1049
1050
1051 # ===========================================================================
1052 # Tier V — Data integrity
1053 # ===========================================================================
1054
1055
1056 class TestDataIntegrity:
1057 """V: atomic writes (no temp files left), JSON round-trip, field type preservation."""
1058
1059 def test_no_temp_files_after_record_pattern(self, repo: pathlib.Path) -> None:
1060 pattern = _make_pattern()
1061 record_pattern(repo, pattern)
1062 entry_dir = h.patterns_dir(repo) / pattern.pattern_id
1063 tmp_files = list(entry_dir.glob(".harmony-tmp-*"))
1064 assert tmp_files == []
1065
1066 def test_no_temp_files_after_save_resolution(self, repo: pathlib.Path) -> None:
1067 pattern = _make_pattern()
1068 record_pattern(repo, pattern)
1069 res = _make_resolution(pattern)
1070 save_resolution(repo, res)
1071 res_dir = h._resolutions_dir(repo, pattern.pattern_id)
1072 tmp_files = list(res_dir.glob(".harmony-tmp-*"))
1073 assert tmp_files == []
1074
1075 def test_no_temp_files_after_save_policy(self, repo: pathlib.Path) -> None:
1076 policy = _make_policy()
1077 save_policy(repo, policy)
1078 tmp_files = list(h.policies_dir(repo).glob(".harmony-tmp-*"))
1079 assert tmp_files == []
1080
1081 def test_pattern_json_round_trip(self, repo: pathlib.Path) -> None:
1082 pattern = _make_pattern(
1083 path="round/trip.mid",
1084 domain="midi",
1085 conflict_type=ConflictType.STRUCTURAL,
1086 description={"beats": 4, "key": "Cmaj"},
1087 )
1088 record_pattern(repo, pattern)
1089 loaded = load_pattern(repo, pattern.pattern_id)
1090 assert loaded is not None
1091 assert loaded.path == "round/trip.mid"
1092 assert loaded.domain == "midi"
1093 assert loaded.conflict_type == ConflictType.STRUCTURAL
1094 assert loaded.description == {"beats": 4, "key": "Cmaj"}
1095 assert loaded.ours_id == pattern.ours_id
1096 assert loaded.theirs_id == pattern.theirs_id
1097
1098 def test_resolution_json_round_trip(self, repo: pathlib.Path) -> None:
1099 pattern = _make_pattern()
1100 record_pattern(repo, pattern)
1101 prov = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
1102 res = _make_resolution(pattern, provenance=prov, confidence=0.77, human_verified=True)
1103 save_resolution(repo, res)
1104 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1105 assert loaded is not None
1106 assert loaded.confidence == pytest.approx(0.77)
1107 assert loaded.human_verified is True
1108 assert loaded.resolved_by.type == "agent"
1109 assert loaded.resolved_by.agent_id == "claude-code"
1110 assert loaded.resolved_by.model_id == "claude-sonnet-4-6"
1111
1112 def test_policy_json_round_trip(self, repo: pathlib.Path) -> None:
1113 policy = _make_policy(
1114 "round-trip-policy",
1115 scope=PolicyScope.WORKSPACE,
1116 action=PolicyAction.ESCALATE,
1117 confidence=0.6,
1118 conflict_type=ConflictType.RELATIONAL,
1119 domain="code",
1120 path_pattern="src/**",
1121 )
1122 save_policy(repo, policy)
1123 loaded = load_policy(repo, policy.policy_id)
1124 assert loaded is not None
1125 assert loaded.scope == PolicyScope.WORKSPACE
1126 assert loaded.action == PolicyAction.ESCALATE
1127 assert loaded.confidence == pytest.approx(0.6)
1128 assert loaded.when.conflict_type == ConflictType.RELATIONAL
1129 assert loaded.when.domain == "code"
1130 assert loaded.when.path_pattern == "src/**"
1131
1132 def test_agent_provenance_round_trip(self) -> None:
1133 p = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
1134 d = p.to_dict()
1135 restored = AgentProvenance.from_dict(d)
1136 assert restored == p
1137
1138 def test_agent_provenance_human_round_trip(self) -> None:
1139 p = AgentProvenance.human()
1140 restored = AgentProvenance.from_dict(p.to_dict())
1141 assert restored == p
1142
1143 def test_recorded_at_is_utc_aware(self, repo: pathlib.Path) -> None:
1144 pattern = _make_pattern()
1145 record_pattern(repo, pattern)
1146 loaded = load_pattern(repo, pattern.pattern_id)
1147 assert loaded is not None
1148 assert loaded.recorded_at.tzinfo is not None
1149
1150 def test_resolved_at_is_utc_aware(self, repo: pathlib.Path) -> None:
1151 pattern = _make_pattern()
1152 record_pattern(repo, pattern)
1153 res = _make_resolution(pattern)
1154 save_resolution(repo, res)
1155 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1156 assert loaded is not None
1157 assert loaded.resolved_at.tzinfo is not None
1158
1159 def test_applied_count_starts_at_zero(self, repo: pathlib.Path) -> None:
1160 pattern = _make_pattern()
1161 record_pattern(repo, pattern)
1162 res = _make_resolution(pattern)
1163 save_resolution(repo, res)
1164 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1165 assert loaded is not None
1166 assert loaded.applied_count == 0
1167
1168
1169 # ===========================================================================
1170 # Tier VI — Security
1171 # ===========================================================================
1172
1173
1174 class TestSecurity:
1175 """VI: path traversal, symlink guards, size caps, crafted policy_id."""
1176
1177 def test_path_traversal_in_pattern_id_rejected(self, repo: pathlib.Path) -> None:
1178 with pytest.raises(ValueError):
1179 h._validate_id("../../../etc/passwd")
1180
1181 def test_path_traversal_in_load_pattern(self, repo: pathlib.Path) -> None:
1182 result = load_pattern(repo, "../evil")
1183 assert result is None
1184
1185 def test_path_traversal_in_load_resolution(self, repo: pathlib.Path) -> None:
1186 result = load_resolution(repo, "a" * 64, "../evil")
1187 assert result is None
1188
1189 def test_path_traversal_in_forget_pattern(self, repo: pathlib.Path) -> None:
1190 result = forget_pattern(repo, "../evil/../../../../../etc")
1191 assert result is False
1192
1193 def test_crafted_policy_id_with_slash(self, repo: pathlib.Path) -> None:
1194 result = load_policy(repo, "../../etc/passwd")
1195 assert result is None
1196
1197 def test_crafted_policy_id_with_null_byte(self, repo: pathlib.Path) -> None:
1198 with pytest.raises(ValueError):
1199 h._validate_policy_id("policy\x00id")
1200
1201 def test_symlinks_in_patterns_dir_skipped(self, repo: pathlib.Path) -> None:
1202 pdir = h.patterns_dir(repo)
1203 pdir.mkdir(parents=True, exist_ok=True)
1204 # Create a symlink in the patterns dir
1205 link = pdir / ("s" * 64)
1206 target = repo / "other"
1207 target.mkdir()
1208 link.symlink_to(target)
1209 results = list_patterns(repo)
1210 assert results == [] # symlink skipped
1211
1212 def test_symlinks_in_policies_dir_skipped(self, repo: pathlib.Path) -> None:
1213 poldir = h.policies_dir(repo)
1214 poldir.mkdir(parents=True, exist_ok=True)
1215 link = poldir / "linked-policy.json"
1216 target = repo / "victim.json"
1217 target.write_text('{"evil": true}')
1218 link.symlink_to(target)
1219 results = list_policies(repo)
1220 assert results == []
1221
1222 def test_oversized_pattern_file_rejected(self, repo: pathlib.Path) -> None:
1223 pattern = _make_pattern()
1224 record_pattern(repo, pattern)
1225 meta_p = h.patterns_dir(repo) / pattern.pattern_id / "pattern.json"
1226 # Overwrite with a file exceeding cap
1227 meta_p.write_bytes(b"x" * (_MAX_PATTERN_BYTES + 1))
1228 result = load_pattern(repo, pattern.pattern_id)
1229 assert result is None
1230
1231 def test_oversized_resolution_file_rejected(self, repo: pathlib.Path) -> None:
1232 pattern = _make_pattern()
1233 record_pattern(repo, pattern)
1234 res = _make_resolution(pattern)
1235 save_resolution(repo, res)
1236 dest = h._resolutions_dir(repo, pattern.pattern_id) / f"{res.resolution_id}.json"
1237 dest.write_bytes(b"y" * (_MAX_RESOLUTION_BYTES + 1))
1238 result = load_resolution(repo, pattern.pattern_id, res.resolution_id)
1239 assert result is None
1240
1241 def test_oversized_policy_file_rejected(self, repo: pathlib.Path) -> None:
1242 policy = _make_policy()
1243 save_policy(repo, policy)
1244 dest = h.policies_dir(repo) / f"{policy.policy_id}.json"
1245 dest.write_bytes(b"z" * (_MAX_POLICY_BYTES + 1))
1246 result = load_policy(repo, policy.policy_id)
1247 assert result is None
1248
1249 def test_malformed_json_pattern_returns_none(self, repo: pathlib.Path) -> None:
1250 pdir = h.patterns_dir(repo)
1251 pid = "a" * 64
1252 entry = pdir / 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 res_dir = h._resolutions_dir(repo, pattern.pattern_id)
1261 res_dir.mkdir(parents=True, exist_ok=True)
1262 rid = "b" * 64
1263 (res_dir / f"{rid}.json").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 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago