137 lines
4.5 KiB
Python
137 lines
4.5 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": client.cookies["backup_tool_csrf"]}
|
|
|
|
|
|
@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")
|
|
data_dir = tmp_path / "data"
|
|
data_dir.mkdir()
|
|
key = tmp_path / "master.key"
|
|
key.write_bytes(b"x" * 32)
|
|
key.chmod(0o600)
|
|
repositories = tmp_path / "repositories"
|
|
restore = tmp_path / "restore"
|
|
repositories.mkdir()
|
|
restore.mkdir()
|
|
settings = Settings(
|
|
data_dir=data_dir,
|
|
database_url=f"sqlite+aiosqlite:///{data_dir / 'db.sqlite'}",
|
|
repository_roots=(repositories,),
|
|
local_source_roots=(source_root,),
|
|
restore_roots=(restore,),
|
|
master_key_file=key,
|
|
)
|
|
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)
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://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()
|
|
data_dir = tmp_path / "data"
|
|
data_dir.mkdir()
|
|
key = tmp_path / "master.key"
|
|
key.write_bytes(b"x" * 32)
|
|
key.chmod(0o600)
|
|
repositories = tmp_path / "repositories"
|
|
restore = tmp_path / "restore"
|
|
repositories.mkdir()
|
|
restore.mkdir()
|
|
settings = Settings(
|
|
data_dir=data_dir,
|
|
database_url=f"sqlite+aiosqlite:///{data_dir / 'db.sqlite'}",
|
|
repository_roots=(repositories,),
|
|
local_source_roots=(allowed,),
|
|
restore_roots=(restore,),
|
|
master_key_file=key,
|
|
)
|
|
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:
|
|
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
|