Create-knowledge-find-decode/tests/test_db_migration_contract.py

from __future__ import annotations

from pathlib import Path


MIGRATION = Path("db/migrations/001_creation_knowledge_schema.sql")
MIGRATIONS_DIR = Path("db/migrations")


def _migration_sql() -> str:
    return MIGRATION.read_text(encoding="utf-8")


def _all_migration_sql() -> str:
    return "\n".join(
        path.read_text(encoding="utf-8")
        for path in sorted(MIGRATIONS_DIR.glob("*.sql"))
    )


def test_creation_knowledge_migration_declares_formal_business_tables():
    sql = _migration_sql()
    expected_tables = [
        "query_batches",
        "queries",
        "acquisition_runs",
        "acquisition_jobs",
        "candidate_items",
        "media_assets",
        "item_classifications",
        "decode_jobs",
        "decode_results",
        "knowledge_particles",
        "scope_results",
        "payload_drafts",
        "ingest_records",
        "contract_snapshots",
    ]

    assert "CREATE SCHEMA IF NOT EXISTS creation_knowledge" in sql
    assert "CREATE EXTENSION IF NOT EXISTS pgcrypto" in sql
    for table in expected_tables:
        assert f"CREATE TABLE IF NOT EXISTS creation_knowledge.{table}" in sql


def test_migration_is_replayable_and_not_sqlite_shaped():
    sql = _migration_sql()

    assert "CREATE TABLE IF NOT EXISTS" in sql
    assert "CREATE INDEX IF NOT EXISTS" in sql
    assert "ON CONFLICT (version) DO UPDATE" in sql
    assert "CREATE OR REPLACE FUNCTION creation_knowledge.touch_updated_at" in sql
    assert "CREATE TRIGGER" in sql

    sqlite_only_tokens = ["AUTOINCREMENT", "PRAGMA", "sqlite_master", "WITHOUT ROWID"]
    assert not any(token in sql.upper() for token in sqlite_only_tokens)


def test_migration_keeps_state_rows_traceable_and_resumable():
    sql = _all_migration_sql()

    for table in [
        "acquisition_runs",
        "acquisition_jobs",
        "candidate_items",
        "decode_jobs",
        "decode_results",
        "payload_drafts",
        "ingest_records",
    ]:
        assert f"creation_knowledge.{table}" in sql
        assert "status text NOT NULL" in sql

    assert "attempt_count integer NOT NULL DEFAULT 0" in sql
    assert "error_message text" in sql
    assert "metadata jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
    assert "source_payload jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
    assert "response_payload jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
    assert "UNIQUE (run_id, query_id, platform)" in sql
    assert "idx_candidate_items_platform_item" in sql


def test_candidate_items_unique_key_migration_is_declared():
    sql = _all_migration_sql()

    assert "ADD COLUMN IF NOT EXISTS unique_key text" in sql
    assert "idx_candidate_items_unique_key" in sql
    assert "WHERE unique_key IS NOT NULL" in sql
    assert "'002_candidate_items_unique_key'" in sql


def test_candidate_items_content_mode_migration_is_declared():
    sql = _all_migration_sql()

    assert "ADD COLUMN IF NOT EXISTS content_mode text" in sql
    assert "idx_candidate_items_content_mode" in sql
    assert "video_post" in sql
    assert "image_post" in sql
    assert "article" in sql
    assert "unsupported" in sql
    assert "'003_candidate_items_content_mode'" in sql


def test_candidate_items_body_text_migration_is_declared():
    sql = _all_migration_sql()

    assert "ADD COLUMN IF NOT EXISTS body_text text" in sql
    assert "source_payload #>> '{detail,data,data,body_text}'" in sql
    assert "LEFT(NULLIF(source_payload #>> '{detail,data,data,body_text}', ''), 50000)" in sql
    assert "'004_candidate_items_body_text'" in sql