Frank Schwenk
88bdf6f34b
- Split monolithic app.py into server/api/ routers - Add Pydantic schemas for input validation - Add /agent/* API with separate AGENT_TOKEN auth - Replace static schema.sql with versioned migrations (datenerhaltend) - Add tools/backup_db.py for SQLite backups - Add Dockerfile HEALTHCHECK and compose.yml AGENT_TOKEN env - Add pytest tests for user and agent API - Update README with API docs and deployment safety notes - Bump version to 1.2.0
132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def test_health(client: TestClient) -> None:
|
|
resp = client.get("/api/health/")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
|
|
|
|
def test_login_ok(client: TestClient) -> None:
|
|
resp = client.post("/api/auth/pin", json={"pin": "1234"})
|
|
assert resp.status_code == 200
|
|
assert "token" in resp.json()
|
|
|
|
|
|
def test_login_wrong(client: TestClient) -> None:
|
|
resp = client.post("/api/auth/pin", json={"pin": "0000"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_config_requires_auth(client: TestClient) -> None:
|
|
resp = client.get("/api/config/")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_config_authenticated(client: TestClient, auth_headers: dict) -> None:
|
|
resp = client.get("/api/config/", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["timezone"] == "Europe/Berlin"
|
|
assert any(s["id"] == "morning" for s in data["slots"])
|
|
|
|
|
|
def test_today_and_log(client: TestClient, auth_headers: dict) -> None:
|
|
resp = client.get("/api/today", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
today = resp.json()
|
|
assert today["slots"][0]["id"] == "morning"
|
|
|
|
resp = client.post(
|
|
"/api/log",
|
|
headers=auth_headers,
|
|
json={"slot_id": "morning", "status": "taken"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["ok"] is True
|
|
assert data["stats"]["total_taken"] == 1
|
|
|
|
|
|
def test_log_unknown_slot(client: TestClient, auth_headers: dict) -> None:
|
|
resp = client.post(
|
|
"/api/log",
|
|
headers=auth_headers,
|
|
json={"slot_id": "nope", "status": "taken"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_snooze(client: TestClient, auth_headers: dict) -> None:
|
|
resp = client.post(
|
|
"/api/snooze",
|
|
headers=auth_headers,
|
|
json={"slot_id": "morning", "minutes": 15},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "snooze_until" in resp.json()
|
|
|
|
|
|
def test_snooze_invalid_minutes(client: TestClient, auth_headers: dict) -> None:
|
|
resp = client.post(
|
|
"/api/snooze",
|
|
headers=auth_headers,
|
|
json={"slot_id": "morning", "minutes": 42},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_agent_api_disabled_without_token(client: TestClient) -> None:
|
|
import server.settings as settings_module
|
|
|
|
original = settings_module.settings.agent_token
|
|
try:
|
|
settings_module.settings.agent_token = ""
|
|
resp = client.get("/agent/health", headers={"Authorization": "Bearer foo"})
|
|
assert resp.status_code == 503
|
|
finally:
|
|
settings_module.settings.agent_token = original
|
|
|
|
|
|
def test_agent_health(client: TestClient, agent_headers: dict) -> None:
|
|
resp = client.get("/agent/health", headers=agent_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "ok"
|
|
assert "db_version" in data
|
|
|
|
|
|
def test_agent_status(client: TestClient, agent_headers: dict) -> None:
|
|
resp = client.get("/agent/status", headers=agent_headers)
|
|
assert resp.status_code == 200
|
|
assert "today" in resp.json()
|
|
assert "stats" in resp.json()
|
|
|
|
|
|
def test_agent_log_and_history(client: TestClient, agent_headers: dict) -> None:
|
|
resp = client.post(
|
|
"/agent/log",
|
|
headers=agent_headers,
|
|
json={"slot_id": "morning", "status": "taken", "source": "agent"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "taken"
|
|
|
|
resp = client.get("/agent/history?days=7", headers=agent_headers)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["days"]) == 7
|
|
|
|
|
|
def test_agent_config(client: TestClient, agent_headers: dict) -> None:
|
|
resp = client.get("/agent/config", headers=agent_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["timezone"] == "Europe/Berlin"
|
|
|
|
|
|
def test_vapid_legacy_path(client: TestClient) -> None:
|
|
resp = client.get("/api/vapid-public-key")
|
|
assert resp.status_code == 200
|
|
assert "key" in resp.json()
|