ontorefactor-governance/tests/test_migrations.py

import sqlite3

from ontorefactor_governance.db import Database


def test_existing_database_receives_llm_audit_columns(tmp_path):
    path = tmp_path / "pre-llm.db"
    connection = sqlite3.connect(path)
    connection.execute("""
        CREATE TABLE governance_agent_runs (
            id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, project_id TEXT NOT NULL,
            source_type TEXT NOT NULL, input_hash TEXT NOT NULL, status TEXT NOT NULL,
            result_json TEXT DEFAULT '{}', trace_json TEXT DEFAULT '[]', error TEXT DEFAULT '',
            created_at TEXT NOT NULL, completed_at TEXT
        )
    """)
    connection.commit()
    connection.close()

    db = Database(f"sqlite:///{path.as_posix()}")
    columns = {row["name"] for row in db.fetchall("PRAGMA table_info(governance_agent_runs)")}
    assert {
        "semantic_mode", "llm_provider", "llm_model", "prompt_version", "input_tokens",
        "output_tokens", "llm_latency_ms", "llm_request_hash", "llm_response_hash",
        "context_hash", "context_scope", "decision_id",
    } <= columns
    decision_tables = db.fetchall("SELECT name FROM sqlite_master WHERE type='table' AND name='governance_decisions'")
    assert decision_tables
    build_tables = {
        row["name"] for row in db.fetchall(
            "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('ontology_builds','ontology_versions','ontology_reports')"
        )
    }
    assert build_tables == {"ontology_builds", "ontology_versions", "ontology_reports"}
    db.close()