63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
|
"""find_for / covers tests (AGENTIC-WP-0010 follow-up)."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
|
||
|
|
from session_memory.curate.catalog import Catalog # noqa: E402
|
||
|
|
from session_memory.curate.schema import ( # noqa: E402
|
||
|
|
Provenance,
|
||
|
|
Resolution,
|
||
|
|
SolutionPattern,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _pattern(pid, src, covers=None, name="P"):
|
||
|
|
return SolutionPattern(
|
||
|
|
id=pid, name=name, version="1.0.0", polarity="problem", problem="p",
|
||
|
|
resolutions=[Resolution(summary="do x")],
|
||
|
|
provenance=Provenance(source_key=src), covers=covers or [])
|
||
|
|
|
||
|
|
|
||
|
|
def test_covers_round_trips(tmp_path):
|
||
|
|
cat = Catalog(str(tmp_path))
|
||
|
|
cat.upsert(_pattern("sp-a", "problem:file_not_read:edit",
|
||
|
|
covers=["file has not been read"]))
|
||
|
|
assert cat.load("sp-a").covers == ["file has not been read"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_find_for_exact_key(tmp_path):
|
||
|
|
cat = Catalog(str(tmp_path))
|
||
|
|
cat.upsert(_pattern(SolutionPattern.make_id("problem:retry_storm:retries"),
|
||
|
|
"problem:retry_storm:retries"))
|
||
|
|
got = cat.find_for("problem:retry_storm:retries")
|
||
|
|
assert got is not None and got.id == "sp-problem-retry_storm-retries"
|
||
|
|
|
||
|
|
|
||
|
|
def test_find_for_covers_match(tmp_path):
|
||
|
|
cat = Catalog(str(tmp_path))
|
||
|
|
cat.upsert(_pattern("sp-rbe", "problem:file_not_read:edit",
|
||
|
|
covers=["file has not been read", "modified since read"]))
|
||
|
|
# a recurring_error signal with a different key but matching fingerprint locus
|
||
|
|
got = cat.find_for(
|
||
|
|
"problem:recurring_error:<tool_use_error>file has not been read yet...",
|
||
|
|
locus="<tool_use_error>file has not been read yet. read it first...")
|
||
|
|
assert got is not None and got.id == "sp-rbe"
|
||
|
|
|
||
|
|
|
||
|
|
def test_find_for_no_match_returns_none(tmp_path):
|
||
|
|
cat = Catalog(str(tmp_path))
|
||
|
|
cat.upsert(_pattern("sp-rbe", "problem:file_not_read:edit",
|
||
|
|
covers=["file has not been read"]))
|
||
|
|
assert cat.find_for("problem:recurring_error:some unrelated error") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_covers_change_versions(tmp_path):
|
||
|
|
cat = Catalog(str(tmp_path))
|
||
|
|
cat.upsert(_pattern("sp-a", "problem:x:y"))
|
||
|
|
p = cat.load("sp-a")
|
||
|
|
p.covers = ["new coverage"]
|
||
|
|
assert cat.upsert(p) == "versioned" # covers is substantive content
|
||
|
|
assert cat.load("sp-a").version == "1.0.1"
|