"""Acceptance tests for the Phase-2 migration script (Task 5). Verifies: 1. The migration creates ``maf_roundtable_seat_runs`` + ``source_event_id`` column + the unique index + ``maf_roundtable_action_secrets`` on SQLite. 2. Running the migration twice is idempotent (no error, no duplicate objects). 3. Phase-1 data (an existing event row) remains readable after migration. """ from __future__ import annotations import importlib.util from pathlib import Path import pytest from sqlalchemy import inspect as sa_inspect from sqlalchemy import text from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from deerflow.persistence.maf_roundtable import make_maf_roundtable_store from deerflow.persistence.maf_roundtable.model import Base def _load_migration_module(): """Load scripts/migrate_maf_roundtable_phase2.py as a module.""" here = Path(__file__).resolve().parent.parent script = here / "scripts" / "migrate_maf_roundtable_phase2.py" spec = importlib.util.spec_from_file_location("migrate_maf_roundtable_phase2", script) assert spec and spec.loader, "could not load migration script" mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod @pytest.mark.asyncio async def test_migration_creates_phase2_objects_and_is_idempotent(): """Run the migration on a fresh Phase-1 schema, then re-run it; both must succeed and produce the expected Phase-2 objects.""" mod = _load_migration_module() engine = create_async_engine("sqlite+aiosqlite://", echo=False) # Build the Phase-1 schema (everything except the Phase-2 additions). # We create the base tables the migration expects to already exist. async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # Drop the Phase-2 tables if create_all created them, so we test creation. await conn.execute(text("DROP TABLE IF EXISTS maf_roundtable_seat_runs")) await conn.execute(text("DROP TABLE IF EXISTS maf_roundtable_action_secrets")) # Seed a Phase-1 event row so we can verify it survives migration. # Use the ORM (not raw SQL) so column Python-defaults apply. from datetime import datetime from deerflow.persistence.maf_roundtable.model import ( MafRoundtableEventRow, MafRoundtableRunRow, MafRoundtableSessionRow, ) sf = async_sessionmaker(engine, expire_on_commit=False) now = datetime(2026, 1, 1) async with sf() as session: session.add(MafRoundtableSessionRow( id="s1", user_id="u1", title="t", task_text="t", mode="structured", status="draft", write_policy="confirm_high_risk", max_rounds=8, created_at=now, updated_at=now, )) session.add(MafRoundtableRunRow( id="r1", session_id="s1", status="pending", created_at=now, updated_at=now, )) session.add(MafRoundtableEventRow( id="e1", session_id="s1", run_id="r1", seq=1, event_type="run.started", created_at=now, )) await session.commit() # Run the migration's sync step directly on a connection (dry_run=False). async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=False, logger=mod.logger, changes_ref=[0]) # Verify all Phase-2 objects exist. def _check(sync_conn): insp = sa_inspect(sync_conn) tables = set(insp.get_table_names()) assert "maf_roundtable_seat_runs" in tables assert "maf_roundtable_action_secrets" in tables event_cols = {c["name"] for c in insp.get_columns("maf_roundtable_events")} assert "source_event_id" in event_cols # Unique indexes. idx_names: set[str] = set() for t in insp.get_table_names(): for idx in insp.get_indexes(t): idx_names.add(idx.get("name")) assert "uq_maf_events_run_source" in idx_names assert "uq_maf_seat_runs_run_agent_turn" in idx_names assert "uq_maf_action_secrets_action" in idx_names async with engine.begin() as conn: await conn.run_sync(_check) # Phase-1 data still readable + the new column is nullable. async with sf() as session: rows = (await session.execute(text("SELECT id, source_event_id FROM maf_roundtable_events"))).all() assert len(rows) == 1 assert rows[0][0] == "e1" assert rows[0][1] is None # old row keeps NULL # Re-run the migration ??must be idempotent (no error). async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=False, logger=mod.logger, changes_ref=[0]) # Dry-run must also not throw and must not write. async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=True, logger=mod.logger, changes_ref=[0]) await engine.dispose() @pytest.mark.asyncio async def test_migration_adds_turn_index_to_legacy_actions_table(): """Phase-1 actions table without turn_index is upgraded idempotently.""" mod = _load_migration_module() engine = create_async_engine("sqlite+aiosqlite://", echo=False) async with engine.begin() as conn: await conn.execute( text( """ CREATE TABLE maf_roundtable_actions ( id VARCHAR(64) NOT NULL PRIMARY KEY, session_id VARCHAR(64) NOT NULL, run_id VARCHAR(64), agent_id VARCHAR(64), tool_name VARCHAR(128) NOT NULL, logical_action_id VARCHAR(128), idempotency_key VARCHAR(191) NOT NULL UNIQUE, target_key VARCHAR(255), risk_level VARCHAR(8) NOT NULL, status VARCHAR(32) NOT NULL, arguments_redacted_json TEXT, result_redacted_json TEXT, external_object_id VARCHAR(255), approval_id VARCHAR(64), attempt_count INTEGER NOT NULL, started_at DATETIME, finished_at DATETIME, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ) """ ) ) async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=False, logger=mod.logger, changes_ref=[0]) def _has_turn_index(sync_conn): insp = sa_inspect(sync_conn) cols = {c["name"] for c in insp.get_columns("maf_roundtable_actions")} assert "turn_index" in cols async with engine.begin() as conn: await conn.run_sync(_has_turn_index) sf = async_sessionmaker(engine, expire_on_commit=False) store = make_maf_roundtable_store(sf) assert store is not None from datetime import datetime from deerflow.persistence.maf_roundtable.model import MafRoundtableSessionRow async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all, tables=[MafRoundtableSessionRow.__table__]) now = datetime(2026, 1, 1) async with sf() as session: session.add( MafRoundtableSessionRow( id="s1", user_id="u1", title="t", task_text="t", mode="structured", status="draft", write_policy="confirm_high_risk", max_rounds=8, created_at=now, updated_at=now, ) ) await session.commit() row = await store.actions.create_action( { "session_id": "s1", "tool_name": "write_file", "idempotency_key": "legacy-1", "turn_index": 1, "risk_level": "R3", "status": "pending", } ) assert row["turn_index"] == 1 async with engine.begin() as conn: await conn.run_sync(mod._do_migrate_sync, dry_run=False, logger=mod.logger, changes_ref=[0]) await engine.dispose() def test_ddl_add_action_turn_index_by_dialect(): mod = _load_migration_module() assert "INTEGER NULL" in mod._ddl_add_action_turn_index("sqlite") assert "INTEGER NULL" in mod._ddl_add_action_turn_index("postgresql") mysql = mod._ddl_add_action_turn_index("mysql") assert "INT NULL" in mysql assert "INTEGER" not in mysql.replace("INT NULL", "") def test_ddl_lists_are_single_statement(): """The DDL constants are lists of single SQL statements (no ``;`` joining).""" mod = _load_migration_module() for stmt in mod._DDL_SEAT_RUNS + mod._DDL_ACTION_SECRETS: # Each statement must not contain a statement separator mid-string # (trailing/leading whitespace stripped). A CREATE TABLE followed by # a semicolon + another CREATE is the bug we fixed. stripped = stmt.strip().rstrip(";").strip() assert stripped.count(";") == 0, f"DDL statement contains ';': {stmt[:80]!r}" # The events index is a single statement string. assert mod._DDL_EVENTS_SOURCE_INDEX.count(";") <= 1