91 lines
3.7 KiB
Python
91 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from backup_tool.api.app import create_app
|
|
from backup_tool.config import Settings
|
|
|
|
PASSWORD = "correct-horse-battery-staple"
|
|
|
|
|
|
async def login(client: httpx.AsyncClient) -> dict[str, str]:
|
|
response = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
|
|
assert response.status_code == 201
|
|
return {"X-CSRF-Token": response.json()["csrf_token"]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_source_probe_archive_and_repository_targeted_job(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "sources"
|
|
source_root.mkdir()
|
|
(source_root / "data.txt").write_text("contents")
|
|
settings = Settings(
|
|
data_dir=tmp_path / "data",
|
|
database_url=f"sqlite+aiosqlite:///{tmp_path / 'data' / 'db.sqlite'}",
|
|
repository_roots=(tmp_path / "repositories",),
|
|
local_source_roots=(source_root,),
|
|
restore_roots=(tmp_path / "restore",),
|
|
master_key_file=tmp_path / "master.key",
|
|
)
|
|
settings.master_key_file.write_bytes(b"x" * 32)
|
|
settings.master_key_file.chmod(0o600)
|
|
for root in (*settings.repository_roots, *settings.restore_roots):
|
|
root.mkdir(parents=True)
|
|
app = create_app(settings)
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
headers = await login(client)
|
|
repository = await client.post(
|
|
"/api/v2/repositories", json={"name": "repo", "relative_path": "main"}, headers=headers
|
|
)
|
|
assert repository.status_code == 201
|
|
source = await client.post(
|
|
"/api/v2/sources",
|
|
json={"name": "local", "kind": "local", "public_config": {"root": str(source_root)}},
|
|
headers=headers,
|
|
)
|
|
assert source.status_code == 201
|
|
source_id = source.json()["id"]
|
|
probe = await client.post(f"/api/v2/sources/{source_id}/probe", headers=headers)
|
|
assert probe.status_code == 200
|
|
assert probe.json()["entry_count"] == 1
|
|
job = await client.post(
|
|
"/api/v2/jobs",
|
|
json={"name": "job", "source_id": source_id, "repository_id": repository.json()["id"], "requested_mode": "full", "exclusions": ["*.tmp"], "retention": {}, "enabled": True, "allow_empty": False},
|
|
headers=headers,
|
|
)
|
|
assert job.status_code == 201
|
|
assert "destination_path" not in job.json()
|
|
archived = await client.delete(f"/api/v2/sources/{source_id}", headers=headers)
|
|
assert archived.status_code == 204
|
|
assert (await client.post(f"/api/v2/sources/{source_id}/probe", headers=headers)).status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_source_rejects_unallowlisted_root(tmp_path: Path) -> None:
|
|
allowed = tmp_path / "allowed"
|
|
allowed.mkdir()
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
key = tmp_path / "master.key"
|
|
key.write_bytes(b"x" * 32)
|
|
key.chmod(0o600)
|
|
settings = Settings(
|
|
data_dir=tmp_path / "data",
|
|
database_url=f"sqlite+aiosqlite:///{tmp_path / 'data' / 'db.sqlite'}",
|
|
repository_roots=(tmp_path / "repositories",),
|
|
local_source_roots=(allowed,),
|
|
restore_roots=(tmp_path / "restore",),
|
|
master_key_file=key,
|
|
)
|
|
settings.repository_roots[0].mkdir()
|
|
settings.restore_roots[0].mkdir()
|
|
app = create_app(settings)
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
|
|
headers = await login(client)
|
|
response = await client.post("/api/v2/sources", json={"name":"bad","kind":"local","public_config":{"root":str(outside)}}, headers=headers)
|
|
assert response.status_code == 422
|