bruno
95bc861cdb
- Router api_v2.py (~100 endpoints) : tokens, users, workspaces/members, collections, pages, proprietes, vues/dashboards, commentaires/mentions, notifications, favoris/tags/recents, partage/publish, historique, sprints, templates, export/import, forges, recherche FTS, admin, webhooks CRUD - Helpers api_v2_helpers.py : Bearer unifie (sha256/expires_at/extension_devices), scopes hierarchiques read<write<admin, pagination + X-Total-Count, ISO-8601, RFC 7807, idempotence, audit, rate-limit par token - Migration 20 : api_tokens.scopes/expires_at, webhook_deliveries, api_audit_log, idempotency_keys - main.py : handler d'erreurs unifie StarletteHTTPException, /docs + /redoc - config : PUBLIC_API_INSECURE_OK (dev only), API_V2_RATE_LIMIT_PER_TOKEN - OpenAPI docs/openapi-v2.json (402 chemins), tests/test_public_api_v2.py (24) - Docs : CHANGELOG (v6.2.0/6.2.1 clipper + v6.3.0), ROADMAP, API_GUIDE_V6, V6_Web_Clipper, README, ARCHITECTURE, /help - Suite complete 668 verte, ruff OK
292 lines
12 KiB
Python
292 lines
12 KiB
Python
"""FlowDeck — Public API v2 tests (v6.3.0).
|
|
|
|
Covers auth + scopes, tokens CRUD, pagination, filters, RFC7807 errors,
|
|
idempotency, webhooks CRUD, search, and the main resource wrappers.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
|
|
def _register_and_token(client, login="apiuser"):
|
|
"""Create a local account, log in, and mint a v2-capable token.
|
|
|
|
Uses the legacy /api/v1/token endpoint (session-authenticated) to obtain a
|
|
first token, then exercises v2.
|
|
"""
|
|
r = client.post("/auth/register", json={
|
|
"email": f"{login}@test.dev", "password": "secret123", "name": login,
|
|
})
|
|
assert r.status_code == 200, r.text
|
|
tok = client.post("/api/v1/token").json()["token"]
|
|
return {"Authorization": f"Bearer {tok}"}, tok
|
|
|
|
|
|
def _v2_token(client, headers, scopes="read,write", name="ci"):
|
|
r = client.post("/api/v2/tokens", json={"name": name, "scopes": scopes}, headers=headers)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["token"]
|
|
|
|
|
|
# ── Auth ──
|
|
|
|
def test_v2_requires_bearer(client):
|
|
r = client.get("/api/v2/collections")
|
|
assert r.status_code == 401
|
|
assert r.headers["content-type"].startswith("application/problem+json")
|
|
assert r.json()["status"] == 401
|
|
|
|
|
|
def test_v2_rejects_invalid_token(client):
|
|
r = client.get("/api/v2/users/me", headers={"Authorization": "Bearer nope"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_v2_fd_public_key_allowed_in_dev(client):
|
|
"""conftest sets PUBLIC_API_INSECURE_OK=true → dev fallback works."""
|
|
r = client.get("/api/v2/users/me", headers={"Authorization": "Bearer fd-public-key"})
|
|
# no users exist yet except seeded tester; may be 200 if a user exists
|
|
assert r.status_code in (200, 401)
|
|
|
|
|
|
def test_v2_me(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.get("/api/v2/users/me", headers=headers)
|
|
assert r.status_code == 200
|
|
d = r.json()
|
|
assert d["login"] == "[email protected]"
|
|
assert "password_hash" not in d
|
|
|
|
|
|
# ── Tokens CRUD + scopes ──
|
|
|
|
def test_v2_token_lifecycle(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.post("/api/v2/tokens", json={"name": "CI", "scopes": "read,write"}, headers=headers)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["token"].startswith("fd_")
|
|
assert data["scopes"] == "read,write"
|
|
tid = data["id"]
|
|
|
|
listing = client.get("/api/v2/tokens", headers=headers).json()["tokens"]
|
|
assert any(t["id"] == tid for t in listing)
|
|
# never expose hash
|
|
assert all("token_hash" not in t for t in listing)
|
|
|
|
rev = client.delete(f"/api/v2/tokens/{tid}", headers=headers)
|
|
assert rev.status_code == 200
|
|
|
|
|
|
def test_v2_invalid_scope_rejected(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.post("/api/v2/tokens", json={"name": "bad", "scopes": "superuser"}, headers=headers)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_v2_read_only_scope_blocks_write(client):
|
|
headers, _ = _register_and_token(client)
|
|
ro = _v2_token(client, headers, scopes="read", name="ro")
|
|
h_ro = {"Authorization": f"Bearer {ro}"}
|
|
r = client.post("/api/v2/collections", json={"name": "Nope"}, headers=h_ro)
|
|
assert r.status_code == 403
|
|
assert r.headers["content-type"].startswith("application/problem+json")
|
|
|
|
|
|
def test_v2_admin_scope_implies_write(client):
|
|
headers, _ = _register_and_token(client)
|
|
admin = _v2_token(client, headers, scopes="admin", name="admin")
|
|
h = {"Authorization": f"Bearer {admin}"}
|
|
r = client.post("/api/v2/collections", json={"name": "AdminDB"}, headers=h)
|
|
assert r.status_code == 201
|
|
|
|
|
|
# ── Collections / pages / properties / views ──
|
|
|
|
def test_v2_collections_crud(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.post("/api/v2/collections", json={"name": "DB1", "description": "d"}, headers=headers)
|
|
assert r.status_code == 201
|
|
cid = r.json()["id"]
|
|
|
|
got = client.get(f"/api/v2/collections/{cid}", headers=headers)
|
|
assert got.status_code == 200
|
|
assert got.json()["name"] == "DB1"
|
|
|
|
patched = client.patch(f"/api/v2/collections/{cid}", json={"name": "DB1b"}, headers=headers)
|
|
assert patched.status_code == 200
|
|
assert client.get(f"/api/v2/collections/{cid}", headers=headers).json()["name"] == "DB1b"
|
|
|
|
deleted = client.delete(f"/api/v2/collections/{cid}", headers=headers)
|
|
assert deleted.status_code == 200
|
|
|
|
|
|
def test_v2_pagination_and_total_count(client):
|
|
headers, _ = _register_and_token(client)
|
|
for i in range(3):
|
|
client.post("/api/v2/collections", json={"name": f"P{i}"}, headers=headers)
|
|
r = client.get("/api/v2/collections?limit=2&offset=0", headers=headers)
|
|
assert r.status_code == 200
|
|
assert r.headers["X-Total-Count"] == "3"
|
|
body = r.json()
|
|
assert len(body["collections"]) == 2
|
|
assert body["limit"] == 2 and body["offset"] == 0
|
|
|
|
|
|
def test_v2_pages_properties_views(client):
|
|
headers, _ = _register_and_token(client)
|
|
cid = client.post("/api/v2/collections", json={"name": "Work"}, headers=headers).json()["id"]
|
|
|
|
page = client.post(f"/api/v2/collections/{cid}/pages", json={"title": "Task 1"}, headers=headers)
|
|
assert page.status_code == 201
|
|
pid = page.json()["id"]
|
|
assert client.get(f"/api/v2/pages/{pid}", headers=headers).json()["title"] == "Task 1"
|
|
|
|
assert client.patch(f"/api/v2/pages/{pid}", json={"title": "Task 1b"}, headers=headers).status_code == 200
|
|
assert client.get(f"/api/v2/pages/{pid}", headers=headers).json()["title"] == "Task 1b"
|
|
|
|
prop = client.post(f"/api/v2/collections/{cid}/properties",
|
|
json={"name": "Status", "prop_type": "select", "options": ["Todo", "Done"]},
|
|
headers=headers)
|
|
assert prop.status_code == 200
|
|
assert client.get(f"/api/v2/collections/{cid}/properties", headers=headers).status_code == 200
|
|
|
|
assert client.get(f"/api/v2/collections/{cid}/views", headers=headers).status_code == 200
|
|
assert client.delete(f"/api/v2/pages/{pid}", headers=headers).status_code == 200
|
|
|
|
|
|
def test_v2_page_filter_and_sort(client):
|
|
headers, _ = _register_and_token(client)
|
|
cid = client.post("/api/v2/collections", json={"name": "F"}, headers=headers).json()["id"]
|
|
client.post(f"/api/v2/collections/{cid}/pages", json={"title": "Alpha"}, headers=headers)
|
|
client.post(f"/api/v2/collections/{cid}/pages", json={"title": "Beta"}, headers=headers)
|
|
r = client.get(f"/api/v2/collections/{cid}/pages?filter[title]=Alpha", headers=headers)
|
|
assert r.status_code == 200
|
|
assert len(r.json()["pages"]) == 1
|
|
|
|
|
|
# ── RFC7807 + idempotency ──
|
|
|
|
def test_v2_error_is_problem_json(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.get("/api/v2/pages/999999", headers=headers)
|
|
assert r.status_code == 404
|
|
assert r.headers["content-type"].startswith("application/problem+json")
|
|
body = r.json()
|
|
assert body["type"] and body["title"] and body["status"] == 404 and body["instance"]
|
|
|
|
|
|
def test_v2_idempotency_key(client):
|
|
headers, _ = _register_and_token(client)
|
|
h = {**headers, "Idempotency-Key": "abc-123"}
|
|
r1 = client.post("/api/v2/collections", json={"name": "Idem"}, headers=h)
|
|
r2 = client.post("/api/v2/collections", json={"name": "Idem"}, headers=h)
|
|
assert r1.status_code == 201 and r2.status_code == 201
|
|
assert r1.json()["id"] == r2.json()["id"]
|
|
|
|
|
|
# ── Search / notifications / favorites / tags / sharing / history ──
|
|
|
|
def test_v2_search(client):
|
|
headers, _ = _register_and_token(client)
|
|
client.post("/api/v2/collections", json={"name": "Searchable"}, headers=headers)
|
|
r = client.get("/api/v2/search?query=Search", headers=headers)
|
|
assert r.status_code == 200
|
|
assert any("Searchable" in x["title"] for x in r.json()["results"])
|
|
|
|
|
|
def test_v2_notifications(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.get("/api/v2/notifications", headers=headers)
|
|
assert r.status_code == 200
|
|
assert client.get("/api/v2/notifications/unread-count", headers=headers).status_code == 200
|
|
assert client.post("/api/v2/notifications/read-all", headers=headers).status_code == 200
|
|
|
|
|
|
def test_v2_favorites_tags_recents(client):
|
|
headers, _ = _register_and_token(client)
|
|
# A `pages` row is needed for favorites FK; collection pages aren't `pages`
|
|
# rows and v2 doesn't expose board page creation, so test tags + recents.
|
|
client.post("/api/v2/collections", json={"name": "Fav"}, headers=headers)
|
|
t = client.post("/api/v2/tags", json={"name": "urgent", "color": "#f00"}, headers=headers)
|
|
assert t.status_code == 200
|
|
assert any(x["name"] == "urgent" for x in client.get("/api/v2/tags", headers=headers).json()["tags"])
|
|
assert client.get("/api/v2/recents", headers=headers).status_code == 200
|
|
|
|
|
|
def test_v2_sharing_and_history(client):
|
|
headers, _ = _register_and_token(client)
|
|
# sharing requires a pages row; create via internal API with the session
|
|
r = client.post("/board/api/pages?section=Private&project=test/test")
|
|
if r.status_code != 200:
|
|
pytest.skip("board page creation unavailable")
|
|
pid = r.json()["id"]
|
|
sh = client.post(f"/api/v2/pages/{pid}/shares",
|
|
json={"email": "[email protected]", "permission": "view"}, headers=headers)
|
|
assert sh.status_code == 200
|
|
assert client.get(f"/api/v2/pages/{pid}/shares", headers=headers).status_code == 200
|
|
assert client.get(f"/api/v2/pages/{pid}/history", headers=headers).status_code == 200
|
|
|
|
|
|
# ── Webhooks CRUD ──
|
|
|
|
def test_v2_webhooks_crud(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.post("/api/v2/webhooks",
|
|
json={"url": "https://example.com/hook", "event": "page.created"}, headers=headers)
|
|
assert r.status_code == 200
|
|
wid = r.json()["id"]
|
|
assert any(w["id"] == wid for w in client.get("/api/v2/webhooks", headers=headers).json()["webhooks"])
|
|
assert client.post(f"/api/v2/webhooks/{wid}/test", headers=headers).status_code == 200
|
|
assert client.get(f"/api/v2/webhooks/{wid}/deliveries", headers=headers).status_code == 200
|
|
assert client.delete(f"/api/v2/webhooks/{wid}", headers=headers).status_code == 200
|
|
|
|
|
|
def test_v2_webhooks_invalid_url(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.post("/api/v2/webhooks", json={"url": "ftp://x", "event": "page.created"}, headers=headers)
|
|
assert r.status_code == 400
|
|
|
|
|
|
# ── Workspaces / sprints / templates ──
|
|
|
|
def test_v2_workspaces_crud(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.post("/api/v2/workspaces", json={"name": "Team"}, headers=headers)
|
|
assert r.status_code == 201
|
|
wid = r.json()["id"]
|
|
assert client.get(f"/api/v2/workspaces/{wid}", headers=headers).status_code == 200
|
|
assert client.get(f"/api/v2/workspaces/{wid}/members", headers=headers).status_code == 200
|
|
assert client.patch(f"/api/v2/workspaces/{wid}", json={"name": "Team2"}, headers=headers).status_code == 200
|
|
assert client.delete(f"/api/v2/workspaces/{wid}", headers=headers).status_code == 200
|
|
|
|
|
|
def test_v2_sprints(client):
|
|
headers, _ = _register_and_token(client)
|
|
cid = client.post("/api/v2/collections", json={"name": "S"}, headers=headers).json()["id"]
|
|
r = client.post(f"/api/v2/collections/{cid}/sprints",
|
|
json={"name": "Sprint 1", "start_date": "2026-01-01", "end_date": "2026-01-15"},
|
|
headers=headers)
|
|
assert r.status_code == 200
|
|
sid = r.json()["id"]
|
|
assert client.get(f"/api/v2/collections/{cid}/sprints", headers=headers).status_code == 200
|
|
assert client.get(f"/api/v2/sprints/{sid}/burndown", headers=headers).status_code == 200
|
|
assert client.delete(f"/api/v2/sprints/{sid}", headers=headers).status_code == 200
|
|
|
|
|
|
def test_v2_templates_database(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.get("/api/v2/templates/database", headers=headers)
|
|
assert r.status_code == 200
|
|
tpls = r.json()["templates"]
|
|
if tpls:
|
|
applied = client.post(f"/api/v2/templates/database/{tpls[0]['id']}/apply", json={}, headers=headers)
|
|
assert applied.status_code == 200
|
|
|
|
|
|
def test_v2_admin_requires_admin(client):
|
|
headers, _ = _register_and_token(client)
|
|
r = client.get("/api/v2/admin/users", headers=headers)
|
|
# first registered user is admin; this endpoint allows admin OR admin scope
|
|
assert r.status_code in (200, 403)
|