from __future__ import annotations import importlib import os import stat from datetime import UTC, datetime, timedelta, timezone from pathlib import Path from typing import Any from uuid import UUID import pytest from alembic import command from alembic.config import Config from sqlalchemy import inspect, select, text from sqlalchemy.ext.asyncio import async_sessionmaker EXPECTED_TABLES = { "alembic_version", "api_tokens", "audit_events", "backups", "executions", "idempotency_records", "jobs", "notification_deliveries", "notification_subscriptions", "repositories", "restores", "schedules", "secrets", "sources", "users", } def settings_for(tmp_path: Path) -> Any: config_module = importlib.import_module("backup_tool.config") key = tmp_path / "master.key" key.write_bytes(os.urandom(32)) key.chmod(stat.S_IRUSR | stat.S_IWUSR) roots = [] for name in ("repositories", "sources", "restores"): root = tmp_path / name root.mkdir() roots.append(root) return config_module.Settings( database_url=f"sqlite+aiosqlite:///{(tmp_path / 'metadata.db').as_posix()}", data_dir=tmp_path, repository_roots=[roots[0]], local_source_roots=[roots[1]], restore_roots=[roots[2]], master_key_file=key, ) def alembic_config(database_url: str) -> Config: project_root = Path(__file__).resolve().parents[2] config = Config(project_root / "backend" / "alembic.ini") config.set_main_option("sqlalchemy.url", database_url) return config @pytest.mark.asyncio async def test_migration_round_trip_and_sqlite_pragmas(tmp_path: Path) -> None: db_module = importlib.import_module("backup_tool.db.engine") settings = settings_for(tmp_path) migration = alembic_config(settings.database_url) command.upgrade(migration, "head") engine = db_module.create_engine(settings) try: await db_module.assert_schema_current(engine, migration) async with engine.connect() as connection: tables = await connection.run_sync( lambda sync_connection: set(inspect(sync_connection).get_table_names()) ) foreign_keys = await connection.scalar(text("PRAGMA foreign_keys")) journal_mode = await connection.scalar(text("PRAGMA journal_mode")) busy_timeout = await connection.scalar(text("PRAGMA busy_timeout")) assert tables == EXPECTED_TABLES assert foreign_keys == 1 assert str(journal_mode).lower() == "wal" assert busy_timeout == settings.sqlite_busy_timeout_ms finally: await engine.dispose() command.downgrade(migration, "base") command.upgrade(migration, "head") @pytest.mark.asyncio async def test_startup_rejects_unmigrated_database(tmp_path: Path) -> None: db_module = importlib.import_module("backup_tool.db.engine") settings = settings_for(tmp_path) migration = alembic_config(settings.database_url) engine = db_module.create_engine(settings) try: with pytest.raises(db_module.SchemaNotCurrentError): await db_module.assert_schema_current(engine, migration) finally: await engine.dispose() def test_database_role_rejects_unmigrated_database(tmp_path: Path) -> None: cli = importlib.import_module("backup_tool.cli") db_module = importlib.import_module("backup_tool.db.engine") settings = settings_for(tmp_path) with pytest.raises(db_module.SchemaNotCurrentError): cli.main(["worker"], settings=settings) @pytest.mark.asyncio async def test_persisted_ids_and_timestamps_are_uuid7_and_aware_utc(tmp_path: Path) -> None: db_module = importlib.import_module("backup_tool.db.engine") models = importlib.import_module("backup_tool.db.models") settings = settings_for(tmp_path) migration = alembic_config(settings.database_url) command.upgrade(migration, "head") engine = db_module.create_engine(settings) supplied = datetime(2024, 6, 1, 12, 30, tzinfo=timezone(timedelta(hours=5, minutes=30))) sessions = async_sessionmaker(engine, expire_on_commit=False) try: async with sessions() as session: user = models.User( username="operator", password_hash="hash", created_at=supplied, updated_at=supplied, ) repository = models.Repository( name="primary", root=str(tmp_path / "repositories"), format_version=1, compression="none", encryption="none", ) session.add_all([user, repository]) await session.commit() user_id = user.id repository_id = repository.id async with sessions() as session: stored_user = await session.scalar(select(models.User).where(models.User.id == user_id)) stored_repository = await session.scalar( select(models.Repository).where(models.Repository.id == repository_id) ) assert stored_user is not None assert stored_repository is not None assert UUID(stored_user.id).version == 7 assert UUID(stored_repository.id).version == 7 assert stored_user.created_at.tzinfo is UTC assert stored_user.created_at == supplied.astimezone(UTC) assert stored_repository.created_at.tzinfo is UTC finally: await engine.dispose() def test_alembic_connections_apply_sqlite_configuration( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: sqlite_runtime = importlib.import_module("backup_tool.db.sqlite") settings = settings_for(tmp_path) migration = alembic_config(settings.database_url) observed: list[tuple[int, str, int]] = [] original = sqlite_runtime.configure_sqlite_connection def recording(connection: Any) -> None: original(connection) cursor = connection.cursor() try: foreign_keys = cursor.execute("PRAGMA foreign_keys").fetchone()[0] journal_mode = cursor.execute("PRAGMA journal_mode").fetchone()[0] busy_timeout = cursor.execute("PRAGMA busy_timeout").fetchone()[0] observed.append((foreign_keys, str(journal_mode).lower(), busy_timeout)) finally: cursor.close() monkeypatch.setattr(sqlite_runtime, "configure_sqlite_connection", recording) monkeypatch.setenv("BACKUP_TOOL_SQLITE_BUSY_TIMEOUT_MS", "7000") command.upgrade(migration, "head") assert observed expected_pragmas = (1, "wal", 7000) assert observed[-1] == expected_pragmas @pytest.mark.asyncio async def test_baseline_constraints_and_indexes_exist(tmp_path: Path) -> None: db_module = importlib.import_module("backup_tool.db.engine") settings = settings_for(tmp_path) migration = alembic_config(settings.database_url) command.upgrade(migration, "head") engine = db_module.create_engine(settings) try: async with engine.connect() as connection: execution_indexes = await connection.run_sync( lambda sync: inspect(sync).get_indexes("executions") ) schedule_uniques = await connection.run_sync( lambda sync: inspect(sync).get_unique_constraints("schedules") ) execution_checks = await connection.run_sync( lambda sync: inspect(sync).get_check_constraints("executions") ) assert any( index["name"] == "uq_executions_active_job" and index["unique"] for index in execution_indexes ) assert any(unique["column_names"] == ["job_id"] for unique in schedule_uniques) check_names = {check["name"] for check in execution_checks} assert {"ck_executions_state", "ck_executions_trigger"} <= check_names finally: await engine.dispose() def test_runtime_never_uses_create_all() -> None: offenders = [] for root in (Path("backend/src"), Path("backend/alembic")): if not root.exists(): continue for path in root.rglob("*.py"): if "create_all" in path.read_text(encoding="utf-8"): offenders.append(path.as_posix()) assert offenders == []