Spaces:
Sleeping
Sleeping
File size: 3,391 Bytes
5b34fc5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | 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()
|