ThothII/harness/tests/test_postgres_session_repository.py

import uuid
from datetime import UTC, datetime

import pytest
from sqlalchemy import create_engine, text
from testcontainers.postgres import PostgresContainer

from tht.decisions import DecisionInput
from tht.session.models import PrincipalContext, SessionManifest
from tht.session.store import SessionError


@pytest.fixture(scope="module")
def database_url():
    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres.get_connection_url()


@pytest.fixture(scope="module", autouse=True)
def migrated(database_url):
    from tht.session.postgres_repository import migrate

    migrate(database_url)


def _manifest(session_id: str) -> SessionManifest:
    return SessionManifest(
        id=session_id,
        created_at=datetime(2026, 7, 16, 10, 0, tzinfo=UTC),
        question="Which patients had an ablation?",
        database="testdb",
        schema="public",
    )


def _repository(database_url, subject: str, *, is_admin: bool = False):
    from tht.session.postgres_repository import PostgresSessionRepository

    return PostgresSessionRepository(
        database_url,
        PrincipalContext(issuer="portal", subject=subject, is_admin=is_admin),
    )


def test_owner_can_read_own_snapshot_but_not_another_owners(database_url):
    alice = _repository(database_url, "alice")
    bob = _repository(database_url, "bob")
    session_id = str(uuid.uuid4())

    alice.create(_manifest(session_id))
    alice.write_artifact(session_id, "cte_sql:eligible_patients", "SELECT 1")
    alice.append_decisions(
        session_id, [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT 1")]
    )

    snapshot = alice.get(session_id)
    assert snapshot.principal == alice.principal
    assert snapshot.artifacts == {"cte_sql:eligible_patients": "SELECT 1"}
    assert snapshot.decisions[0].seq == 1
    with pytest.raises(SessionError, match="Sessione non trovata"):
        bob.get(session_id)


def test_admin_can_read_another_owners_session(database_url):
    owner = _repository(database_url, "owner")
    admin = _repository(database_url, "admin", is_admin=True)
    session_id = str(uuid.uuid4())
    owner.create(_manifest(session_id))

    assert admin.get(session_id).manifest.id == session_id


def test_non_superuser_runtime_login_can_assume_the_restricted_runtime_role(database_url):
    admin = create_engine(database_url)
    runtime_url = admin.url.set(
        username="thoth_sessions_test_login", password="runtime-test-only"
    )
    try:
        with admin.begin() as connection:
            connection.exec_driver_sql(
                "CREATE ROLE thoth_sessions_test_login LOGIN NOINHERIT PASSWORD 'runtime-test-only'"
            )
            connection.exec_driver_sql("GRANT thoth_sessions_runtime TO thoth_sessions_test_login")
        repository = _repository(
            runtime_url.render_as_string(hide_password=False), "runtime-user"
        )
        session_id = str(uuid.uuid4())

        repository.create(_manifest(session_id))

        assert repository.get(session_id).manifest.id == session_id
        repository.close()
    finally:
        with admin.begin() as connection:
            connection.exec_driver_sql("DROP ROLE IF EXISTS thoth_sessions_test_login")
        admin.dispose()


def test_delete_cascades_content_and_leaves_content_free_tombstone(database_url):
    repository = _repository(database_url, "alice")
    session_id = str(uuid.uuid4())
    repository.create(_manifest(session_id))
    repository.write_artifact(session_id, "sql_final", "SELECT confidential_value")
    repository.append_decisions(
        session_id,
        [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT confidential_value")],
    )

    repository.delete(session_id)

    with pytest.raises(SessionError, match="Sessione non trovata"):
        repository.get(session_id)
    engine = create_engine(database_url)
    try:
        with engine.connect() as connection:
            assert connection.execute(
                text("SELECT count(*) FROM thoth_sessions.session_artifacts WHERE session_id = :session_id"),
                {"session_id": session_id},
            ).scalar_one() == 0
            assert connection.execute(
                text("SELECT count(*) FROM thoth_sessions.review_decisions WHERE session_id = :session_id"),
                {"session_id": session_id},
            ).scalar_one() == 0
            columns = connection.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_schema = 'thoth_sessions' AND table_name = 'audit_log'"
                )
            ).scalars().all()
            tombstone = connection.execute(
                text(
                    "SELECT action, session_id, actor_issuer, actor_subject "
                    "FROM thoth_sessions.audit_log WHERE session_id = :session_id"
                ),
                {"session_id": session_id},
            ).one()
    finally:
        engine.dispose()
    assert (tombstone[0], str(tombstone[1]), *tombstone[2:]) == (
        "session_deleted",
        session_id,
        "portal",
        "alice",
    )
    assert not {"content", "artifact_content", "detail", "metadata"} & set(columns)


def test_session_schema_does_not_create_or_invoke_embeddings(database_url, monkeypatch):
    import tht.session.postgres_repository as repository_module

    monkeypatch.setattr(
        repository_module,
        "_embed",
        lambda *_: pytest.fail("session persistence must not invoke embeddings"),
        raising=False,
    )
    repository = _repository(database_url, "alice")
    session_id = str(uuid.uuid4())
    repository.create(_manifest(session_id))
    repository.write_artifact(session_id, "evidence", '{"sources": []}')

    engine = create_engine(database_url)
    try:
        with engine.connect() as connection:
            columns = connection.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_schema = 'thoth_sessions'"
                )
            ).scalars().all()
    finally:
        engine.dispose()
    assert all("embedding" not in column for column in columns)