russell@unturf.com
04a2b27902
Mirrors `open_shards(paths) → Shards`: arborist-viz and other read-only
consumers import `open_sessions(path) → SessionsView`, exposing:
- counts() → SessionCounts(n_sessions, n_nodes, n_audit_events)
- list_sessions() → list[SessionRow]
- get_session(sid), get_node(bates), children_of, path_to_root
- all_nodes_in_session(sid), branches_in_session(sid)
- find(query, limit) — FTS5 over question+answer+cited_titles,
skips synthetic-root nodes (seq=0)
- find_by_cache_key(cache_key)
- chain_check() → (intact, breaks)
Opens sqlite with mode=ro + check_same_thread=False for WSGI worker
pools, serialized internally by an RLock. Bootstrap via
SessionStore.open().close() ensures the schema exists before the
read-only handle attaches.
9 tests pass: counts, list, get_node, path_to_root crossing sessions,
find FTS5, find-skips-synth-root, chain integrity, missing-db
fallback, branches in session, find_by_cache_key.
arborist-viz commit lands separately.
113 lines
3.3 KiB
Python
113 lines
3.3 KiB
Python
"""Tests for arborist.read.open_sessions / SessionsView."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from arborist.qa.session import SessionStore
|
|
from arborist.read import open_sessions
|
|
|
|
|
|
@pytest.fixture
|
|
def sessions_db(tmp_path: Path) -> Path:
|
|
db = tmp_path / "sessions.db"
|
|
store = SessionStore.open(db)
|
|
a = store.create_session(sid="A")
|
|
a.add_node("who is spider man?", "ck1", "STRICT",
|
|
answer_text="Spider-Man is from Marvel.\n "
|
|
'[E1 | Spider-Man | abc123: "..."]')
|
|
a.add_node("who created him?", "ck2", "STRICT",
|
|
answer_text="Stan Lee.\n "
|
|
'[E1 | Spider-Man | abc123: "..."]')
|
|
b = store.create_session(sid="B")
|
|
b.add_node("how do you cook risotto?", "ck3", "HYBRID",
|
|
answer_text="Stir constantly.")
|
|
store.close()
|
|
return db
|
|
|
|
|
|
def test_counts(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
c = v.counts()
|
|
assert c.n_sessions == 2
|
|
assert c.n_nodes == 3 # synthetic roots excluded
|
|
assert c.n_audit_events >= 5 # 2 init + 3 added
|
|
v.close()
|
|
|
|
|
|
def test_list_sessions(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
sids = {s.sid for s in v.list_sessions()}
|
|
assert sids == {"A", "B"}
|
|
a = v.get_session("A")
|
|
assert a is not None
|
|
assert a.n_nodes == 2
|
|
assert a.session_root # non-empty hex
|
|
v.close()
|
|
|
|
|
|
def test_get_node_and_path_to_root(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
n = v.get_node("A-000002")
|
|
assert n is not None
|
|
assert n.question == "who created him?"
|
|
assert n.cited_titles == ["Spider-Man"]
|
|
path = v.path_to_root("A-000002")
|
|
assert [p.bates for p in path] == ["A-000002", "A-000001", "A-000000"]
|
|
v.close()
|
|
|
|
|
|
def test_find_fts5_across_sessions(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
hits = v.find("spider")
|
|
assert {h.sid for h in hits} == {"A"}
|
|
hits = v.find("risotto")
|
|
assert {h.sid for h in hits} == {"B"}
|
|
v.close()
|
|
|
|
|
|
def test_find_skips_synthetic_root(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
hits = v.find("anything that wouldn't match")
|
|
assert hits == []
|
|
# Even targeting empty content the synth root never surfaces.
|
|
v.close()
|
|
|
|
|
|
def test_chain_check_intact(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
intact, breaks = v.chain_check()
|
|
assert breaks == 0
|
|
assert intact >= 5
|
|
v.close()
|
|
|
|
|
|
def test_missing_db_returns_empty_view(tmp_path: Path):
|
|
"""Open a path that doesn't exist yet — view is created (the
|
|
bootstrap inside SessionsView ensures the schema), but everything
|
|
reads back empty rather than erroring."""
|
|
db = tmp_path / "fresh.db"
|
|
v = open_sessions(db)
|
|
assert v.is_open
|
|
assert v.counts().n_sessions == 0
|
|
assert v.list_sessions() == []
|
|
assert v.find("anything") == []
|
|
assert v.chain_check() == (0, 0)
|
|
v.close()
|
|
|
|
|
|
def test_branches_in_session(sessions_db: Path):
|
|
"""Sanity: with two single-line sessions and no forks, branches=[]."""
|
|
v = open_sessions(sessions_db)
|
|
assert v.branches_in_session("A") == []
|
|
assert v.branches_in_session("B") == []
|
|
v.close()
|
|
|
|
|
|
def test_find_by_cache_key(sessions_db: Path):
|
|
v = open_sessions(sessions_db)
|
|
hits = v.find_by_cache_key("ck1")
|
|
assert len(hits) == 1
|
|
assert hits[0].sid == "A"
|
|
v.close()
|