from __future__ import annotations

import hashlib
import os
from io import StringIO
from pathlib import Path

import pytest
from alembic import command
from alembic.config import Config
from alembic.script import ScriptDirectory
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import create_async_engine


def test_alembic_upgrade_and_downgrade(tmp_path: Path, monkeypatch: object) -> None:
    root = Path(__file__).parents[1]
    database = tmp_path / "migration.db"
    url = f"sqlite+aiosqlite:///{database}"
    monkeypatch.setenv("SCRIPT_BUILD_WRITE_DATABASE_URL", url)  # type: ignore[attr-defined]
    config = Config(str(root / "alembic.ini"))
    command.upgrade(config, "head")
    sync_engine = create_engine(f"sqlite:///{database}")
    expected = {
        "script_build_mission_binding",
        "script_build_input_snapshot",
        "script_build_artifact_version",
        "script_build_publication",
    }
    assert expected <= set(inspect(sync_engine).get_table_names())
    command.downgrade(config, "base")
    assert not (expected & set(inspect(sync_engine).get_table_names()))
    sync_engine.dispose()


def test_migration_checksum_manifest() -> None:
    root = Path(__file__).parents[1]
    manifest = root / "migrations" / "CHECKSUMS.sha256"
    lines = [line.split(maxsplit=1) for line in manifest.read_text(encoding="utf-8").splitlines()]
    assert lines
    for expected, relative in lines:
        path = root / relative
        assert hashlib.sha256(path.read_bytes()).hexdigest() == expected


def test_revision_identifiers_fit_alembic_mysql_version_column() -> None:
    root = Path(__file__).parents[1]
    scripts = ScriptDirectory.from_config(Config(str(root / "alembic.ini")))
    assert all(len(revision.revision) <= 32 for revision in scripts.walk_revisions())


def test_mysql_offline_sql_compiles_without_credentials(monkeypatch: object) -> None:
    root = Path(__file__).parents[1]
    output = StringIO()
    monkeypatch.setenv(  # type: ignore[attr-defined]
        "SCRIPT_BUILD_WRITE_DATABASE_URL", "mysql+asyncmy://"
    )
    config = Config(str(root / "alembic.ini"), output_buffer=output)
    command.upgrade(config, "head", sql=True)
    sql = output.getvalue()
    assert sql.count("CREATE TABLE script_build_") == 5
    assert "CREATE TABLE script_build_paragraph" not in sql
    assert "CREATE TABLE script_build_element" not in sql
    assert "DROP CHECK ck_artifact_phase_one_type" in sql
    assert "ADD CONSTRAINT ck_artifact_business_type CHECK" in sql
    assert "'root_delivery_manifest'" in sql
    assert "CREATE TABLE script_build_http_command" in sql
    assert "CONSTRAINT uq_publication_build_type UNIQUE" in sql
    assert "owner_epoch BIGINT NOT NULL" in sql
    assert "owner_acquired_at DATETIME(6)" in sql
    assert sql.count("WITH CASCADED CHECK OPTION") == 4
    assert "uq_script_build_record_reson_trace" in sql


def test_phase_three_offline_downgrade_fails_closed(monkeypatch: object) -> None:
    root = Path(__file__).parents[1]
    output = StringIO()
    monkeypatch.setenv(  # type: ignore[attr-defined]
        "SCRIPT_BUILD_WRITE_DATABASE_URL", "mysql+asyncmy://"
    )
    config = Config(str(root / "alembic.ini"), output_buffer=output)
    with pytest.raises(RuntimeError, match="online safety preflight"):
        command.downgrade(
            config,
            "0003_phase_three_publication:0002_phase_two_artifacts",
            sql=True,
        )


def test_phase_two_artifacts_make_downgrade_refuse_data_loss(
    tmp_path: Path, monkeypatch: object
) -> None:
    root = Path(__file__).parents[1]
    database = tmp_path / "migration-refusal.db"
    url = f"sqlite+aiosqlite:///{database}"
    monkeypatch.setenv("SCRIPT_BUILD_WRITE_DATABASE_URL", url)  # type: ignore[attr-defined]
    config = Config(str(root / "alembic.ini"))
    command.upgrade(config, "head")
    sync_engine = create_engine(f"sqlite:///{database}")
    with sync_engine.begin() as connection:
        connection.execute(
            text(
                "INSERT INTO script_build_artifact_version "
                "(id, script_build_id, task_id, attempt_id, spec_version, artifact_type, "
                "legacy_branch_id, canonical_json, canonical_sha256, state, created_at, frozen_at) "
                "VALUES (1, 1, 'task', 'attempt', 1, 'paragraph', 1, '{}', :digest, "
                "'frozen', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
            ),
            {"digest": "a" * 64},
        )
    with pytest.raises(RuntimeError, match="cannot downgrade while phase-two artifacts exist"):
        command.downgrade(config, "0001_phase_one")
    assert inspect(sync_engine).has_table("script_build_artifact_version")
    sync_engine.dispose()


@pytest.mark.mysql
@pytest.mark.asyncio
async def test_explicit_mysql_dsn_is_reachable() -> None:
    dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_DSN")
    if not dsn:
        pytest.skip("SCRIPT_BUILD_TEST_MYSQL_DSN is not configured")
    engine = create_async_engine(dsn, pool_pre_ping=True)
    try:
        async with engine.connect() as connection:
            assert await connection.scalar(text("SELECT 1")) == 1
    finally:
        await engine.dispose()


@pytest.mark.mysql
@pytest.mark.asyncio
async def test_mysql_phase_three_schema_semantics() -> None:
    dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_DSN")
    if not dsn:
        pytest.skip("SCRIPT_BUILD_TEST_MYSQL_DSN is not configured")
    engine = create_async_engine(dsn, pool_pre_ping=True, isolation_level="READ COMMITTED")
    try:
        async with engine.connect() as connection:
            version = str(await connection.scalar(text("SELECT VERSION()"))).split("-", 1)[0]
            assert tuple(int(part) for part in version.split(".")[:3]) >= (8, 0, 16)
            assert await connection.scalar(text("SELECT @@transaction_isolation")) == (
                "READ-COMMITTED"
            )
            assert await connection.scalar(text("SELECT version_num FROM alembic_version")) == (
                "0003_phase_three_publication"
            )
            precision = await connection.scalar(
                text(
                    "SELECT DATETIME_PRECISION FROM information_schema.COLUMNS "
                    "WHERE TABLE_SCHEMA=DATABASE() "
                    "AND TABLE_NAME='script_build_mission_binding' "
                    "AND COLUMN_NAME='owner_acquired_at'"
                )
            )
            assert precision == 6
            views = set(
                (
                    await connection.execute(
                        text(
                            "SELECT TABLE_NAME FROM information_schema.VIEWS "
                            "WHERE TABLE_SCHEMA=DATABASE()"
                        )
                    )
                ).scalars()
            )
            assert {
                "script_build_candidate_paragraph_v",
                "script_build_candidate_element_v",
                "script_build_candidate_paragraph_element_v",
                "script_build_runtime_record_v",
            } <= views

            transaction = await connection.begin_nested()
            try:
                await connection.execute(
                    text(
                        "INSERT INTO script_build_candidate_paragraph_v "
                        "(script_build_id, branch_id, paragraph_index, level, is_active) "
                        "VALUES (900000, 1, 1, 1, 1)"
                    )
                )
                with pytest.raises(DBAPIError):
                    await connection.execute(
                        text(
                            "INSERT INTO script_build_candidate_paragraph_v "
                            "(script_build_id, branch_id, paragraph_index, level, is_active) "
                            "VALUES (900000, 0, 2, 1, 1)"
                        )
                    )
                await connection.execute(
                    text(
                        "INSERT INTO script_build_runtime_record_v "
                        "(id, execution_id, topic_build_id, topic_id, status, reson_trace_id, "
                        "is_deleted, is_favorited) VALUES "
                        "(900000, 1, 1, 1, 'running', 'mysql-gate-runtime', 0, 0)"
                    )
                )
                with pytest.raises(DBAPIError):
                    await connection.execute(
                        text(
                            "UPDATE script_build_runtime_record_v SET status='success' "
                            "WHERE id=900000"
                        )
                    )
                with pytest.raises(DBAPIError):
                    await connection.execute(
                        text(
                            "INSERT INTO script_build_http_command "
                            "(principal_scope_sha256, route_family, idempotency_key, "
                            "request_fingerprint, state, created_at, updated_at) VALUES "
                            "(:digest, 'gate', 'bad-state', :digest, 'invalid', NOW(6), NOW(6))"
                        ),
                        {"digest": "a" * 64},
                    )
            finally:
                await transaction.rollback()
    finally:
        await engine.dispose()


@pytest.mark.mysql
@pytest.mark.asyncio
async def test_mysql_phase_three_least_privilege_accounts() -> None:
    read_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_READ_DSN")
    runtime_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_RUNTIME_DSN")
    final_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_FINAL_DSN")
    if not all((read_dsn, runtime_dsn, final_dsn)):
        pytest.skip("the three least-privilege MySQL DSNs are not configured")
    engines = [
        create_async_engine(str(read_dsn), isolation_level="READ COMMITTED"),
        create_async_engine(str(runtime_dsn), isolation_level="READ COMMITTED"),
        create_async_engine(str(final_dsn), isolation_level="READ COMMITTED"),
    ]
    try:
        async with engines[0].connect() as connection:
            with pytest.raises(DBAPIError):
                await connection.execute(
                    text(
                        "INSERT INTO script_build_http_command "
                        "(principal_scope_sha256, route_family, idempotency_key, "
                        "request_fingerprint, state, created_at, updated_at) VALUES "
                        "(:digest, 'grant-gate', 'read', :digest, 'reserved', NOW(6), NOW(6))"
                    ),
                    {"digest": "b" * 64},
                )

        async with engines[1].connect() as connection:
            transaction = await connection.begin()
            try:
                with pytest.raises(DBAPIError):
                    await connection.execute(
                        text(
                            "INSERT INTO script_build_paragraph "
                            "(script_build_id, branch_id, paragraph_index, level, is_active) "
                            "VALUES (930001, 0, 1, 1, 1)"
                        )
                    )
                await connection.execute(
                    text(
                        "INSERT INTO script_build_candidate_paragraph_v "
                        "(script_build_id, branch_id, paragraph_index, level, is_active) "
                        "VALUES (930001, 1, 1, 1, 1)"
                    )
                )
                await connection.execute(
                    text(
                        "INSERT INTO script_build_runtime_record_v "
                        "(id, execution_id, topic_build_id, topic_id, status, reson_trace_id, "
                        "is_deleted, is_favorited) VALUES "
                        "(930001, 1, 1, 1, 'running', 'least-privilege-runtime', 0, 0)"
                    )
                )
                with pytest.raises(DBAPIError):
                    await connection.execute(
                        text(
                            "UPDATE script_build_runtime_record_v SET status='success' "
                            "WHERE id=930001"
                        )
                    )
            finally:
                await transaction.rollback()

        async with engines[2].connect() as connection:
            transaction = await connection.begin()
            try:
                await connection.execute(
                    text(
                        "INSERT INTO script_build_paragraph "
                        "(script_build_id, branch_id, paragraph_index, level, is_active) "
                        "VALUES (930002, 0, 1, 1, 1)"
                    )
                )
                with pytest.raises(DBAPIError):
                    await connection.execute(
                        text(
                            "INSERT INTO script_build_http_command "
                            "(principal_scope_sha256, route_family, idempotency_key, "
                            "request_fingerprint, state, created_at, updated_at) VALUES "
                            "(:digest, 'grant-gate', 'final', :digest, 'reserved', NOW(6), NOW(6))"
                        ),
                        {"digest": "c" * 64},
                    )
            finally:
                await transaction.rollback()
    finally:
        for engine in engines:
            await engine.dispose()