From 8f784848d7baf0118ae0680f38b93344dbf993ee Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Mon, 27 Jul 2026 19:06:14 +0200 Subject: [PATCH] test(v2): define runtime and metadata contracts --- tests/integration/test_migrations.py | 108 +++++++++++++++++++++++++++ tests/unit/test_config.py | 99 ++++++++++++++++++++++++ 2 files changed, 207 insertions(+) create mode 100644 tests/integration/test_migrations.py create mode 100644 tests/unit/test_config.py diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py new file mode 100644 index 0000000..a9f556a --- /dev/null +++ b/tests/integration/test_migrations.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import importlib +import os +import stat +from pathlib import Path +from typing import Any + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import inspect, text + +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: + config = Config("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_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 == [] diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..df80b93 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import importlib +import os +import stat +from datetime import UTC +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +cli = importlib.import_module("backup_tool.cli") +clock = importlib.import_module("backup_tool.clock") +config = importlib.import_module("backup_tool.config") +ids = importlib.import_module("backup_tool.ids") +build_parser = cli.build_parser +SystemClock = clock.SystemClock +Settings = config.Settings +new_uuid7 = ids.new_uuid7 + + +def valid_settings(tmp_path: Path, **overrides: object) -> Any: + data = tmp_path / "data" + repositories = tmp_path / "repositories" + sources = tmp_path / "sources" + restores = tmp_path / "restores" + for path in (data, repositories, sources, restores): + path.mkdir() + key = tmp_path / "master.key" + key.write_bytes(os.urandom(32)) + key.chmod(stat.S_IRUSR | stat.S_IWUSR) + values: dict[str, object] = { + "database_url": f"sqlite+aiosqlite:///{(data / 'metadata.db').as_posix()}", + "data_dir": data, + "repository_roots": [repositories], + "local_source_roots": [sources], + "restore_roots": [restores], + "master_key_file": key, + } + values.update(overrides) + return Settings(**values) + + +def test_settings_canonicalize_absolute_roots_and_database(tmp_path: Path) -> None: + settings = valid_settings(tmp_path) + expected_database = (tmp_path / "data" / "metadata.db").resolve() + expected_repositories = ((tmp_path / "repositories").resolve(),) + expected_sources = ((tmp_path / "sources").resolve(),) + expected_restores = ((tmp_path / "restores").resolve(),) + assert settings.database_path == expected_database + assert settings.repository_roots == expected_repositories + assert settings.local_source_roots == expected_sources + assert settings.restore_roots == expected_restores + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("database_url", "sqlite+aiosqlite:///relative.db"), + ("worker_concurrency", 2), + ("min_free_percent", 0), + ("min_free_percent", 100), + ], +) +def test_settings_reject_unsafe_values(tmp_path: Path, field: str, value: object) -> None: + with pytest.raises(ValidationError): + valid_settings(tmp_path, **{field: value}) + + +def test_settings_reject_unknown_fields(tmp_path: Path) -> None: + with pytest.raises(ValidationError): + valid_settings(tmp_path, unexpected=True) + + +def test_settings_reject_insecure_master_key_permissions(tmp_path: Path) -> None: + settings = valid_settings(tmp_path) + settings.master_key_file.chmod(0o644) + with pytest.raises(ValidationError, match="0600"): + Settings(**settings.model_dump()) + + +def test_uuid7_is_versioned_and_monotonically_ordered() -> None: + identifiers = [new_uuid7() for _ in range(100)] + assert all(identifier.version == 7 for identifier in identifiers) + assert identifiers == sorted(identifiers) + assert len(set(identifiers)) == len(identifiers) + + +def test_system_clock_returns_aware_utc() -> None: + now = SystemClock().now() + assert now.tzinfo is UTC + + +def test_cli_declares_isolated_runtime_roles() -> None: + parser = build_parser() + for role in ("web", "scheduler", "worker", "migrate", "admin"): + args = parser.parse_args([role]) + assert args.role == role