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


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