tests/test_db_migration.py

import importlib
import json
import os
import sqlite3
from pathlib import Path


def _legacy_database(path: Path) -> None:
    connection = sqlite3.connect(path)
    connection.executescript(
        """
        CREATE TABLE exchanges (
          id VARCHAR PRIMARY KEY, call_id VARCHAR NOT NULL, exchange_index INTEGER NOT NULL,
          year INTEGER, question_text TEXT NOT NULL, answer_text TEXT NOT NULL,
          audio_clip_filename VARCHAR NOT NULL, duration_s INTEGER,
          bootstrap_rasiah VARCHAR, flagged_broken BOOLEAN NOT NULL,
          created_at DATETIME NOT NULL
        );
        CREATE TABLE annotators (
          id INTEGER PRIMARY KEY AUTOINCREMENT, email VARCHAR UNIQUE NOT NULL,
          created_at DATETIME NOT NULL, current_exchange_id VARCHAR
        );
        CREATE TABLE annotations (
          id INTEGER PRIMARY KEY AUTOINCREMENT, exchange_id VARCHAR NOT NULL,
          annotator_id INTEGER NOT NULL, condition VARCHAR NOT NULL,
          idk BOOLEAN NOT NULL, flagged_broken BOOLEAN NOT NULL,
          flag_reason VARCHAR, rasiah VARCHAR, bavelas VARCHAR, bull VARCHAR,
          created_at DATETIME NOT NULL,
          UNIQUE(exchange_id, annotator_id, condition)
        );
        INSERT INTO exchanges VALUES
          ('old_1','old',1,2019,'Q','A','old.wav',2,'direct',0,'2026-01-01');
        INSERT INTO annotators VALUES
          (1,'legacy@example.com','2026-01-01',NULL);
        INSERT INTO annotations VALUES
          (1,'old_1',1,'text_audio',0,0,NULL,'direct',NULL,NULL,'2026-01-01');
        """
    )
    connection.commit()
    connection.close()


def test_additive_migration_preserves_legacy_and_claims_study_tasks(tmp_path: Path) -> None:
    database = tmp_path / "annotator.db"
    _legacy_database(database)
    os.environ["ANNOTATOR_DB_PATH"] = str(database)
    from app import db as db_module

    db = importlib.reload(db_module)
    db.init_db()
    session = db.get_session()
    try:
        assert session.query(db.Annotation).count() == 1
        package = Path(__file__).parents[1] / "data"
        seeded = db.seed_study(
            session,
            package / "study_manifest.json",
            package / "study_assignments.json",
        )
        assert seeded == {"items": 600, "tasks": 3600}
        annotator = db.enroll_annotator(session, "P-TEST", "some", "fluent")
        first_task, first_item = db.claim_task(session, annotator, "development")
        clean = {
            "gate": "clarification_repair",
            "gate_confidence": 90,
            "responsiveness": None,
            "rasiah": None,
            "supplied_information": None,
            "confidence": None,
            "delivery_descriptors_json": None,
            "audio_revision": None,
            "audible_event": None,
            "audible_event_other": None,
            "rationale": "The executive asks for clarification.",
            "response_time_ms": 20000,
            "audio_played_ms": 10000,
            "audio_completed": True,
        }
        assert db.complete_task(session, annotator, first_task.id, clean)
        second_task, second_item = db.claim_task(session, annotator, "development")
        assert second_item.id != first_item.id
        assert second_task.condition in {"text_only", "text_audio"}
        assert session.query(db.Annotation).count() == 2
    finally:
        session.close()