From 45a526a1e9e4587c3360d04b6be11bc4497e746b Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Mon, 27 Jul 2026 19:34:14 +0200 Subject: [PATCH] test(v2): prove runtime dispatch and persistence invariants --- tests/integration/test_migrations.py | 115 ++++++++++++++++++++++++++- tests/unit/test_config.py | 31 +++++++- 2 files changed, 144 insertions(+), 2 deletions(-) diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index c7c0acb..68b6306 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -3,13 +3,16 @@ 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, text +from sqlalchemy import inspect, select, text +from sqlalchemy.ext.asyncio import async_sessionmaker EXPECTED_TABLES = { "alembic_version", @@ -98,6 +101,116 @@ async def test_startup_rejects_unmigrated_database(tmp_path: Path) -> None: 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")): diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index df80b93..44579a3 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -94,6 +94,35 @@ def test_system_clock_returns_aware_utc() -> None: def test_cli_declares_isolated_runtime_roles() -> None: parser = build_parser() - for role in ("web", "scheduler", "worker", "migrate", "admin"): + for role in ("web", "scheduler", "worker", "admin"): args = parser.parse_args([role]) assert args.role == role + + +def test_cli_dispatches_selected_role(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + settings = valid_settings(tmp_path) + observed: list[Any] = [] + + def handler(received: Any) -> int: + observed.append(received) + return 23 + + monkeypatch.setattr(cli, "require_current_schema", lambda _settings: None) + monkeypatch.setitem(cli.ROLE_HANDLERS, "web", handler) + assert cli.main(["web"], settings=settings) == 23 + assert observed == [settings] + + +def test_migrate_upgrade_dispatches_to_alembic( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + settings = valid_settings(tmp_path) + calls: list[tuple[str, str]] = [] + + def upgrade(_config: Any, revision: str) -> None: + calls.append(("upgrade", revision)) + + monkeypatch.setattr(cli.command, "upgrade", upgrade) + assert cli.main(["migrate", "upgrade"], settings=settings) == 0 + expected_calls = [("upgrade", "head")] + assert calls == expected_calls