155 lines
5.8 KiB
Python
155 lines
5.8 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
|
|
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, resolve_principal
|
|
from tht.session.store import SessionError
|
|
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() == {}
|
|
|
|
|
|
@pytest.mark.parametrize("session_id", ["2026-01-01-000000-test", "2026-01-01-000000-x", "s1"])
|
|
def test_filesystem_repository_reads_safe_legacy_session_ids(tmp_path, session_id):
|
|
repository = FilesystemSessionRepository(
|
|
tmp_path / "home", "demo", PrincipalContext(issuer="local", subject="alice")
|
|
)
|
|
directory = repository.root / session_id
|
|
directory.mkdir(parents=True)
|
|
_manifest(session_id).to_yaml(directory / "session_manifest.yaml")
|
|
|
|
assert repository.get(session_id).manifest.id == session_id
|
|
|
|
|
|
@pytest.mark.parametrize("session_id", ["..", "a/b", "/absolute", "space id"])
|
|
def test_filesystem_repository_rejects_unsafe_legacy_session_ids(tmp_path, session_id):
|
|
repository = FilesystemSessionRepository(
|
|
tmp_path / "home", "demo", PrincipalContext(issuer="local", subject="alice")
|
|
)
|
|
|
|
with pytest.raises(SessionError):
|
|
repository.get(session_id)
|
|
|
|
|
|
def test_postgres_session_storage_requires_trusted_principal_environment(tmp_path, monkeypatch):
|
|
config = _config(tmp_path).model_copy(update={"session_storage": {
|
|
"type": "postgres_direct",
|
|
"connection": _db().model_dump(by_alias=True),
|
|
}})
|
|
monkeypatch.delenv("THT_PRINCIPAL_ISSUER", raising=False)
|
|
monkeypatch.delenv("THT_PRINCIPAL_SUBJECT", raising=False)
|
|
|
|
with pytest.raises(SessionError, match="THT_PRINCIPAL_ISSUER"):
|
|
resolve_principal(config)
|
|
|
|
|
|
def test_postgres_session_storage_uses_only_explicit_trusted_principal(tmp_path, monkeypatch):
|
|
config = _config(tmp_path).model_copy(update={"session_storage": {
|
|
"type": "postgres_direct",
|
|
"connection": _db().model_dump(by_alias=True),
|
|
}})
|
|
monkeypatch.setenv("THT_PRINCIPAL_ISSUER", "portal")
|
|
monkeypatch.setenv("THT_PRINCIPAL_SUBJECT", "alice")
|
|
monkeypatch.setenv("THT_PRINCIPAL_DISPLAY_NAME", "Alice")
|
|
monkeypatch.setenv("THT_PRINCIPAL_IS_ADMIN", "TRUE")
|
|
|
|
assert resolve_principal(config) == PrincipalContext(
|
|
issuer="portal", subject="alice", display_name="Alice", is_admin=True
|
|
)
|