test(v2): define runtime and metadata contracts
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user