54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
|
"""
|
||
|
|
Test helpers — populate the in-memory cache directly without touching Postgres.
|
||
|
|
|
||
|
|
All collision detection, recollection rendering, and queue routing logic operates
|
||
|
|
entirely on the module-level dicts in festinger.cache, so unit tests can inject
|
||
|
|
state there directly and assert outcomes without a live database.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from festinger import cache
|
||
|
|
from festinger.cache import SoasRow, UrdEdge
|
||
|
|
|
||
|
|
|
||
|
|
def reset_cache() -> None:
|
||
|
|
"""Clear all in-memory state. Call at the start of each test."""
|
||
|
|
cache.soas_by_token.clear()
|
||
|
|
cache.soas_by_id.clear()
|
||
|
|
cache.urd_by_concept.clear()
|
||
|
|
cache.urd_by_concept_dim.clear()
|
||
|
|
cache.pending_conflicts.clear()
|
||
|
|
cache._encounter_deltas.clear()
|
||
|
|
|
||
|
|
|
||
|
|
def add_soas(id: int, token: str, saliency: float = 0.0, novelty: float = 0.0) -> SoasRow:
|
||
|
|
row = SoasRow(id=id, token=token, saliency=saliency, novelty=novelty)
|
||
|
|
cache.soas_by_token[token] = row
|
||
|
|
cache.soas_by_id[id] = token
|
||
|
|
return row
|
||
|
|
|
||
|
|
|
||
|
|
def add_urd(
|
||
|
|
concept_id: int,
|
||
|
|
parent_id: int,
|
||
|
|
dim_id: int,
|
||
|
|
is_isa: bool,
|
||
|
|
confidence: float = 0.9,
|
||
|
|
source: str = "test",
|
||
|
|
) -> UrdEdge:
|
||
|
|
parent_token = cache.soas_by_id.get(parent_id, str(parent_id))
|
||
|
|
dim_token = cache.soas_by_id.get(dim_id, str(dim_id))
|
||
|
|
edge = UrdEdge(
|
||
|
|
concept_id=concept_id,
|
||
|
|
parent_id=parent_id,
|
||
|
|
dim_id=dim_id,
|
||
|
|
is_isa=is_isa,
|
||
|
|
confidence=confidence,
|
||
|
|
source=source,
|
||
|
|
parent_token=parent_token,
|
||
|
|
dim_token=dim_token,
|
||
|
|
)
|
||
|
|
cache.urd_by_concept.setdefault(concept_id, []).append(edge)
|
||
|
|
cache.urd_by_concept_dim[(concept_id, dim_id)] = edge
|
||
|
|
return edge
|