test(v2): cover repository safety failure paths

tests/integration/test_repository_safety.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import json
+import shutil
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import httpx
 import pytest
@@ -9,22 +11,26 @@ 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) -> Settings:
+
+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()
-    return Settings(
-        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: 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(**values)
 
 
 def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> None:
@@ -38,6 +44,41 @@ def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) ->
     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")
+
+
+@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_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) -> None:
     settings = settings_for(tmp_path)
     result = initialize(settings, "main", "none", "none")
@@ -48,6 +89,55 @@ def test_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) ->
         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"]
+        with patch(
+            "backup_tool.api.app.audit", new=AsyncMock(side_effect=RuntimeError("db failure"))
+        ):
+            response = await client.post(
+                "/api/v2/repositories",
+                json={
+                    "name": "main",
+                    "relative_path": "main",
+                    "compression": "none",
+                    "encryption": "none",
+                },
+                headers={"X-CSRF-Token": csrf},
+            )
+        assert response.status_code == 409
+        assert not (settings.repository_roots[0] / "main").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)
@@ -60,10 +150,7 @@ async def test_admin_can_list_and_inspect_repositories(tmp_path: Path) -> None:
         transport=httpx.ASGITransport(app=app), base_url="https://test"
     ) as client:
         assert (
-            await client.post(
-                "/api/v2/setup",
-                json={"username": "admin", "password": "a secure password"},
-            )
+            await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
         ).status_code == 201
         csrf = client.cookies["backup_tool_csrf"]
         created = await client.post(