90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
import httpx
|
|
import pytest
|
|
from backup_tool.api.app import create_app
|
|
from backup_tool.config import Settings
|
|
from backup_tool.db.models import RepositoryDataKeyEpoch
|
|
from sqlalchemy import select
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_create_and_inspect_allowlisted_repository(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
key = tmp_path / "key"
|
|
key.write_bytes(b"x" * 32)
|
|
key.chmod(0o600)
|
|
root = tmp_path / "repos"
|
|
root.mkdir()
|
|
settings = 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'}",
|
|
)
|
|
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=cast(Any, app))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
setup = await client.post(
|
|
"/api/v2/setup", json={"username": "admin", "password": "a secure password"}
|
|
)
|
|
assert setup.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, created.text
|
|
body = created.json()
|
|
assert body["name"] == "main"
|
|
assert body["format_version"] == 1
|
|
assert (root / "main" / "repository.json").is_file()
|
|
encrypted = await client.post(
|
|
"/api/v2/repositories",
|
|
json={
|
|
"name": "encrypted",
|
|
"relative_path": "encrypted",
|
|
"compression": "none",
|
|
"encryption": "aes-256-gcm",
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert encrypted.status_code == 201, encrypted.text
|
|
async with app.state.sessions() as db:
|
|
epochs = list(
|
|
(
|
|
await db.scalars(
|
|
select(RepositoryDataKeyEpoch).where(
|
|
RepositoryDataKeyEpoch.repository_id == encrypted.json()["id"]
|
|
)
|
|
)
|
|
).all()
|
|
)
|
|
assert len(epochs) == 1
|
|
assert epochs[0].state == "active"
|
|
got = await client.get(f"/api/v2/repositories/{body['id']}")
|
|
assert got.status_code == 200
|
|
changed = await client.patch(
|
|
f"/api/v2/repositories/{body['id']}",
|
|
json={"compression": "gzip"},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert changed.status_code == 409
|
|
await app.state.engine.dispose()
|