gps-denied-onboard/tests/unit/c6_tile_cache/test_migrations_runner.py

"""AZ-304 NFR-reliability-retry + unit tests for the c6_tile_cache migration runner.

These tests stub :mod:`gps_denied_onboard.components.c6_tile_cache.migrations`
internals so the runner's retry / DSN-resolution / error-mapping paths can
be exercised without a real Postgres. The integration suite
(``test_postgres_schema.py``, ``@pytest.mark.docker``) covers the happy
path end-to-end.
"""

from __future__ import annotations

from typing import Any

import pytest
from sqlalchemy.exc import DBAPIError

from gps_denied_onboard.components.c6_tile_cache import migrations as migrations_module
from gps_denied_onboard.components.c6_tile_cache.config import C6TileCacheConfig
from gps_denied_onboard.components.c6_tile_cache.migrations import (
    MigrationError,
    MigrationResult,
    apply_migrations,
)
from gps_denied_onboard.config.schema import Config


def _config(dsn: str = "postgresql://user:pass@host:5432/db") -> Config:
    return Config.with_blocks(c6_tile_cache=C6TileCacheConfig(postgres_dsn=dsn))


class _FakeSerializationFailure(Exception):
    """Lightweight stand-in for ``psycopg.errors.SerializationFailure``."""

    sqlstate = "40001"


def _make_dbapi_error(orig: Exception) -> DBAPIError:
    """Construct a SQLAlchemy DBAPIError wrapping ``orig`` (matches runtime shape)."""
    return DBAPIError(statement="", params=None, orig=orig)


# ----------------------------------------------------------------------
# DSN resolution.


def test_resolve_dsn_prefers_config_over_env(monkeypatch: pytest.MonkeyPatch) -> None:
    # Arrange
    monkeypatch.setenv("DB_URL", "postgresql://env-host/db")
    cfg = _config("postgresql://cfg-host/db")

    # Act
    dsn = migrations_module._resolve_dsn(cfg)

    # Assert
    assert dsn == "postgresql://cfg-host/db"


def test_resolve_dsn_falls_back_to_env(monkeypatch: pytest.MonkeyPatch) -> None:
    # Arrange
    monkeypatch.setenv("DB_URL", "postgresql://env-host/db")
    cfg = _config("")

    # Act
    dsn = migrations_module._resolve_dsn(cfg)

    # Assert
    assert dsn == "postgresql://env-host/db"


def test_resolve_dsn_raises_when_no_source(monkeypatch: pytest.MonkeyPatch) -> None:
    # Arrange
    monkeypatch.delenv("DB_URL", raising=False)
    cfg = _config("")

    # Act + Assert
    with pytest.raises(MigrationError, match="no DSN available"):
        migrations_module._resolve_dsn(cfg)


def test_to_sqlalchemy_url_rewrites_postgresql_prefix() -> None:
    # Act + Assert
    assert migrations_module._to_sqlalchemy_url("postgresql://h/db") == "postgresql+psycopg://h/db"
    # Already-prefixed URLs pass through unchanged.
    assert (
        migrations_module._to_sqlalchemy_url("postgresql+psycopg://h/db")
        == "postgresql+psycopg://h/db"
    )


# ----------------------------------------------------------------------
# NFR-reliability-retry: one SerializationFailure → retry → succeed.


def test_nfr_reliability_retry_once_on_serialization_failure(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Arrange — stub all DB-touching internals so the retry path is the only
    # variable. Both iterations see the DB BELOW head pre-upgrade; the first
    # upgrade attempt raises SQLSTATE 40001, the second succeeds. Post-upgrade
    # revision is read once after the successful attempt.
    rev_sequence = iter([None, None, "0002_c6_tile_identity_and_lru"])
    upgrade_calls: list[int] = []

    def fake_upgrade(_cfg: Any, _target: str) -> None:
        upgrade_calls.append(len(upgrade_calls))
        if len(upgrade_calls) == 1:
            raise _make_dbapi_error(_FakeSerializationFailure("conflict"))

    def fake_current_revision(_url: str) -> str | None:
        return next(rev_sequence)

    monkeypatch.setattr(
        "gps_denied_onboard.components.c6_tile_cache.migrations.command.upgrade",
        fake_upgrade,
    )
    monkeypatch.setattr(migrations_module, "_current_revision", fake_current_revision)
    monkeypatch.setattr(
        migrations_module,
        "_head_revision",
        lambda _cfg: "0002_c6_tile_identity_and_lru",
    )
    monkeypatch.setattr(
        migrations_module,
        "_resolve_applied",
        lambda _cfg, _pre, _post: ["0002_c6_tile_identity_and_lru"],
    )

    # Act
    result = apply_migrations(_config())

    # Assert
    assert isinstance(result, MigrationResult)
    assert result.applied == ["0002_c6_tile_identity_and_lru"]
    assert result.no_op is False
    assert len(upgrade_calls) == 2, "runner must retry on SQLSTATE 40001"


def test_nfr_reliability_terminal_after_two_serialization_failures(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Arrange
    def fake_upgrade(_cfg: Any, _target: str) -> None:
        raise _make_dbapi_error(_FakeSerializationFailure("persistent conflict"))

    monkeypatch.setattr(
        "gps_denied_onboard.components.c6_tile_cache.migrations.command.upgrade",
        fake_upgrade,
    )
    monkeypatch.setattr(migrations_module, "_current_revision", lambda _url: None)
    monkeypatch.setattr(
        migrations_module,
        "_head_revision",
        lambda _cfg: "0002_c6_tile_identity_and_lru",
    )

    # Act + Assert
    with pytest.raises(MigrationError, match="database error"):
        apply_migrations(_config())


def test_nfr_reliability_does_not_retry_on_non_serialization_dbapi_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Arrange
    upgrade_calls: list[int] = []

    class _GenericDbApiOrig(Exception):
        sqlstate = "42P01"  # "undefined_table" — NOT a serialization conflict

    def fake_upgrade(_cfg: Any, _target: str) -> None:
        upgrade_calls.append(len(upgrade_calls))
        raise _make_dbapi_error(_GenericDbApiOrig("missing relation"))

    monkeypatch.setattr(
        "gps_denied_onboard.components.c6_tile_cache.migrations.command.upgrade",
        fake_upgrade,
    )
    monkeypatch.setattr(migrations_module, "_current_revision", lambda _url: None)
    monkeypatch.setattr(
        migrations_module,
        "_head_revision",
        lambda _cfg: "0002_c6_tile_identity_and_lru",
    )

    # Act + Assert
    with pytest.raises(MigrationError):
        apply_migrations(_config())
    assert len(upgrade_calls) == 1, "non-40001 errors must fail-fast (no retry)"


def test_apply_migrations_logs_no_op_when_already_at_head(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # Arrange
    monkeypatch.setattr(
        migrations_module, "_current_revision", lambda _url: "0002_c6_tile_identity_and_lru"
    )
    monkeypatch.setattr(
        migrations_module, "_head_revision", lambda _cfg: "0002_c6_tile_identity_and_lru"
    )

    # Act
    result = apply_migrations(_config())

    # Assert
    assert result == MigrationResult(
        applied=[], current_revision="0002_c6_tile_identity_and_lru", no_op=True
    )


def test_migration_error_is_not_a_tile_cache_error_member() -> None:
    """`MigrationError` deliberately lives outside the TileCacheError family."""
    # Arrange
    from gps_denied_onboard.components.c6_tile_cache.errors import TileCacheError

    # Act + Assert
    assert not issubclass(MigrationError, TileCacheError)