awg-dash/backend/tests/test_store_migrations.py

"""Тесты схемы, миграций и проверки версии при старте (issue #20)."""

from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest

from awg_api.store import (
    EXPECTED_TABLES,
    LATEST_VERSION,
    MIGRATIONS,
    Database,
    Migration,
    SchemaError,
    StorageError,
    current_version,
    migrate,
    resolve_db_path,
    verify_schema,
)
from awg_api.store.migrations import applied_migrations, table_columns, table_names


def test_empty_data_dir_creates_schema(tmp_path: Path) -> None:
    """GIVEN пустой /data, WHEN старт, THEN схема v1 создана автоматически."""
    db_path = tmp_path / "data" / "awg-dash.db"
    assert not db_path.parent.exists()

    with Database(db_path) as db:
        assert db.schema_version == LATEST_VERSION == 1
        assert db.migration_report is not None
        assert db.migration_report.applied == (1,)
        names = table_names(db.connection)
        assert set(EXPECTED_TABLES) <= names
        for table, columns in EXPECTED_TABLES.items():
            assert columns <= table_columns(db.connection, table), table


def test_wal_mode_is_enabled(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        assert db.journal_mode.lower() == "wal"
        assert db.is_wal is True


def test_seed_rows_present(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        row = db.connection.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
        assert row is not None and row["mode"] == "off"
        geoip = db.connection.execute("SELECT cron, provider FROM geoip_schedule WHERE id = 1").fetchone()
        assert geoip is not None
        assert geoip["provider"] == "ripe"


def test_migrate_is_idempotent(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        report = db.migrate()
        assert report.applied == ()
        assert report.changed is False
        assert report.from_version == report.to_version == LATEST_VERSION


def test_applied_migrations_map(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        assert applied_migrations(db.connection) == {1: "initial-state-schema"}


def test_auto_migrate_false_verifies_existing_schema(tmp_path: Path) -> None:
    path = tmp_path / "awg-dash.db"
    with Database(path):
        pass
    with Database(path, auto_migrate=False) as db:
        assert db.migration_report is None
        assert current_version(db.connection) == LATEST_VERSION


def test_auto_migrate_false_on_unmigrated_db_raises(tmp_path: Path) -> None:
    conn = sqlite3.connect(str(tmp_path / "empty.db"))
    try:
        with pytest.raises(SchemaError, match="не мигрирована"):
            verify_schema(conn)
    finally:
        conn.close()


def test_verify_schema_detects_missing_table(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        db.connection.execute("DROP TABLE routing_rules")
        with pytest.raises(SchemaError, match="routing_rules"):
            db.verify_schema()


def test_verify_schema_detects_missing_column(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        db.connection.execute("ALTER TABLE devices DROP COLUMN address")
        with pytest.raises(SchemaError, match="address"):
            db.verify_schema()


def test_verify_schema_rejects_newer_schema(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        db.connection.execute(
            "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
            (99, "from-the-future", "2099-01-01T00:00:00.000Z"),
        )
        with pytest.raises(SchemaError, match="новее"):
            db.verify_schema()
        with pytest.raises(SchemaError, match="новее"):
            db.migrate()


def test_failed_migration_leaves_schema_untouched(tmp_path: Path) -> None:
    broken = MIGRATIONS + (
        Migration(version=2, name="broken", statements=("CREATE TABLE t2 (id INTEGER)", "NOT SQL AT ALL")),
    )
    with Database(tmp_path / "awg-dash.db") as db:
        with pytest.raises(StorageError, match="миграция v2"):
            migrate(db.connection, migrations=broken)
        assert current_version(db.connection) == 1
        assert "t2" not in table_names(db.connection)


def test_migrate_rejects_downgrade(tmp_path: Path) -> None:
    with Database(tmp_path / "awg-dash.db") as db:
        db.connection.execute(
            "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
            (2, "newer", "2099-01-01T00:00:00.000Z"),
        )
        with pytest.raises(SchemaError, match="новее"):
            migrate(db.connection, migrations=MIGRATIONS)


def test_resolve_db_path_defaults_and_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("AWG_DASH_DB", raising=False)
    assert resolve_db_path() == Path("/data/awg-dash.db")
    monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "custom.db"))
    assert resolve_db_path() == tmp_path / "custom.db"
    assert resolve_db_path(tmp_path / "explicit.db") == tmp_path / "explicit.db"


def test_database_uses_env_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "nested" / "env.db"))
    with Database() as db:
        assert db.path == tmp_path / "nested" / "env.db"
        assert db.path.exists()


def test_unwritable_path_raises_storage_error(tmp_path: Path) -> None:
    blocker = tmp_path / "blocked"
    blocker.write_text("not a directory")
    with pytest.raises(StorageError, match="каталог"):
        Database(blocker / "awg-dash.db")