103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
import uuid
|
|
|
|
from tht.config import DatabaseConfig, load_config
|
|
from tht.decisions import DecisionInput
|
|
from tht.session.filesystem_repository import FilesystemSessionRepository
|
|
from tht.session.models import PrincipalContext, SessionManifest
|
|
from tht.session.repository import build_session_repository
|
|
from tht.session.store import create_session
|
|
|
|
|
|
def _db() -> DatabaseConfig:
|
|
return DatabaseConfig(database="testdb", schema="public", user="u", password="p") # noqa: S106
|
|
|
|
|
|
def _config(tmp_path):
|
|
path = tmp_path / "workspace.yaml"
|
|
path.write_text(
|
|
"dwh:\n"
|
|
" type: postgres_direct\n"
|
|
" connection:\n"
|
|
" database: testdb\n"
|
|
" schema: public\n"
|
|
" user: user\n"
|
|
" password: secret\n"
|
|
)
|
|
return load_config(path)
|
|
|
|
|
|
def _manifest(session_id: str) -> SessionManifest:
|
|
return SessionManifest(
|
|
id=session_id,
|
|
created_at="2026-07-16T10:00:00Z",
|
|
question="Which patients had an ablation?",
|
|
database="testdb",
|
|
schema="public",
|
|
)
|
|
|
|
|
|
def test_new_sessions_use_uuid4_ids(tmp_path):
|
|
manifest = create_session("Which patients had an ablation?", _db(), tmp_path)
|
|
|
|
session_uuid = uuid.UUID(manifest.id, version=4)
|
|
assert str(session_uuid) == manifest.id
|
|
assert session_uuid.version == 4
|
|
|
|
|
|
def test_filesystem_repository_scopes_sessions_to_local_principal_root(tmp_path):
|
|
config = _config(tmp_path)
|
|
principal = PrincipalContext(issuer="local", subject="alice")
|
|
repository = build_session_repository(config, principal, home=tmp_path / "alice-home")
|
|
session_id = str(uuid.uuid4())
|
|
|
|
repository.create(_manifest(session_id))
|
|
|
|
assert repository.root == tmp_path / "alice-home/workspaces/workspace/sessions"
|
|
assert (repository.root / session_id / "session_manifest.yaml").exists()
|
|
|
|
|
|
def test_filesystem_repository_reads_phase_artifacts_and_appends_decisions(tmp_path):
|
|
repository = FilesystemSessionRepository(
|
|
tmp_path / "home", "demo", PrincipalContext(issuer="local", subject="alice")
|
|
)
|
|
session_id = str(uuid.uuid4())
|
|
repository.create(_manifest(session_id))
|
|
|
|
repository.write_artifact(session_id, "sql_final", "SELECT 1")
|
|
repository.write_artifact(session_id, "evidence", '{"sources": []}')
|
|
decisions = repository.append_decisions(
|
|
session_id,
|
|
[DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT 1")],
|
|
)
|
|
snapshot = repository.get(session_id)
|
|
|
|
assert repository.read_artifact(session_id, "sql_final") == "SELECT 1"
|
|
assert repository.read_artifact(session_id, "evidence") == '{"sources": []}'
|
|
assert snapshot.artifacts["sql_final"] == "SELECT 1"
|
|
assert snapshot.artifacts["evidence"] == '{"sources": []}'
|
|
assert [decision.type for decision in snapshot.decisions] == ["sql_approved"]
|
|
assert decisions[0].seq == 1
|
|
|
|
|
|
def test_tht_home_overrides_local_repository_root(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("THT_HOME", str(tmp_path / "override"))
|
|
config = _config(tmp_path)
|
|
|
|
repository = build_session_repository(
|
|
config, PrincipalContext(issuer="local", subject="alice")
|
|
)
|
|
|
|
assert config.paths.sessions == tmp_path / "override/workspaces/workspace/sessions"
|
|
assert repository.root == tmp_path / "override/workspaces/workspace/sessions"
|
|
|
|
|
|
def test_filesystem_repository_keeps_preferences_per_principal(tmp_path):
|
|
home = tmp_path / "home"
|
|
alice = FilesystemSessionRepository(home, "demo", PrincipalContext(issuer="local", subject="alice"))
|
|
bob = FilesystemSessionRepository(home, "demo", PrincipalContext(issuer="local", subject="bob"))
|
|
|
|
alice.set_preferences({"model": "glm"})
|
|
|
|
assert alice.get_preferences() == {"model": "glm"}
|
|
assert bob.get_preferences() == {}
|