200 lines
8.0 KiB
Python
200 lines
8.0 KiB
Python
"""Acceptance coverage for the additive Phase-5 target-lease migration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect, text
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from deerflow.persistence.maf_roundtable.model import Base
|
|
|
|
|
|
def _migration_module():
|
|
script = Path(__file__).resolve().parent.parent / "scripts" / "migrate_maf_roundtable_phase5.py"
|
|
spec = importlib.util.spec_from_file_location("migrate_maf_roundtable_phase5", script)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _assert_complete_lease_schema(sync_conn, module) -> None:
|
|
inspector = inspect(sync_conn)
|
|
assert module.validate_lease_table_schema(inspector) == []
|
|
columns = {column["name"]: column for column in inspector.get_columns(module.TABLE)}
|
|
assert all(columns[name]["nullable"] is False for name in columns)
|
|
assert inspector.get_pk_constraint(module.TABLE)["constrained_columns"] == ["target_key"]
|
|
assert any(
|
|
constraint["column_names"] == ["action_id"]
|
|
for constraint in inspector.get_unique_constraints(module.TABLE)
|
|
)
|
|
assert any(
|
|
foreign_key["constrained_columns"] == ["action_id"]
|
|
and foreign_key["referred_table"] == "maf_roundtable_actions"
|
|
and foreign_key["referred_columns"] == ["id"]
|
|
and str((foreign_key.get("options") or {}).get("ondelete")).upper() == "CASCADE"
|
|
for foreign_key in inspector.get_foreign_keys(module.TABLE)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase5_lease_migration_is_additive_and_idempotent():
|
|
module = _migration_module()
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
await conn.execute(text("DROP TABLE maf_roundtable_action_leases"))
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(module._migrate_sync, dry_run=False)
|
|
|
|
def check(sync_conn):
|
|
inspector = inspect(sync_conn)
|
|
assert "maf_roundtable_action_leases" in inspector.get_table_names()
|
|
columns = {column["name"] for column in inspector.get_columns("maf_roundtable_action_leases")}
|
|
assert {"target_key", "action_id", "lease_expires_at"} <= columns
|
|
indexes = {index["name"] for index in inspector.get_indexes("maf_roundtable_action_leases")}
|
|
assert {"ix_maf_action_leases_action_id", "ix_maf_action_leases_lease_expires_at"} <= indexes
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(check)
|
|
await conn.execute(text(f"DROP INDEX {module.ACTION_INDEX}"))
|
|
await conn.execute(text(f"DROP INDEX {module.EXPIRY_INDEX}"))
|
|
await conn.run_sync(module._migrate_sync, dry_run=False)
|
|
await conn.run_sync(check)
|
|
await conn.run_sync(module._migrate_sync, dry_run=True)
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_incomplete_lease_table_is_rebuilt_with_full_constraints():
|
|
module = _migration_module()
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
await conn.execute(text(f"DROP TABLE {module.TABLE}"))
|
|
await conn.execute(text(f"""
|
|
CREATE TABLE {module.TABLE} (
|
|
target_key VARCHAR(255),
|
|
action_id VARCHAR(64),
|
|
lease_expires_at DATETIME,
|
|
created_at DATETIME,
|
|
updated_at DATETIME
|
|
)
|
|
"""))
|
|
await conn.run_sync(module._migrate_sync, dry_run=False)
|
|
await conn.run_sync(_assert_complete_lease_schema, module)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonempty_incomplete_lease_table_fails_without_rewriting_data():
|
|
module = _migration_module()
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
await conn.execute(text(f"DROP TABLE {module.TABLE}"))
|
|
await conn.execute(text(f"""
|
|
CREATE TABLE {module.TABLE} (
|
|
target_key VARCHAR(255),
|
|
action_id VARCHAR(64),
|
|
lease_expires_at DATETIME,
|
|
created_at DATETIME,
|
|
updated_at DATETIME
|
|
)
|
|
"""))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {module.TABLE} "
|
|
"(target_key, action_id, lease_expires_at, created_at, updated_at) "
|
|
"VALUES ('opaque-1', 'action-1', '2026-01-01', '2026-01-01', '2026-01-01')"
|
|
))
|
|
|
|
with pytest.raises(RuntimeError, match="has data.*incomplete.*manual"):
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(module._migrate_sync, dry_run=False)
|
|
|
|
async with engine.begin() as conn:
|
|
row = (
|
|
await conn.execute(text(f"SELECT target_key, action_id FROM {module.TABLE}"))
|
|
).one()
|
|
assert tuple(row) == ("opaque-1", "action-1")
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonempty_complete_columns_missing_fk_and_unique_constraint_fails():
|
|
module = _migration_module()
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
await conn.execute(text(f"DROP TABLE {module.TABLE}"))
|
|
await conn.execute(text(f"""
|
|
CREATE TABLE {module.TABLE} (
|
|
target_key VARCHAR(255) NOT NULL PRIMARY KEY,
|
|
action_id VARCHAR(64) NOT NULL,
|
|
lease_expires_at DATETIME NOT NULL,
|
|
created_at DATETIME NOT NULL,
|
|
updated_at DATETIME NOT NULL
|
|
)
|
|
"""))
|
|
await conn.execute(text(f"CREATE INDEX {module.ACTION_INDEX} ON {module.TABLE} (action_id)"))
|
|
await conn.execute(text(f"CREATE INDEX {module.EXPIRY_INDEX} ON {module.TABLE} (lease_expires_at)"))
|
|
await conn.execute(text(
|
|
f"INSERT INTO {module.TABLE} "
|
|
"(target_key, action_id, lease_expires_at, created_at, updated_at) "
|
|
"VALUES ('opaque-2', 'action-2', '2026-01-01', '2026-01-01', '2026-01-01')"
|
|
))
|
|
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(module._migrate_sync, dry_run=False)
|
|
message = str(exc_info.value)
|
|
assert "foreign key" in message
|
|
assert "unique constraint" in message
|
|
|
|
async with engine.begin() as conn:
|
|
assert (await conn.execute(text(f"SELECT COUNT(*) FROM {module.TABLE}"))).scalar_one() == 1
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase5_migration_dry_run_is_safe_when_table_is_absent():
|
|
module = _migration_module()
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(module._migrate_sync, dry_run=True)
|
|
tables = await conn.run_sync(lambda sync_conn: inspect(sync_conn).get_table_names())
|
|
assert "maf_roundtable_action_leases" not in tables
|
|
await engine.dispose()
|
|
|
|
|
|
def test_phase5_ddl_uses_sqlite_datetime():
|
|
module = _migration_module()
|
|
ddl = module._create_lease_table_sql(SimpleDialect("sqlite"))
|
|
assert "lease_expires_at DATETIME NOT NULL" in ddl
|
|
assert "created_at DATETIME NOT NULL" in ddl
|
|
|
|
|
|
def test_phase5_ddl_uses_postgresql_timestamp_with_timezone():
|
|
module = _migration_module()
|
|
ddl = module._create_lease_table_sql(SimpleDialect("postgresql"))
|
|
assert "lease_expires_at TIMESTAMP WITH TIME ZONE NOT NULL" in ddl
|
|
assert "created_at TIMESTAMP WITH TIME ZONE NOT NULL" in ddl
|
|
assert "DATETIME" not in ddl
|
|
|
|
|
|
def test_phase5_ddl_uses_mysql_microsecond_datetime():
|
|
module = _migration_module()
|
|
ddl = module._create_lease_table_sql(SimpleDialect("mysql"))
|
|
assert "lease_expires_at DATETIME(6) NOT NULL" in ddl
|
|
assert "created_at DATETIME(6) NOT NULL" in ddl
|
|
|
|
|
|
class SimpleDialect:
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|