110 lines
3.4 KiB
Python
110 lines
3.4 KiB
Python
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:
|
|
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_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 == []
|