from __future__ import annotations import json import shutil from pathlib import Path from unittest.mock import patch import httpx import pytest from backup_tool.api.app import create_app from backup_tool.config import Settings from backup_tool.repository import RepositoryError, initialize, inspect_repository PASSWORD = "a secure password" def settings_for(tmp_path: Path, **overrides: object) -> Settings: key = tmp_path / "key" key.write_bytes(b"x" * 32) key.chmod(0o600) root = tmp_path / "repos" root.mkdir() values: dict[str, object] = { "repository_roots": (root,), "local_source_roots": (tmp_path,), "restore_roots": (tmp_path,), "master_key_file": key, "data_dir": tmp_path, "database_url": f"sqlite+aiosqlite:///{tmp_path / 'db.sqlite'}", "min_free_bytes": 1, } values.update(overrides) return Settings.model_validate(values) def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> None: settings = settings_for(tmp_path) with ( patch("backup_tool.repository.os.replace", side_effect=OSError("disk failure")), pytest.raises(OSError), ): initialize(settings, "main", "none", "none") assert not (settings.repository_roots[0] / "main").exists() assert not list(settings.repository_roots[0].glob(".main.staging-*")) @pytest.mark.parametrize("relative_path", ["/absolute", "../escape"]) def test_initialization_rejects_absolute_and_traversal_paths( tmp_path: Path, relative_path: str ) -> None: with pytest.raises(RepositoryError, match="path"): initialize(settings_for(tmp_path), relative_path, "none", "none") def test_initialization_rejects_symlinked_destination_component(tmp_path: Path) -> None: settings = settings_for(tmp_path) outside = tmp_path / "outside" outside.mkdir() (settings.repository_roots[0] / "linked").symlink_to(outside, target_is_directory=True) with pytest.raises(RepositoryError, match="symlink"): initialize(settings, "linked/child", "none", "none") def test_initialization_supports_not_yet_created_nested_path(tmp_path: Path) -> None: settings = settings_for(tmp_path) result = initialize(settings, "nested/main", "none", "none") assert result.root == settings.repository_roots[0] / "nested" / "main" assert (result.root / "repository.json").is_file() @pytest.mark.parametrize( ("minimum", "usage"), [ ("min_free_bytes", shutil._ntuple_diskusage(total=100, used=99, free=1)), ("min_free_percent", shutil._ntuple_diskusage(total=100, used=96, free=4)), ], ) def test_initialization_rejects_insufficient_capacity( tmp_path: Path, minimum: str, usage: shutil._ntuple_diskusage ) -> None: settings = settings_for(tmp_path, **{minimum: 5}) with ( patch("backup_tool.repository.shutil.disk_usage", return_value=usage), pytest.raises(RepositoryError, match="capacity"), ): initialize(settings, "main", "none", "none") def test_initialization_emits_complete_repository_protocol_metadata( tmp_path: Path, ) -> None: result = initialize(settings_for(tmp_path), "main", "none", "none") try: payload = json.loads((result.root / "repository.json").read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise AssertionError("repository metadata was not readable JSON") from error assert set(payload) == { "repository_id", "format_version", "digest_algorithm", "compression", "encryption", "created_at", } assert payload["repository_id"][14] == "7" assert payload["digest_algorithm"] == "sha256" assert payload["encryption"] == {"mode": "none", "key_id": None} assert payload["created_at"].endswith("Z") def test_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) -> None: settings = settings_for(tmp_path) result = initialize(settings, "main", "none", "none") (result.root / "repository.json").write_text( '{"encryption":"none","compression":"none","format_version":1}\n' ) with pytest.raises(RepositoryError, match="canonical"): inspect_repository(settings, result.root) def test_repository_inspection_rejects_swapped_symlink_and_out_of_allowlist( tmp_path: Path, ) -> None: settings = settings_for(tmp_path) result = initialize(settings, "main", "none", "none") outside = tmp_path / "outside" outside.mkdir() (outside / "repository.json").write_text(json.dumps({"not": "a repository"}), encoding="utf-8") result.root.rename(settings.repository_roots[0] / "real-main") result.root.symlink_to(outside, target_is_directory=True) with pytest.raises(RepositoryError, match="escapes"): inspect_repository(settings, result.root) with pytest.raises(RepositoryError, match="escapes"): inspect_repository(settings, outside) @pytest.mark.asyncio async def test_database_failure_removes_published_repository(tmp_path: Path) -> None: settings = settings_for(tmp_path) app = create_app(settings) from backup_tool.db.models import Base async with app.state.engine.begin() as connection: await connection.run_sync(Base.metadata.create_all) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="https://test" ) as client: assert ( await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD}) ).status_code == 201 csrf = client.cookies["backup_tool_csrf"] first = await client.post( "/api/v2/repositories", json={ "name": "main", "relative_path": "first", "compression": "none", "encryption": "none", }, headers={"X-CSRF-Token": csrf}, ) assert first.status_code == 201 response = await client.post( "/api/v2/repositories", json={ "name": "main", "relative_path": "second", "compression": "none", "encryption": "none", }, headers={"X-CSRF-Token": csrf}, ) assert response.status_code == 409 assert not (settings.repository_roots[0] / "second").exists() await app.state.engine.dispose() @pytest.mark.asyncio async def test_admin_can_list_and_inspect_repositories(tmp_path: Path) -> None: settings = settings_for(tmp_path) app = create_app(settings) from backup_tool.db.models import Base async with app.state.engine.begin() as connection: await connection.run_sync(Base.metadata.create_all) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="https://test" ) as client: assert ( await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD}) ).status_code == 201 csrf = client.cookies["backup_tool_csrf"] created = await client.post( "/api/v2/repositories", json={ "name": "main", "relative_path": "main", "compression": "none", "encryption": "none", }, headers={"X-CSRF-Token": csrf}, ) assert created.status_code == 201 listed = await client.get("/api/v2/repositories") assert listed.status_code == 200 assert listed.json()["items"][0]["id"] == created.json()["id"] inspection = await client.get(f"/api/v2/repositories/{created.json()['id']}/inspection") assert inspection.status_code == 200 assert inspection.json()["format_version"] == 1 await app.state.engine.dispose()