"""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