129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
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", "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
|