breathe-memory/tests/test_anchor_extractor.py
mvyshhnyvetska 4e671a0a83 Initial release: breathe-memory v0.1.0
Context optimization and associative memory for LLM applications.
Two-phase system: SYNAPSE (pre-generation memory injection) +
GraphCompactor (structured context compression).

- Interface-based, storage-agnostic, LLM-agnostic
- Memory Nexus: PostgreSQL + pgvector reference backend
- Zero mandatory dependencies beyond stdlib
- 28 tests passing, clean install verified

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-26 13:50:00 +01:00

69 lines
2.3 KiB
Python

"""Tests for AnchorExtractor."""
import pytest
from breathe.anchor_extractor import AnchorExtractor, AnchorResult
def make_extractor(concepts=None):
return AnchorExtractor(known_concepts=concepts or {})
def test_known_concept_match():
extractor = make_extractor({"FastAPI": "uuid-001", "Redis": "uuid-002"})
result = extractor.extract("How do I add caching to FastAPI?")
entity_texts = [a.text.lower() for a in result.entity_anchors]
assert "fastapi" in entity_texts
def test_known_concept_sets_node_id():
extractor = make_extractor({"FastAPI": "uuid-001"})
result = extractor.extract("I am working on FastAPI")
matched = [a for a in result.anchors if a.matched_node_id]
assert len(matched) >= 1
assert matched[0].matched_node_id == "uuid-001"
def test_temporal_en():
extractor = make_extractor()
result = extractor.extract("remember when we fixed the bug yesterday?")
temporal = [a for a in result.anchors if a.anchor_type == "temporal"]
assert len(temporal) >= 1
def test_technical_anchor():
extractor = make_extractor()
result = extractor.extract("check the routes.py file for the API handler")
tech = [a for a in result.anchors if a.anchor_type == "technical"]
assert any("routes.py" in a.text for a in tech)
def test_deduplication():
extractor = make_extractor({"FastAPI": "uuid-001"})
# FastAPI appears twice — should deduplicate
result = extractor.extract("FastAPI is great, I love FastAPI")
fastapi_anchors = [a for a in result.anchors if "fastapi" in a.text.lower()]
assert len(fastapi_anchors) == 1
def test_conversation_mode_work():
extractor = make_extractor()
result = extractor.extract("class MyModel: def __init__(): import asyncpg from FastAPI")
assert result.conversation_mode == "work"
def test_conversation_mode_casual():
extractor = make_extractor()
result = extractor.extract("hey")
assert result.conversation_mode == "casual"
def test_empty_message():
extractor = make_extractor()
result = extractor.extract("")
assert isinstance(result, AnchorResult)
assert len(result.anchors) == 0
def test_node_ids_property():
extractor = make_extractor({"Redis": "uuid-redis"})
result = extractor.extract("I need Redis for caching")
assert "uuid-redis" in result.node_ids