86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
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
|
|
|
|
|
|
def settings_for(tmp_path: Path) -> 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,
|
|
)
|
|
|
|
|
|
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-*"))
|
|
|
|
|
|
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(result.root)
|
|
|
|
|
|
@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": "a secure 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()
|