151 lines
5.1 KiB
Python
151 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from tests.conftest import make_settings
|
|
|
|
|
|
async def setup_admin(
|
|
client, username: str = "admin", password: str = "correct horse battery staple"
|
|
):
|
|
return await client.post("/api/v2/setup", json={"username": username, "password": password})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_livez_is_public_and_request_ids_are_returned(app_client) -> None:
|
|
client, _ = app_client
|
|
response = await client.get("/livez")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "alive"}
|
|
assert UUID(response.headers["X-Request-ID"]).version == 7
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_readyz_rejects_unmigrated_database(tmp_path) -> None:
|
|
settings = make_settings(tmp_path)
|
|
app = importlib.import_module("backup_tool.api.app").create_app(settings)
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="https://testserver"
|
|
) as client:
|
|
response = await client.get("/readyz")
|
|
await app.state.engine.dispose()
|
|
assert response.status_code == 503
|
|
assert response.json()["code"] == "schema_not_current"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_readyz_is_truthful_before_and_after_setup(app_client) -> None:
|
|
client, _ = app_client
|
|
before = await client.get("/readyz")
|
|
assert before.status_code == 503
|
|
assert before.headers["content-type"].startswith("application/problem+json")
|
|
assert before.json()["code"] == "setup_required"
|
|
|
|
assert (await setup_admin(client)).status_code == 201
|
|
after = await client.get("/readyz")
|
|
assert after.status_code == 200
|
|
assert after.json()["status"] == "ready"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_protected_endpoint_uses_rfc9457_problem(app_client) -> None:
|
|
client, _ = app_client
|
|
response = await client.get("/api/v2/audit")
|
|
assert response.status_code == 401
|
|
body = response.json()
|
|
assert response.headers["content-type"].startswith("application/problem+json")
|
|
assert {"type", "title", "status", "detail", "instance", "code"} <= body.keys()
|
|
assert body["code"] == "authentication_required"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cursor_pagination_is_stable(app_client) -> None:
|
|
client, _ = app_client
|
|
assert (await setup_admin(client)).status_code == 201
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
headers = {"X-CSRF-Token": csrf}
|
|
for purpose in ("one", "two", "three"):
|
|
response = await client.post(
|
|
"/api/v2/admin/secrets",
|
|
json={"purpose": purpose, "value": f"secret-{purpose}"},
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
first = await client.get("/api/v2/audit", params={"limit": 2})
|
|
assert first.status_code == 200
|
|
first_body = first.json()
|
|
assert len(first_body["items"]) == 2
|
|
assert first_body["next_cursor"]
|
|
|
|
second = await client.get(
|
|
"/api/v2/audit", params={"limit": 2, "cursor": first_body["next_cursor"]}
|
|
)
|
|
assert second.status_code == 200
|
|
assert not (
|
|
{item["id"] for item in first_body["items"]}
|
|
& {item["id"] for item in second.json()["items"]}
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_etag_conflict_is_rejected(app_client) -> None:
|
|
client, _ = app_client
|
|
created = await setup_admin(client)
|
|
user_id = created.json()["id"]
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
current = await client.get(f"/api/v2/admin/users/{user_id}")
|
|
assert current.status_code == 200
|
|
etag = current.headers["ETag"]
|
|
|
|
updated = await client.patch(
|
|
f"/api/v2/admin/users/{user_id}",
|
|
json={"state": "active"},
|
|
headers={"If-Match": etag, "X-CSRF-Token": csrf},
|
|
)
|
|
assert updated.status_code == 200
|
|
|
|
stale = await client.patch(
|
|
f"/api/v2/admin/users/{user_id}",
|
|
json={"state": "active"},
|
|
headers={"If-Match": etag, "X-CSRF-Token": csrf},
|
|
)
|
|
assert stale.status_code == 412
|
|
assert stale.json()["code"] == "etag_mismatch"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idempotency_key_is_bound_to_request_digest(app_client) -> None:
|
|
client, _ = app_client
|
|
assert (await setup_admin(client)).status_code == 201
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
headers = {"X-CSRF-Token": csrf, "Idempotency-Key": "stable-key"}
|
|
expires = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
|
|
first = await client.post(
|
|
"/api/v2/auth/tokens",
|
|
json={"scopes": ["audit:read"], "expires_at": expires},
|
|
headers=headers,
|
|
)
|
|
assert first.status_code == 201
|
|
replay = await client.post(
|
|
"/api/v2/auth/tokens",
|
|
json={"scopes": ["audit:read"], "expires_at": expires},
|
|
headers=headers,
|
|
)
|
|
assert replay.status_code == 200
|
|
assert replay.json()["id"] == first.json()["id"]
|
|
assert replay.json()["token"] is None
|
|
|
|
mismatch = await client.post(
|
|
"/api/v2/auth/tokens",
|
|
json={"scopes": ["admin:write"], "expires_at": expires},
|
|
headers=headers,
|
|
)
|
|
assert mismatch.status_code == 409
|
|
assert mismatch.json()["code"] == "idempotency_mismatch"
|