96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
|
"""Contract tests: any BaseEventStore implementation must pass these."""
|
||
|
|
from app.infrastructure.perception.base_event_store import BaseEventStore
|
||
|
|
from app.infrastructure.perception.mock_event_store import MockEventStore
|
||
|
|
|
||
|
|
|
||
|
|
def _store() -> BaseEventStore:
|
||
|
|
return MockEventStore()
|
||
|
|
|
||
|
|
|
||
|
|
def test_is_base_event_store():
|
||
|
|
assert isinstance(_store(), BaseEventStore)
|
||
|
|
|
||
|
|
|
||
|
|
def test_all_returns_list():
|
||
|
|
result = _store().all()
|
||
|
|
assert isinstance(result, list)
|
||
|
|
assert len(result) > 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_known_id():
|
||
|
|
store = _store()
|
||
|
|
first = store.all()[0]
|
||
|
|
result = store.get(first["id"])
|
||
|
|
assert result is not None
|
||
|
|
assert result["id"] == first["id"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_unknown_returns_none():
|
||
|
|
assert _store().get("does-not-exist") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_filter_by_impact():
|
||
|
|
store = _store()
|
||
|
|
highs = store.filter(impact_level="high", limit=100)
|
||
|
|
assert all(e["impact_level"] == "high" for e in highs)
|
||
|
|
|
||
|
|
|
||
|
|
def test_filter_limit():
|
||
|
|
store = _store()
|
||
|
|
result = store.filter(limit=3)
|
||
|
|
assert len(result) <= 3
|
||
|
|
|
||
|
|
|
||
|
|
def test_stats_keys():
|
||
|
|
stats = _store().stats()
|
||
|
|
for key in ("total", "high_impact", "medium_impact", "recent_90d"):
|
||
|
|
assert key in stats, f"missing key: {key}"
|
||
|
|
|
||
|
|
|
||
|
|
def test_upsert_and_get():
|
||
|
|
store = _store()
|
||
|
|
event = {
|
||
|
|
"id": "test-upsert-001",
|
||
|
|
"source": "TEST",
|
||
|
|
"source_label": "Test Source",
|
||
|
|
"standard_code": "TST-001",
|
||
|
|
"title": "Test Event",
|
||
|
|
"summary": "A test event",
|
||
|
|
"full_text_url": "https://example.com",
|
||
|
|
"status": "draft",
|
||
|
|
"impact_level": "low",
|
||
|
|
"published_at": "2026-01-01",
|
||
|
|
"effective_at": None,
|
||
|
|
"category": "test",
|
||
|
|
"tags": ["test"],
|
||
|
|
"content_hash": "abc123",
|
||
|
|
"previous_hash": None,
|
||
|
|
}
|
||
|
|
store.upsert(event)
|
||
|
|
result = store.get("test-upsert-001")
|
||
|
|
assert result is not None
|
||
|
|
assert result["title"] == "Test Event"
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_by_standard_code():
|
||
|
|
store = _store()
|
||
|
|
first = store.all()[0]
|
||
|
|
result = store.get_by_standard_code(first["standard_code"])
|
||
|
|
assert result is not None
|
||
|
|
assert result["standard_code"] == first["standard_code"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_upsert_updates_existing():
|
||
|
|
store = _store()
|
||
|
|
first = store.all()[0]
|
||
|
|
original_id = first["id"]
|
||
|
|
store.upsert({"id": original_id, "title": "Updated Title", "impact_level": first["impact_level"],
|
||
|
|
"standard_code": first.get("standard_code", ""), "source": first["source"],
|
||
|
|
"source_label": first.get("source_label", ""), "summary": "Updated",
|
||
|
|
"full_text_url": "", "status": first["status"], "published_at": first.get("published_at", ""),
|
||
|
|
"effective_at": None, "category": first.get("category", ""), "tags": [],
|
||
|
|
"content_hash": "newhash", "previous_hash": None})
|
||
|
|
result = store.get(original_id)
|
||
|
|
assert result is not None
|
||
|
|
assert result["title"] == "Updated Title"
|