"""Unit tests for the memory DDL — the only path that reaches an existing database.
There is no migration framework here: `MemoryStore._get_pool()` runs `_build_ddl()`
on every process start, and on a database that already has the table only the
`ADD COLUMN IF NOT EXISTS` statements do anything. A column that is declared in the
CREATE TABLE but forgotten in that block exists on a fresh database and nowhere else.
"""
import re
from navi.memory._ddl import _build_ddl
def _facts_table(ddl: list[str]) -> str:
for stmt in ddl:
block = re.search(r"CREATE TABLE IF NOT EXISTS memory_facts \((.*?)\n\s*\)", stmt, re.DOTALL)
if block:
return block.group(1)
raise AssertionError("memory_facts is no longer declared in _build_ddl")
def test_merged_into_is_migrated_and_declared():
for pgvector in (False, True):
ddl = _build_ddl(pgvector)
assert any(
"ALTER TABLE memory_facts ADD COLUMN IF NOT EXISTS merged_into" in stmt
for stmt in ddl
), f"The migration never adds merged_into (pgvector={pgvector})"
assert "merged_into" in _facts_table(ddl)
def test_no_index_on_the_migrated_column():
"""Almost every read filters `merged_into IS NULL`, so an index over it cannot be
selective — and one placed in the CREATE list would run before the ALTER that
adds the column, failing the whole batch on an existing database.
"""
for pgvector in (False, True):
for stmt in _build_ddl(pgvector):
if "CREATE INDEX" in stmt:
assert "merged_into" not in stmt, f"an index mentions merged_into: {stmt[:80]}"