feat(agent-office): tarot_readings 1회성 마이그레이션 스크립트 (3 테스트)

agent-office/scripts/migrate_tarot_to_lab.py

@@ -0,0 +1,81 @@
+"""1회성 마이그레이션 — agent_office.db.tarot_readings → tarot.db.tarot_readings.
+
+멱등성: 이미 존재하는 id는 SKIP.
+
+실행:
+    docker exec agent-office python /app/scripts/migrate_tarot_to_lab.py
+
+또는 호스트에서 직접:
+    AGENT_OFFICE_DB=/path/to/agent_office.db TAROT_DB=/path/to/tarot.db \\
+        python scripts/migrate_tarot_to_lab.py
+"""
+import os
+import sqlite3
+import sys
+
+
+SRC = os.getenv("AGENT_OFFICE_DB", "/app/data/agent_office.db")
+DST = os.getenv("TAROT_DB", "/app/data/tarot.db")
+
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS tarot_readings (
+    id                  INTEGER PRIMARY KEY AUTOINCREMENT,
+    created_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+    spread_type         TEXT NOT NULL,
+    category            TEXT,
+    question            TEXT,
+    cards               TEXT NOT NULL,
+    interpretation_json TEXT,
+    summary             TEXT,
+    model               TEXT,
+    tokens_in           INTEGER,
+    tokens_out          INTEGER,
+    cost_usd            REAL,
+    confidence          TEXT,
+    favorite            INTEGER NOT NULL DEFAULT 0,
+    note                TEXT
+);
+"""
+
+
+def migrate() -> int:
+    """이관된 row 수 반환."""
+    src = sqlite3.connect(SRC)
+    src.row_factory = sqlite3.Row
+    dst = sqlite3.connect(DST)
+    dst.execute("PRAGMA journal_mode=WAL")
+    dst.executescript(SCHEMA)
+
+    rows = src.execute("SELECT * FROM tarot_readings").fetchall()
+    if not rows:
+        src.close(); dst.close()
+        return 0
+
+    all_cols = list(rows[0].keys())
+
+    moved = 0
+    for r in rows:
+        exists = dst.execute("SELECT 1 FROM tarot_readings WHERE id=?", (r["id"],)).fetchone()
+        if exists:
+            continue
+        # NULL 값은 INSERT에서 제외 → 목적지 스키마의 DEFAULT가 적용되도록 함
+        # (예: created_at이 NULL이면 strftime() 기본값 사용)
+        cols = [c for c in all_cols if r[c] is not None]
+        placeholders = ",".join("?" * len(cols))
+        cols_str = ",".join(cols)
+        dst.execute(
+            f"INSERT INTO tarot_readings ({cols_str}) VALUES ({placeholders})",
+            tuple(r[c] for c in cols),
+        )
+        moved += 1
+    dst.commit()
+    src.close(); dst.close()
+    return moved
+
+
+if __name__ == "__main__":
+    moved = migrate()
+    total = sqlite3.connect(SRC).execute("SELECT COUNT(*) FROM tarot_readings").fetchone()[0]
+    print(f"migrated {moved} / {total} rows from {SRC} to {DST}")
+    sys.exit(0)

agent-office/tests/test_migrate_tarot.py

@@ -0,0 +1,72 @@
+"""migrate_tarot_to_lab.py 단위 테스트 — 멱등성 + 데이터 보존."""
+import sqlite3
+import sys
+import os
+import pytest
+
+
+@pytest.fixture
+def src_db(tmp_path):
+    p = tmp_path / "agent_office.db"
+    conn = sqlite3.connect(str(p))
+    conn.execute("""
+        CREATE TABLE tarot_readings (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            created_at TEXT, spread_type TEXT, category TEXT, question TEXT,
+            cards TEXT, interpretation_json TEXT, summary TEXT, model TEXT,
+            tokens_in INTEGER, tokens_out INTEGER, cost_usd REAL,
+            confidence TEXT, favorite INTEGER, note TEXT
+        )
+    """)
+    conn.execute("""
+        INSERT INTO tarot_readings (id, spread_type, category, cards, model, favorite)
+        VALUES (1, 'three_card', '연애', '[]', 'm', 0),
+               (2, 'one_card', '재물', '[]', 'm', 1)
+    """)
+    conn.commit()
+    conn.close()
+    return str(p)
+
+
+@pytest.fixture
+def dst_db(tmp_path):
+    return str(tmp_path / "tarot.db")
+
+
+def _import_migrate(src, dst, monkeypatch):
+    monkeypatch.setenv("AGENT_OFFICE_DB", src)
+    monkeypatch.setenv("TAROT_DB", dst)
+    sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
+    import migrate_tarot_to_lab as m
+    import importlib
+    importlib.reload(m)
+    return m
+
+
+def test_first_run_copies_all_rows(src_db, dst_db, monkeypatch):
+    m = _import_migrate(src_db, dst_db, monkeypatch)
+    moved = m.migrate()
+    assert moved == 2
+    conn = sqlite3.connect(dst_db)
+    rows = conn.execute("SELECT id, spread_type, category FROM tarot_readings ORDER BY id").fetchall()
+    conn.close()
+    assert rows == [(1, "three_card", "연애"), (2, "one_card", "재물")]
+
+
+def test_idempotent_second_run(src_db, dst_db, monkeypatch):
+    m = _import_migrate(src_db, dst_db, monkeypatch)
+    m.migrate()
+    moved2 = m.migrate()
+    assert moved2 == 0
+
+
+def test_partial_migration(src_db, dst_db, monkeypatch):
+    """dst에 id=1만 있는 상태에서 다시 돌리면 id=2만 옮김."""
+    m = _import_migrate(src_db, dst_db, monkeypatch)
+    m.migrate()
+    conn = sqlite3.connect(dst_db)
+    conn.execute("DELETE FROM tarot_readings WHERE id=2")
+    conn.commit()
+    conn.close()
+    moved = m.migrate()
+    assert moved == 1