Files

130 lines
5.1 KiB
Python

"""
REST API Endpoint Tests
Tests Anonymous intake, auth sessions, claims, and role-based permissions.
"""
import pytest
from starlette.testclient import TestClient
from thinkstorm.main import app
from thinkstorm.database import init_db
from thinkstorm.config import config
@pytest.fixture(scope="module", autouse=True)
def setup_app():
init_db()
def test_api_anonymous_submission():
client = TestClient(app)
resp = client.post("/api/ideas", json={"text": "Decentralized task scheduler with automated cron jobs and webhooks."})
assert resp.status_code == 201
data = resp.json()
assert data["id"].startswith("TS-")
assert data["lifecycle_state"] == "SUBMITTED"
def test_api_ideas_list():
client = TestClient(app)
# 1. Unauthenticated -> 401 Unauthorized
resp_unauth = client.get("/api/ideas")
assert resp_unauth.status_code == 401
# 2. Authenticated -> 200 OK list
login_resp = client.post("/api/auth/login", json={"username": "researcher", "password": "thinkstorm_user_2026"})
assert login_resp.status_code == 200
token = login_resp.json()["token"]
resp_auth = client.get("/api/ideas", headers={"Authorization": f"Bearer {token}"})
assert resp_auth.status_code == 200
data = resp_auth.json()
assert isinstance(data, list)
def test_api_auth_and_claim_flow():
client = TestClient(app)
# 1. Submit idea
sub_resp = client.post("/api/ideas", json={"text": "Real-time vector search indexer for personal notes."})
idea_id = sub_resp.json()["id"]
# Make idea available in db
from thinkstorm.database import get_db
with get_db() as conn:
conn.execute("UPDATE ideas SET lifecycle_state = 'AVAILABLE' WHERE id = ?", (idea_id,))
# 2. Try to claim anonymously -> 401
claim_unauth = client.post(f"/api/ideas/{idea_id}/claim")
assert claim_unauth.status_code == 401
# 3. Login as researcher
login_resp = client.post("/api/auth/login", json={"username": "researcher", "password": "thinkstorm_user_2026"})
assert login_resp.status_code == 200
token = login_resp.json()["token"]
# 4. Claim idea with session
claim_auth = client.post(f"/api/ideas/{idea_id}/claim", headers={"Authorization": f"Bearer {token}"})
assert claim_auth.status_code == 200
assert "claimed" in claim_auth.json()["message"].lower()
def test_admin_endpoints_permission():
client = TestClient(app)
# 1. Anonymously access admin -> 401 or 403
adm_unauth = client.get("/api/admin/prompts")
assert adm_unauth.status_code in (401, 403)
# 2. Login as admin
login_adm = client.post("/api/auth/login", json={"username": "admin", "password": config.admin_bootstrap_key})
assert login_adm.status_code == 200
adm_token = login_adm.json()["token"]
# 3. Admin access prompts
adm_prompts = client.get("/api/admin/prompts", headers={"Authorization": f"Bearer {adm_token}"})
assert adm_prompts.status_code == 200
prompts = adm_prompts.json()
assert len(prompts) > 0
# Full prompt template visible to admin
assert prompts[0]["system_prompt"] != "[REDACTED: ADMIN ONLY]"
def test_unauthenticated_cannot_browse_ideas():
client = TestClient(app)
# 1. Anonymous submission is allowed
sub_resp = client.post("/api/ideas", json={"text": "New idea submitted by anonymous visitor."})
assert sub_resp.status_code == 201
idea_id = sub_resp.json()["id"]
# 2. Unauthenticated GET /ideas (HTML Browse page) -> Redirect to /login
html_ideas_resp = client.get("/ideas", follow_redirects=False)
assert html_ideas_resp.status_code in (302, 307)
assert html_ideas_resp.headers.get("location") == "/login"
# 3. Unauthenticated GET /ideas/{idea_id} (HTML Dossier page) -> Redirect to /login
html_detail_resp = client.get(f"/ideas/{idea_id}", follow_redirects=False)
assert html_detail_resp.status_code in (302, 307)
assert html_detail_resp.headers.get("location") == "/login"
# 4. Unauthenticated GET /api/ideas (API Browse list) -> 401 Unauthorized
api_ideas_resp = client.get("/api/ideas")
assert api_ideas_resp.status_code == 401
assert "authentication required" in api_ideas_resp.json()["detail"].lower()
# 5. Unauthenticated GET /api/ideas/{idea_id} (API Dossier detail) -> 401 Unauthorized
api_detail_resp = client.get(f"/api/ideas/{idea_id}")
assert api_detail_resp.status_code == 401
assert "authentication required" in api_detail_resp.json()["detail"].lower()
# 6. Authenticated user can browse and view ideas
login_resp = client.post("/api/auth/login", json={"username": "researcher", "password": "thinkstorm_user_2026"})
assert login_resp.status_code == 200
token = login_resp.json()["token"]
auth_headers = {"Authorization": f"Bearer {token}"}
# Authenticated API list -> 200
auth_list_resp = client.get("/api/ideas", headers=auth_headers)
assert auth_list_resp.status_code == 200
assert isinstance(auth_list_resp.json(), list)
# Authenticated API detail -> 200
auth_detail_resp = client.get(f"/api/ideas/{idea_id}", headers=auth_headers)
assert auth_detail_resp.status_code == 200
assert auth_detail_resp.json()["id"] == idea_id