232 lines
9.0 KiB
Python
232 lines
9.0 KiB
Python
"""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
|