script-build-and-framework-tmp/script_build_host/tests/test_migrations.py

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 sqlalchemy import create_engine, inspect, text
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_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_") == 4
    assert "script_build_paragraph" not in sql
    assert "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


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()