426 lines
15 KiB
Python
426 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import os
|
|
import sqlite3
|
|
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",
|
|
"execution_events",
|
|
"idempotency_records",
|
|
"jobs",
|
|
"notification_deliveries",
|
|
"notification_delivery_attempts",
|
|
"notification_email_settings",
|
|
"notification_events",
|
|
"notification_signing_keys",
|
|
"notification_subscriptions",
|
|
"repositories",
|
|
"repository_data_key_epochs",
|
|
"restores",
|
|
"schedules",
|
|
"secrets",
|
|
"sessions",
|
|
"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_local_only_sources_migration_rejects_existing_remote_sources(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
settings = settings_for(tmp_path)
|
|
migration = alembic_config(settings.database_url)
|
|
command.upgrade(migration, "0005_restore_dry_run")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("remote", "sftp", "{}", "[]", "active", None, "source-remote"),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="found 1 non-local source row"):
|
|
command.upgrade(migration, "head")
|
|
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute("UPDATE sources SET kind = ? WHERE id = ?", ("local", "source-remote"))
|
|
|
|
command.upgrade(migration, "head")
|
|
with (
|
|
sqlite3.connect(settings.database_path) as connection,
|
|
pytest.raises(sqlite3.IntegrityError),
|
|
):
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("remote-two", "sftp", "{}", "[]", "active", None, "source-remote-two"),
|
|
)
|
|
|
|
command.downgrade(migration, "0005_restore_dry_run")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("remote-three", "sftp", "{}", "[]", "active", None, "source-remote-three"),
|
|
)
|
|
|
|
|
|
def test_0009_allows_ssh_sources_and_refuses_populated_downgrade(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
settings = settings_for(tmp_path)
|
|
migration = alembic_config(settings.database_url)
|
|
command.upgrade(migration, "0008_notification_outbox")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("local", "local", '{"root":"/tmp"}', "[]", "active", None, "source-local"),
|
|
)
|
|
|
|
command.upgrade(migration, "head")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
local_source = connection.execute("SELECT kind FROM sources").fetchone()
|
|
assert local_source is not None
|
|
assert local_source[0] == "local"
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
"ssh",
|
|
"ssh",
|
|
'{"hostname":"backup.example.test"}',
|
|
'["secret"]',
|
|
"active",
|
|
None,
|
|
"source-ssh",
|
|
),
|
|
)
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("sftp", "sftp", "{}", "[]", "active", None, "source-sftp"),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="found 1 SSH source row"):
|
|
command.downgrade(migration, "0008_notification_outbox")
|
|
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute("DELETE FROM sources WHERE id = ?", ("source-ssh",))
|
|
command.downgrade(migration, "0008_notification_outbox")
|
|
with (
|
|
sqlite3.connect(settings.database_path) as connection,
|
|
pytest.raises(sqlite3.IntegrityError),
|
|
):
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sources (name, kind, public_config, secret_refs, state, last_probe, id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
"ssh-after-downgrade",
|
|
"ssh",
|
|
"{}",
|
|
"[]",
|
|
"active",
|
|
None,
|
|
"source-ssh-after",
|
|
),
|
|
)
|
|
|
|
|
|
def test_0008_migrates_unexpected_legacy_notification_delivery(tmp_path: Path) -> None:
|
|
settings = settings_for(tmp_path)
|
|
migration = alembic_config(settings.database_url)
|
|
command.upgrade(migration, "0007_repository_data_key_epochs")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO notification_subscriptions
|
|
(id, channel, event_filters, destination_config, state)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
"subscription",
|
|
"email",
|
|
'["execution.queued"]',
|
|
'{"recipients":["a@example.test"]}',
|
|
"active",
|
|
),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO notification_deliveries
|
|
(id, event_id, subscription_id, attempt, state, response_class)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
("delivery", "legacy-event", "subscription", 1, "delivered", "http_200"),
|
|
)
|
|
command.upgrade(migration, "head")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
event = connection.execute("SELECT type, payload FROM notification_events").fetchone()
|
|
attempt = connection.execute(
|
|
"SELECT number, outcome, response_class FROM notification_delivery_attempts"
|
|
).fetchone()
|
|
assert event is not None and event[0] == "notification.legacy"
|
|
assert "legacy-event" in event[1]
|
|
assert attempt is not None
|
|
assert attempt[0] == 1
|
|
assert attempt[1] == "delivered"
|
|
assert attempt[2] == "http_200"
|
|
|
|
|
|
def test_0007_downgrade_refuses_populated_key_metadata(tmp_path: Path) -> None:
|
|
settings = settings_for(tmp_path)
|
|
migration = alembic_config(settings.database_url)
|
|
command.upgrade(migration, "head")
|
|
with sqlite3.connect(settings.database_path) as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO repositories
|
|
(id, name, root, format_version, compression, encryption, active_data_key_id, state)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
"repository",
|
|
"repository",
|
|
str(settings.repository_roots[0]),
|
|
1,
|
|
"none",
|
|
"none",
|
|
"epoch",
|
|
"active",
|
|
),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO repository_data_key_epochs (id, repository_id, key_id, state)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
("epoch-row", "repository", "epoch", "active"),
|
|
)
|
|
|
|
with pytest.raises(
|
|
RuntimeError, match="cannot downgrade while repository data key metadata exists"
|
|
):
|
|
command.downgrade(migration, "0006_local_sources_only")
|
|
|
|
|
|
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 == []
|