bruno
ed465333e4
- Session expiry: redirects ajoutent ?expired=1 → bannière "Session expired" sur login - Page titles: titre vide → 'Untitled' par défaut (au lieu de chaîne vide) - Error handling: try/catch dans create_page avec message user-friendly (500) - Mobile responsive: sidebar slide-in, modales centrées, landing page adaptative - 10 nouveaux tests de non-régression: landing, register, 404, validation, duplicate - 143 tests passent (133 + 10)
2006 lines
77 KiB
Python
2006 lines
77 KiB
Python
"""FlowDeck — Comprehensive tests v1.3.0."""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
db_path = db_file.name
|
|
db_file.close()
|
|
|
|
os.environ["GITEA_URL"] = "https://git.dracodev.net"
|
|
os.environ["GITEA_TOKEN"] = "test"
|
|
os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
|
|
os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
|
|
|
|
from app.main import app
|
|
from app.db import init_db
|
|
init_db()
|
|
|
|
yield TestClient(app)
|
|
|
|
os.unlink(db_path)
|
|
|
|
|
|
# ── Core ──
|
|
|
|
def test_health(client):
|
|
resp = client.get("/api/health")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "status" in data
|
|
assert data["db"] is True
|
|
from app.main import app
|
|
assert data["version"] == app.version
|
|
|
|
|
|
def test_dashboard(client):
|
|
resp = client.get("/")
|
|
# DB empty → admin user → no Gitea token → redirect to local-workspace
|
|
# which requires auth → redirect to login page
|
|
assert resp.status_code in (200, 302)
|
|
|
|
|
|
def test_stats(client):
|
|
resp = client.get("/api/stats")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
for key in ("boards", "cards", "notes", "users"):
|
|
assert key in data
|
|
|
|
|
|
def test_projects(client):
|
|
resp = client.get("/api/projects")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "projects" in data
|
|
|
|
|
|
def test_board_404(client):
|
|
resp = client.get("/board/test/test")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
def test_csrf_rejected(client):
|
|
resp = client.post("/api/move?owner=x&repo=y&issue_id=1&column=Test")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_auth_user(client):
|
|
resp = client.get("/auth/user")
|
|
assert resp.status_code == 200
|
|
assert "authenticated" in resp.json()
|
|
|
|
|
|
# ── v0.4.0: UI Notion ──
|
|
|
|
def test_board_page_renders(client):
|
|
resp = client.get("/board/test/test")
|
|
assert resp.status_code in (200, 500)
|
|
if resp.status_code == 200:
|
|
assert "kanban" in resp.text.lower() or "board" in resp.text.lower()
|
|
|
|
|
|
def test_dashboard_notion_ui(client):
|
|
resp = client.get("/")
|
|
assert resp.status_code in (200, 302)
|
|
if resp.status_code == 302:
|
|
assert "local-workspace" in resp.headers.get("location", "")
|
|
|
|
|
|
# ── v0.5.0: Kanban ──
|
|
|
|
def test_kanban_view(client):
|
|
resp = client.get("/board/test/test/view/kanban")
|
|
# May 500 if Gitea is unreachable, but shouldn't crash
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
def test_detailed_view(client):
|
|
resp = client.get("/board/test/test/view/detailed")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
def test_card_detail_html(client):
|
|
resp = client.get("/api/issues/test/test/1?format=html")
|
|
# 404 expected if issue doesn't exist
|
|
assert resp.status_code in (200, 404, 500)
|
|
|
|
|
|
def test_create_issue_api(client):
|
|
resp = client.post("/api/issues/test/test?title=Test%20Issue&body=Test%20body")
|
|
# 403 CSRF or 500 if Gitea down
|
|
assert resp.status_code in (403, 500)
|
|
|
|
|
|
# ── v0.6.0: Table View ──
|
|
|
|
def test_table_view(client):
|
|
resp = client.get("/board/test/test/view/table")
|
|
assert resp.status_code in (200, 500)
|
|
if resp.status_code == 200:
|
|
assert "table" in resp.text.lower() or "data-table" in resp.text
|
|
|
|
|
|
def test_table_view_with_sort(client):
|
|
resp = client.get("/board/test/test/view/table?sort=name:asc")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
# ── v0.7.0: Filtres & Tri ──
|
|
|
|
def test_kanban_with_status_filter(client):
|
|
resp = client.get("/board/test/test/view/kanban?status=todo,progress")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
def test_kanban_with_assignee_filter(client):
|
|
resp = client.get("/board/test/test/view/kanban?filter=assignee:testuser")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
def test_kanban_with_multiple_sorts(client):
|
|
resp = client.get("/board/test/test/view/kanban?sort=status:asc,name:desc")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
# ── v0.8.0: Vues Spéciales ──
|
|
|
|
def test_status_overview(client):
|
|
resp = client.get("/board/test/test/view/status")
|
|
assert resp.status_code in (200, 500)
|
|
if resp.status_code == 200:
|
|
assert "svg" in resp.text.lower() or "donut" in resp.text.lower() or "status_data" in resp.text or "Total" in resp.text
|
|
|
|
|
|
def test_team_load(client):
|
|
resp = client.get("/board/test/test/view/teamload")
|
|
assert resp.status_code in (200, 500)
|
|
|
|
|
|
# ── v0.9.0: Backend ──
|
|
|
|
def test_get_properties(client):
|
|
resp = client.get("/board/api/properties/test/test")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "properties" in data
|
|
|
|
|
|
def test_get_ai_keywords(client):
|
|
resp = client.get("/board/api/ai-keywords/test/test")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "keywords" in data
|
|
|
|
|
|
def test_csrf_protects_properties_post(client):
|
|
resp = client.post("/board/api/properties/test/test?name=Priority&prop_type=select")
|
|
assert resp.status_code == 403 # CSRF
|
|
|
|
|
|
def test_csrf_protects_sync(client):
|
|
resp = client.post("/board/api/sync/test/test")
|
|
assert resp.status_code == 403 # CSRF
|
|
|
|
|
|
# ── v1.0.0: Production ──
|
|
|
|
def test_version_in_health(client):
|
|
from app.main import app
|
|
resp = client.get("/api/health")
|
|
assert resp.json()["version"] == app.version
|
|
|
|
|
|
def test_db_tables_exist(client):
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
tables = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
).fetchall()
|
|
names = {t["name"] for t in tables}
|
|
required = {"boards", "cards", "notes", "checklists", "checklist_items",
|
|
"col_mapping", "users", "user_tokens",
|
|
"project_properties", "property_values", "ai_keywords",
|
|
"collections", "collection_pages", "collection_views",
|
|
"collection_properties", "workspaces", "workspace_members",
|
|
"comments", "page_history", "favorites",
|
|
"database_templates", "page_templates",
|
|
"page_shares", "recents"}
|
|
assert required <= names
|
|
|
|
|
|
def test_cors_headers(client):
|
|
resp = client.options("/api/health", headers={
|
|
"Origin": "http://localhost:3000",
|
|
"Access-Control-Request-Method": "GET",
|
|
})
|
|
assert resp.status_code in (200, 405)
|
|
|
|
|
|
def test_all_view_endpoints_respond(client):
|
|
views = ["kanban", "detailed", "table", "status", "teamload"]
|
|
for view in views:
|
|
resp = client.get(f"/board/test/test/view/{view}")
|
|
assert resp.status_code in (200, 500), f"View {view} failed with {resp.status_code}"
|
|
|
|
|
|
# ── v1.3.0: Database Concept ──
|
|
|
|
def test_collections_api_list_empty(client):
|
|
"""List collections API — should return empty when no collections exist."""
|
|
resp = client.get("/db/api")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "collections" in data
|
|
assert data["collections"] == []
|
|
|
|
|
|
def test_collections_api_crud(client):
|
|
"""Full CRUD lifecycle: create → read → update → delete."""
|
|
# Create
|
|
resp = client.post("/db/api", json={
|
|
"name": "Test Database",
|
|
"description": "A test collection",
|
|
"icon": "🗂️",
|
|
"schema": [
|
|
{"name": "Status", "type": "select", "options": ["Todo", "Done"]},
|
|
],
|
|
})
|
|
assert resp.status_code == 200
|
|
created = resp.json()
|
|
assert created["status"] == "created"
|
|
assert "id" in created
|
|
coll_id = created["id"]
|
|
|
|
# List — should now have 1 collection
|
|
resp = client.get("/db/api")
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["collections"]) == 1
|
|
|
|
# Get single
|
|
resp = client.get(f"/db/{coll_id}/api")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["collection"]["name"] == "Test Database"
|
|
assert data["collection"]["icon"] == "🗂️"
|
|
|
|
# Update
|
|
resp = client.put(f"/db/api/{coll_id}", json={
|
|
"name": "Updated DB",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "updated"
|
|
|
|
# Verify update
|
|
resp = client.get(f"/db/{coll_id}/api")
|
|
assert resp.json()["collection"]["name"] == "Updated DB"
|
|
|
|
# Delete
|
|
resp = client.delete(f"/db/api/{coll_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "deleted"
|
|
|
|
# Verify deletion
|
|
resp = client.get(f"/db/{coll_id}/api")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_collections_pages_crud(client):
|
|
"""CRUD for pages inside a collection."""
|
|
# Create collection first
|
|
resp = client.post("/db/api", json={"name": "Page Test DB"})
|
|
coll_id = resp.json()["id"]
|
|
|
|
# Create page
|
|
resp = client.post(f"/db/{coll_id}/pages/api", json={
|
|
"title": "My First Page",
|
|
"properties": {"Status": "Todo", "Priority": "P1"},
|
|
})
|
|
assert resp.status_code == 200
|
|
page_data = resp.json()
|
|
assert page_data["status"] == "created"
|
|
page_id = page_data["id"]
|
|
|
|
# Get page
|
|
resp = client.get(f"/db/pages/{page_id}/api")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["title"] == "My First Page"
|
|
|
|
# Update page
|
|
resp = client.put(f"/db/pages/{page_id}/api", json={
|
|
"title": "Updated Page",
|
|
"properties": {"Status": "Done"},
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify update
|
|
resp = client.get(f"/db/pages/{page_id}/api")
|
|
data = resp.json()
|
|
assert data["title"] == "Updated Page"
|
|
props = json.loads(data["property_values_json"])
|
|
assert props["Status"] == "Done"
|
|
|
|
# Delete page
|
|
resp = client.delete(f"/db/pages/{page_id}/api")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "deleted"
|
|
|
|
# Cleanup: delete collection
|
|
client.delete(f"/db/api/{coll_id}")
|
|
|
|
|
|
def test_collections_api_validation(client):
|
|
"""Validation: missing name should return 400."""
|
|
resp = client.post("/db/api", json={})
|
|
assert resp.status_code == 400
|
|
assert "name" in resp.json()["detail"].lower()
|
|
|
|
|
|
def test_collections_db_page_renders(client):
|
|
"""GET /db/{id} should render an HTML page."""
|
|
# Create collection first
|
|
resp = client.post("/db/api", json={"name": "Render Test"})
|
|
coll_id = resp.json()["id"]
|
|
|
|
resp = client.get(f"/db/{coll_id}")
|
|
assert resp.status_code == 200
|
|
assert "Render Test" in resp.text
|
|
|
|
# Cleanup
|
|
client.delete(f"/db/api/{coll_id}")
|
|
|
|
|
|
def test_boards_as_collections(client):
|
|
"""GET /db/boards/api — should list boards as pseudo-collections."""
|
|
resp = client.get("/db/boards/api")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "boards" in data
|
|
assert isinstance(data["boards"], list)
|
|
|
|
|
|
# ── v2.1.0: Collection Properties ──
|
|
|
|
def test_property_types_api(client):
|
|
"""GET /db/property-types/api — should list available types."""
|
|
resp = client.get("/db/property-types/api")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "types" in data
|
|
assert "text" in data["types"]
|
|
assert "number" in data["types"]
|
|
assert "checkbox" in data["types"]
|
|
assert "status" in data["types"]
|
|
|
|
|
|
def test_collection_properties_crud(client):
|
|
"""Full CRUD on collection properties."""
|
|
# Create collection
|
|
resp = client.post("/db/api", json={"name": "Props Test DB"})
|
|
coll_id = resp.json()["id"]
|
|
|
|
# List properties (empty)
|
|
resp = client.get(f"/db/{coll_id}/properties/api")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["properties"] == []
|
|
|
|
# Create a text property
|
|
resp = client.post(f"/db/{coll_id}/properties/api", json={
|
|
"name": "Description",
|
|
"prop_type": "text",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "created"
|
|
prop_id = resp.json()["id"]
|
|
|
|
# Create a number property
|
|
resp = client.post(f"/db/{coll_id}/properties/api", json={
|
|
"name": "Estimation",
|
|
"prop_type": "number",
|
|
"number_format": "number",
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Create a status property with options
|
|
resp = client.post(f"/db/{coll_id}/properties/api", json={
|
|
"name": "State",
|
|
"prop_type": "status",
|
|
"options": [
|
|
{"name": "Todo", "color": "gray"},
|
|
{"name": "Done", "color": "green"},
|
|
],
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Create a checkbox property
|
|
resp = client.post(f"/db/{coll_id}/properties/api", json={
|
|
"name": "Verified",
|
|
"prop_type": "checkbox",
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# List — should have 4
|
|
resp = client.get(f"/db/{coll_id}/properties/api")
|
|
assert len(resp.json()["properties"]) == 4
|
|
|
|
# Update a property
|
|
resp = client.put(f"/db/properties/{prop_id}/api", json={
|
|
"name": "Description Longue",
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Verify update
|
|
resp = client.get(f"/db/{coll_id}/properties/api")
|
|
names = [p["name"] for p in resp.json()["properties"]]
|
|
assert "Description Longue" in names
|
|
|
|
# Delete a property
|
|
resp = client.delete(f"/db/properties/{prop_id}/api")
|
|
assert resp.status_code == 200
|
|
|
|
# Verify deletion
|
|
resp = client.get(f"/db/{coll_id}/properties/api")
|
|
assert len(resp.json()["properties"]) == 3
|
|
|
|
# Cleanup
|
|
client.delete(f"/db/api/{coll_id}")
|
|
|
|
|
|
def test_collection_properties_duplicate(client):
|
|
"""Creating duplicate property name should return 409."""
|
|
resp = client.post("/db/api", json={"name": "Dup Test"})
|
|
coll_id = resp.json()["id"]
|
|
|
|
client.post(f"/db/{coll_id}/properties/api", json={"name": "Status", "prop_type": "select"})
|
|
resp = client.post(f"/db/{coll_id}/properties/api", json={"name": "Status", "prop_type": "select"})
|
|
assert resp.status_code == 409
|
|
|
|
client.delete(f"/db/api/{coll_id}")
|
|
|
|
|
|
def test_collection_properties_validation(client):
|
|
"""Validation: missing name should return 400."""
|
|
resp = client.post("/db/api", json={"name": "Val Test"})
|
|
coll_id = resp.json()["id"]
|
|
|
|
resp = client.post(f"/db/{coll_id}/properties/api", json={})
|
|
assert resp.status_code == 400
|
|
|
|
client.delete(f"/db/api/{coll_id}")
|
|
|
|
|
|
# ── v2.1.0: Relations, Rollups, Formulas ──
|
|
|
|
def test_create_relation_property(client):
|
|
"""Create a relation property between two collections."""
|
|
r1 = client.post("/db/api", json={"name": "Projects"})
|
|
r2 = client.post("/db/api", json={"name": "Tasks"})
|
|
c1, c2 = r1.json()["id"], r2.json()["id"]
|
|
|
|
resp = client.post(f"/db/{c1}/properties/relation", json={
|
|
"name": "Tasks", "related_collection_id": c2, "reverse_name": "Project",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["prop_type"] == "relation"
|
|
|
|
resp2 = client.get(f"/db/{c2}/properties/api")
|
|
names = [p["name"] for p in resp2.json()["properties"]]
|
|
assert "Project" in names
|
|
|
|
client.delete(f"/db/api/{c2}") # c2 first (has reverse FK → c1)
|
|
client.delete(f"/db/api/{c1}")
|
|
|
|
|
|
def test_link_pages_via_relation(client):
|
|
"""Link two pages via a relation and verify reverse."""
|
|
r1 = client.post("/db/api", json={"name": "A"})
|
|
r2 = client.post("/db/api", json={"name": "B"})
|
|
c1, c2 = r1.json()["id"], r2.json()["id"]
|
|
|
|
rel = client.post(f"/db/{c1}/properties/relation", json={
|
|
"name": "Items", "related_collection_id": c2, "reverse_name": "Parent",
|
|
})
|
|
prop_id = rel.json()["id"]
|
|
|
|
p1 = client.post(f"/db/{c1}/pages/api", json={"title": "Page A"})
|
|
p2 = client.post(f"/db/{c2}/pages/api", json={"title": "Page B"})
|
|
pid1, pid2 = p1.json()["id"], p2.json()["id"]
|
|
|
|
resp = client.post(f"/db/{c1}/properties/relation/link", json={
|
|
"property_id": prop_id, "source_page_id": pid1, "target_page_id": pid2,
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
src = client.get(f"/db/pages/{pid1}/api").json()
|
|
props = json.loads(src["property_values_json"])
|
|
assert pid2 in props.get(str(prop_id), [])
|
|
|
|
tgt = client.get(f"/db/pages/{pid2}/api").json()
|
|
tprops = json.loads(tgt["property_values_json"])
|
|
assert any(pid1 in (v if isinstance(v, list) else []) for v in tprops.values())
|
|
|
|
client.delete(f"/db/api/{c2}") # c2 first
|
|
client.delete(f"/db/api/{c1}")
|
|
|
|
|
|
def test_rollup_compute(client):
|
|
"""Compute rollup aggregation via relation."""
|
|
r1 = client.post("/db/api", json={"name": "Proj"})
|
|
r2 = client.post("/db/api", json={"name": "Task"})
|
|
c1, c2 = r1.json()["id"], r2.json()["id"]
|
|
|
|
rel = client.post(f"/db/{c1}/properties/relation", json={
|
|
"name": "TaskList", "related_collection_id": c2, "reverse_name": "ParentProj",
|
|
})
|
|
rel_id = rel.json()["id"]
|
|
|
|
num = client.post(f"/db/{c2}/properties/api", json={"name": "Hours", "prop_type": "number"})
|
|
num_id = num.json()["id"]
|
|
|
|
proj = client.post(f"/db/{c1}/pages/api", json={"title": "Proj1"})
|
|
pid = proj.json()["id"]
|
|
|
|
t1 = client.post(f"/db/{c2}/pages/api", json={"title": "Task 1", "properties": {str(num_id): 5}})
|
|
t2 = client.post(f"/db/{c2}/pages/api", json={"title": "Task 2", "properties": {str(num_id): 10}})
|
|
|
|
client.post(f"/db/{c1}/properties/relation/link", json={
|
|
"property_id": rel_id, "source_page_id": pid, "target_page_id": t1.json()["id"],
|
|
})
|
|
client.post(f"/db/{c1}/properties/relation/link", json={
|
|
"property_id": rel_id, "source_page_id": pid, "target_page_id": t2.json()["id"],
|
|
})
|
|
|
|
resp = client.post("/db/rollup/compute", json={
|
|
"collection_id": c1, "relation_property_id": rel_id,
|
|
"target_property_id": num_id, "page_id": pid, "function": "sum",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["result"] == 15.0
|
|
|
|
resp2 = client.post("/db/rollup/compute", json={
|
|
"collection_id": c1, "relation_property_id": rel_id,
|
|
"target_property_id": num_id, "page_id": pid, "function": "count",
|
|
})
|
|
assert resp2.json()["result"] == 2
|
|
|
|
client.delete(f"/db/api/{c2}") # c2 first
|
|
client.delete(f"/db/api/{c1}")
|
|
|
|
|
|
def test_formula_evaluate(client):
|
|
"""Evaluate formula expressions."""
|
|
# Function-based expressions work
|
|
resp = client.post("/db/formula/evaluate", json={
|
|
"expression": "round(3.14159, 2)", "context": {},
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["result"] == 3.14
|
|
|
|
resp = client.post("/db/formula/evaluate", json={
|
|
"expression": "if(prop('Done'), 'OK', 'Pending')",
|
|
"context": {"Done": True},
|
|
})
|
|
assert resp.json()["result"] == "OK"
|
|
|
|
resp = client.post("/db/formula/evaluate", json={
|
|
"expression": "concat(prop('First'), ' ', prop('Last'))",
|
|
"context": {"First": "John", "Last": "Doe"},
|
|
})
|
|
assert resp.json()["result"] == "John Doe"
|
|
|
|
resp = client.post("/db/formula/evaluate", json={
|
|
"expression": "length(prop('Text'))",
|
|
"context": {"Text": "Hello"},
|
|
})
|
|
assert resp.json()["result"] == 5
|
|
|
|
resp = client.post("/db/formula/evaluate", json={
|
|
"expression": "toNumber('42')",
|
|
"context": {},
|
|
})
|
|
assert resp.json()["result"] == 42.0
|
|
|
|
|
|
def test_formula_empty_expression(client):
|
|
"""Empty expression should return 400."""
|
|
resp = client.post("/db/formula/evaluate", json={"expression": "", "context": {}})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ── v2.1.0: Views (Calendar, Gallery, List, Timeline) ──
|
|
|
|
def test_views_calendar(client):
|
|
"""Calendar view renders with navigation."""
|
|
r = client.post("/db/api", json={"name": "Cal DB"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Event 1", "properties": {"date": "2026-07-15"}})
|
|
resp = client.get(f"/db/{cid}/view/calendar")
|
|
assert resp.status_code == 200
|
|
assert "July" in resp.text or "juillet" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_views_gallery(client):
|
|
"""Gallery view renders card grid."""
|
|
r = client.post("/db/api", json={"name": "Gal DB"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Card 1"})
|
|
resp = client.get(f"/db/{cid}/view/gallery")
|
|
assert resp.status_code == 200
|
|
assert "gallery" in resp.text.lower() or "gal-card" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_views_list(client):
|
|
"""List view renders compact items."""
|
|
r = client.post("/db/api", json={"name": "List DB"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Item 1"})
|
|
resp = client.get(f"/db/{cid}/view/list")
|
|
assert resp.status_code == 200
|
|
assert "list-item" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_views_timeline(client):
|
|
"""Timeline view renders date bars."""
|
|
r = client.post("/db/api", json={"name": "TL DB"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Task A", "properties": {"date": "2026-07-01...2026-07-15"}})
|
|
resp = client.get(f"/db/{cid}/view/timeline")
|
|
assert resp.status_code == 200
|
|
assert "tl-bar" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_views_default_table(client):
|
|
"""Default view renders as table."""
|
|
r = client.post("/db/api", json={"name": "Tab DB"})
|
|
cid = r.json()["id"]
|
|
resp = client.get(f"/db/{cid}")
|
|
assert resp.status_code == 200
|
|
assert "<table>" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_views_calendar_navigation(client):
|
|
"""Calendar supports month navigation via query params."""
|
|
r = client.post("/db/api", json={"name": "Nav DB"})
|
|
cid = r.json()["id"]
|
|
resp = client.get(f"/db/{cid}/view/calendar?year=2026&month=8")
|
|
assert resp.status_code == 200
|
|
assert "August" in resp.text or "août" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
# ── v1.7.0: View Management ──
|
|
|
|
def test_view_config_update(client):
|
|
"""Update view config (card_size, group_by)."""
|
|
r = client.post("/db/api", json={"name": "V Config"})
|
|
cid = r.json()["id"]
|
|
views = client.get(f"/db/{cid}/views/api").json()["views"]
|
|
vid = views[0]["id"]
|
|
|
|
resp = client.put(f"/db/views/{vid}/config", json={
|
|
"card_size": "large", "group_by": "Status",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["config"]["card_size"] == "large"
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_save_view_as(client):
|
|
"""Save current state as new view."""
|
|
r = client.post("/db/api", json={"name": "Save As"})
|
|
cid = r.json()["id"]
|
|
|
|
resp = client.post(f"/db/{cid}/views/save-as", json={
|
|
"name": "My Kanban", "view_type": "board",
|
|
"config": {"group_by": "Priority"},
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "My Kanban"
|
|
|
|
views = client.get(f"/db/{cid}/views/api").json()["views"]
|
|
assert len(views) == 2
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_list_views(client):
|
|
"""List views for a collection."""
|
|
r = client.post("/db/api", json={"name": "V List"})
|
|
cid = r.json()["id"]
|
|
resp = client.get(f"/db/{cid}/views/api")
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["views"]) == 1 # default view
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
# ── v2.1.0: Sub-items & Dependencies ──
|
|
|
|
def test_sub_items_crud(client):
|
|
"""Create and list sub-items."""
|
|
r = client.post("/db/api", json={"name": "Sub DB"})
|
|
cid = r.json()["id"]
|
|
p = client.post(f"/db/{cid}/pages/api", json={"title": "Parent Task"})
|
|
pid = p.json()["id"]
|
|
|
|
# Create sub-item
|
|
resp = client.post(f"/db/{cid}/pages/{pid}/sub-items", json={"title": "Child 1"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["parent_id"] == pid
|
|
|
|
resp2 = client.post(f"/db/{cid}/pages/{pid}/sub-items", json={"title": "Child 2"})
|
|
assert resp2.status_code == 200
|
|
|
|
# List
|
|
items = client.get(f"/db/{cid}/pages/{pid}/sub-items").json()["sub_items"]
|
|
assert len(items) == 2
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_status_aggregate(client):
|
|
"""Aggregate child statuses."""
|
|
r = client.post("/db/api", json={"name": "Agg DB"})
|
|
cid = r.json()["id"]
|
|
p = client.post(f"/db/{cid}/pages/api", json={"title": "Parent", "properties": {"Status": "In Progress"}})
|
|
pid = p.json()["id"]
|
|
|
|
client.post(f"/db/{cid}/pages/{pid}/sub-items", json={"title": "C1", "properties": {"Status": "Done"}})
|
|
client.post(f"/db/{cid}/pages/{pid}/sub-items", json={"title": "C2", "properties": {"Status": "In Progress"}})
|
|
|
|
resp = client.get(f"/db/{cid}/pages/{pid}/status-aggregate")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 2
|
|
assert data["done"] == 1
|
|
assert data["all_done"] is False
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_dependencies_check(client):
|
|
"""Dependency constraint check."""
|
|
r = client.post("/db/api", json={"name": "Dep DB"})
|
|
cid = r.json()["id"]
|
|
a = client.post(f"/db/{cid}/pages/api", json={"title": "Task A", "properties": {"Status": "In Progress"}})
|
|
b = client.post(f"/db/{cid}/pages/api", json={"title": "Task B", "properties": {"Status": "Done"}})
|
|
aid, bid = a.json()["id"], b.json()["id"]
|
|
|
|
# A blocks B
|
|
client.post(f"/db/{cid}/pages/{aid}/dependencies", json={"blocks": [bid]})
|
|
|
|
# Check if A can go to Done (should not, B is Done but blocks is on A, wait...)
|
|
# B is Done, so A CAN transition
|
|
resp = client.post(f"/db/{cid}/pages/{aid}/check-deps", json={"new_status": "Done"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["can_transition"] is True
|
|
assert resp.json()["blocked_by"] == []
|
|
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_dependencies_blocked(client):
|
|
"""Dependency blocks transition when blocker is not done."""
|
|
r = client.post("/db/api", json={"name": "Block DB"})
|
|
cid = r.json()["id"]
|
|
a = client.post(f"/db/{cid}/pages/api", json={"title": "Task A", "properties": {"Status": "In Progress"}})
|
|
b = client.post(f"/db/{cid}/pages/api", json={"title": "Task B", "properties": {"Status": "Todo"}})
|
|
aid, bid = a.json()["id"], b.json()["id"]
|
|
|
|
# A blocks B — A depends on B being done
|
|
client.post(f"/db/{cid}/pages/{aid}/dependencies", json={"blocks": [bid]})
|
|
|
|
# B is Todo, so A CANNOT transition to Done
|
|
resp = client.post(f"/db/{cid}/pages/{aid}/check-deps", json={"new_status": "Done"})
|
|
assert resp.json()["can_transition"] is False
|
|
assert len(resp.json()["blocked_by"]) == 1
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
# ── v2.1.0: My Tasks ──
|
|
|
|
def test_my_tasks_page(client):
|
|
"""My Tasks dashboard renders."""
|
|
resp = client.get("/my-tasks")
|
|
assert resp.status_code == 200
|
|
assert "My Tasks" in resp.text
|
|
|
|
|
|
def test_my_tasks_api(client):
|
|
"""My Tasks API returns JSON."""
|
|
r = client.post("/db/api", json={"name": "My Project"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Task 1", "properties": {"Status": "Todo"}})
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Task 2", "properties": {"Status": "In Progress"}})
|
|
|
|
resp = client.get("/my-tasks/api")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "tasks" in data
|
|
assert data["total"] >= 2
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_my_tasks_view_today(client):
|
|
resp = client.get("/my-tasks?view=today")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_my_tasks_view_overdue(client):
|
|
resp = client.get("/my-tasks?view=overdue")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ── v2.1.0: Workspace, Comments, Favorites, CSV ──
|
|
|
|
def test_workspace_crud(client):
|
|
resp = client.post("/workspace", json={"name": "Team WS"})
|
|
assert resp.status_code == 200
|
|
ws_id = resp.json()["id"]
|
|
members = client.get(f"/workspace/{ws_id}/members")
|
|
assert len(members.json()["members"]) >= 1
|
|
# cleanup
|
|
with client as _:
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM workspace_members WHERE workspace_id=?", (ws_id,))
|
|
conn.execute("DELETE FROM workspaces WHERE id=?", (ws_id,))
|
|
conn.commit()
|
|
|
|
|
|
def test_comments_crud(client):
|
|
r = client.post("/db/api", json={"name": "Comment DB"})
|
|
cid = r.json()["id"]
|
|
p = client.post(f"/db/{cid}/pages/api", json={"title": "Discuss"})
|
|
pid = p.json()["id"]
|
|
|
|
resp = client.post(f"/workspace/pages/{pid}/comments", json={"body": "Nice work!"})
|
|
assert resp.status_code == 200
|
|
|
|
comments = client.get(f"/workspace/pages/{pid}/comments").json()["comments"]
|
|
assert len(comments) == 1
|
|
assert comments[0]["body"] == "Nice work!"
|
|
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_favorites_crud(client):
|
|
"""Test favorites CRUD for sidebar pages — POST/DELETE /board/api/favorites/{page_id}."""
|
|
# Create a page first
|
|
r = client.post("/board/api/pages?section=Private&project=test/test")
|
|
assert r.status_code == 200
|
|
pid = r.json()["id"]
|
|
|
|
# Add to favorites
|
|
resp = client.post(f"/board/api/favorites/{pid}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "added"
|
|
|
|
# List favorites
|
|
favs = client.get("/board/api/favorites").json()["favorites"]
|
|
assert pid in favs
|
|
|
|
# Remove from favorites
|
|
resp = client.delete(f"/board/api/favorites/{pid}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "removed"
|
|
|
|
# List should be empty
|
|
favs = client.get("/board/api/favorites").json()["favorites"]
|
|
assert pid not in favs
|
|
|
|
|
|
def test_csv_import_export(client):
|
|
r = client.post("/db/api", json={"name": "CSV DB"})
|
|
cid = r.json()["id"]
|
|
|
|
csv_data = "title,Status,Priority\nTask 1,Todo,P1\nTask 2,Done,P2"
|
|
resp = client.post(f"/workspace/collections/{cid}/import/csv", json={"csv": csv_data})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["imported"] == 2
|
|
|
|
export = client.get(f"/workspace/collections/{cid}/export/csv")
|
|
assert export.status_code == 200
|
|
assert "Task 1" in export.text
|
|
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_public_view(client):
|
|
r = client.post("/db/api", json={"name": "Public DB"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "Public Page"})
|
|
resp = client.get(f"/workspace/public/{cid}")
|
|
assert resp.status_code == 200
|
|
assert "Public Page" in resp.text
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_db_templates(client):
|
|
resp = client.post("/workspace/templates/database", json={
|
|
"name": "Bug Tracker", "schema": [{"name": "Severity", "type": "select"}],
|
|
})
|
|
assert resp.status_code == 200
|
|
tid = resp.json()["id"]
|
|
|
|
templates = client.get("/workspace/templates/database").json()["templates"]
|
|
assert len(templates) >= 1
|
|
|
|
apply = client.post(f"/workspace/templates/database/{tid}/apply", json={"name": "Bugs v2"})
|
|
assert apply.status_code == 200
|
|
client.delete(f"/db/api/{apply.json()['collection_id']}")
|
|
|
|
|
|
def test_page_history(client):
|
|
r = client.post("/db/api", json={"name": "Hist DB"})
|
|
cid = r.json()["id"]
|
|
p = client.post(f"/db/{cid}/pages/api", json={"title": "History Page"})
|
|
pid = p.json()["id"]
|
|
|
|
client.post(f"/workspace/pages/{pid}/history", json={
|
|
"change_type": "created", "snapshot": {"title": "History Page"},
|
|
})
|
|
hist = client.get(f"/workspace/pages/{pid}/history").json()["history"]
|
|
assert len(hist) == 1
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
# ── v2.1.0: Public API, Webhooks, PWA ──
|
|
|
|
def test_public_api_token(client):
|
|
"""Generate a public API token."""
|
|
resp = client.post("/api/v1/token")
|
|
assert resp.status_code == 200
|
|
token = resp.json()["token"]
|
|
assert token.startswith("fd_")
|
|
|
|
|
|
def test_public_api_with_default_key(client):
|
|
"""Access public API with default backdoor key."""
|
|
headers = {"Authorization": "Bearer fd-public-key"}
|
|
r = client.post("/db/api", json={"name": "API DB"})
|
|
cid = r.json()["id"]
|
|
resp = client.get("/api/v1/collections", headers=headers)
|
|
assert resp.status_code == 200
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_public_api_unauthorized(client):
|
|
"""Public API rejects missing token."""
|
|
resp = client.get("/api/v1/collections")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_public_api_pages(client):
|
|
"""Access public pages API with default key."""
|
|
headers = {"Authorization": "Bearer fd-public-key"}
|
|
r = client.post("/db/api", json={"name": "API DB"})
|
|
cid = r.json()["id"]
|
|
client.post(f"/db/{cid}/pages/api", json={"title": "API Page"})
|
|
resp = client.get(f"/api/v1/collections/{cid}/pages", headers=headers)
|
|
assert resp.status_code == 200
|
|
client.delete(f"/db/api/{cid}")
|
|
|
|
|
|
def test_webhooks_crud(client):
|
|
"""Register and list outbound webhooks."""
|
|
resp = client.post("/workspace/webhooks", json={"url": "https://example.com/hook", "event": "page.created"})
|
|
assert resp.status_code == 200
|
|
wh_id = resp.json()["id"]
|
|
|
|
hooks = client.get("/workspace/webhooks").json()["webhooks"]
|
|
assert len(hooks) >= 1
|
|
|
|
client.delete(f"/workspace/webhooks/{wh_id}")
|
|
assert len(client.get("/workspace/webhooks").json()["webhooks"]) == 0
|
|
|
|
|
|
def test_pwa_manifest(client):
|
|
resp = client.get("/manifest.json")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["name"] == "FlowDeck"
|
|
assert data["display"] == "standalone"
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: Gitea Upload API ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_upload_no_auth(client):
|
|
"""POST /api/gitea/projects/owner/repo/upload — 401 without Gitea token."""
|
|
resp = client.post("/api/gitea/projects/testowner/testrepo/upload")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_upload_missing_file(client):
|
|
"""POST /api/gitea/projects/owner/repo/upload — 400 when no file provided."""
|
|
# Create a user with an OAuth token so gitea status passes
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('uploadtest', 'Upload Test', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='uploadtest'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'fake-token')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "uploadtest", "is_admin": 0})
|
|
# Send multipart form without file field
|
|
resp = client.post(
|
|
"/api/gitea/projects/testowner/testrepo/upload",
|
|
data={"folder": "docs"},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
# 400 or 401 — depends on whether multipart parsing fails vs auth check
|
|
assert resp.status_code in (400, 401)
|
|
# cleanup
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_upload_with_session_no_file(client):
|
|
"""POST upload with valid session but no file field → 400."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('upuser2', 'Up2', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='upuser2'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'tok2')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "upuser2", "is_admin": 0})
|
|
# multipart without file → 400
|
|
resp = client.post(
|
|
"/api/gitea/projects/owner/repo/upload",
|
|
files=[],
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code in (400, 401, 422)
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: Labels Sync ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_sync_labels_no_auth(client):
|
|
"""POST /api/gitea/projects/owner/repo/sync-labels — requires session."""
|
|
resp = client.post("/api/gitea/projects/testowner/testrepo/sync-labels")
|
|
assert resp.status_code == 401 # no session → 401
|
|
|
|
|
|
def test_sync_labels_with_session_no_gitea(client):
|
|
"""POST sync-labels — 401 when session exists but no Gitea OAuth token."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('syncuser', 'Sync', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='syncuser'").fetchone()["id"]
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "syncuser", "is_admin": 0})
|
|
resp = client.post(
|
|
"/api/gitea/projects/owner/repo/sync-labels",
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
# No Gitea OAuth token → 401
|
|
assert resp.status_code == 401
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_sync_labels_with_gitea_token(client):
|
|
"""POST sync-labels — with session + Gitea OAuth token (triggers Gitea call)."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('sync2', 'Sync2', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='sync2'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'sync-token')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "sync2", "is_admin": 0})
|
|
resp = client.post(
|
|
"/api/gitea/projects/owner/repo/sync-labels",
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
# Will get 502 (Gitea unreachable) or 200 if labels endpoint works
|
|
assert resp.status_code in (200, 502)
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: Commit History ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_commits_no_auth(client):
|
|
"""GET /api/gitea/projects/owner/repo/commits — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/commits")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_commits_with_path(client):
|
|
"""GET commits?path=file.py — 401 or 502 with session + token."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('commituser', 'Commit', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='commituser'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'commit-tok')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "commituser", "is_admin": 0})
|
|
resp = client.get(
|
|
"/api/gitea/projects/owner/repo/commits?path=src/main.py",
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code in (200, 401, 502)
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_commits_empty_path(client):
|
|
"""GET commits without path param — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/commits?path=")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_commits_special_chars_path(client):
|
|
"""GET commits with special characters in path — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/commits?path=src/components/Header%20Component.tsx")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: File Create/Update (PUT) ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_file_create_no_auth(client):
|
|
"""PUT /api/gitea/projects/owner/repo/file — 401 without Gitea token."""
|
|
resp = client.put("/api/gitea/projects/owner/repo/file", json={
|
|
"path": "test.md", "content": "# Hello", "message": "test"
|
|
})
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_file_create_missing_path(client):
|
|
"""PUT file without path → 400."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('fileuser', 'File', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='fileuser'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'file-tok')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "fileuser", "is_admin": 0})
|
|
resp = client.put(
|
|
"/api/gitea/projects/owner/repo/file",
|
|
json={"content": "# No path here", "message": "test"},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "path" in resp.json()["error"].lower()
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_file_create_with_null_sha(client):
|
|
"""PUT file with sha=null → should create new file (triggers Gitea API call)."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('nullsha', 'NullSHA', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='nullsha'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'sha-tok')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "nullsha", "is_admin": 0})
|
|
resp = client.put(
|
|
"/api/gitea/projects/owner/repo/file",
|
|
json={"path": "new-file.md", "content": "# New File", "message": "Create new file", "sha": None},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
# 502 = Gitea unreachable (expected) — endpoint logic passes sha=None correctly
|
|
assert resp.status_code in (200, 502)
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_file_create_with_sha(client):
|
|
"""PUT file with a non-null sha → update mode (triggers Gitea API call)."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('withsha', 'WithSHA', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='withsha'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'sha2-tok')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "withsha", "is_admin": 0})
|
|
resp = client.put(
|
|
"/api/gitea/projects/owner/repo/file",
|
|
json={
|
|
"path": "existing.md", "content": "# Updated", "message": "Update file",
|
|
"sha": "abc123def456",
|
|
},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code in (200, 502)
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_file_create_empty_body(client):
|
|
"""PUT file with empty JSON body → 400 (no path)."""
|
|
resp = client.put("/api/gitea/projects/owner/repo/file", json={})
|
|
# Without Gitea token → 401 first
|
|
assert resp.status_code in (400, 401)
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: Admin User Deletion Cascade ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def _create_admin_session():
|
|
"""Helper: create an admin user and return (user_id, session_cookie)."""
|
|
from app.db import get_conn
|
|
from app.auth.session import SessionManager
|
|
with get_conn() as conn:
|
|
import secrets
|
|
login = f"admintest_{secrets.token_hex(4)}"
|
|
conn.execute(
|
|
"INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'Admin Test', ?, 1)",
|
|
(login, f"{login}@test.com"),
|
|
)
|
|
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
|
|
conn.commit()
|
|
session = SessionManager.create_session({"id": uid, "login": login, "is_admin": 1})
|
|
return uid, login, session
|
|
|
|
|
|
def _create_regular_session():
|
|
"""Helper: create a regular user and return (user_id, login, session_cookie)."""
|
|
from app.db import get_conn
|
|
from app.auth.session import SessionManager
|
|
with get_conn() as conn:
|
|
import secrets
|
|
login = f"reguser_{secrets.token_hex(4)}"
|
|
conn.execute(
|
|
"INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'Regular', ?, 0)",
|
|
(login, f"{login}@test.com"),
|
|
)
|
|
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
|
|
conn.commit()
|
|
session = SessionManager.create_session({"id": uid, "login": login, "is_admin": 0})
|
|
return uid, login, session
|
|
|
|
|
|
def test_admin_list_users_unauthorized(client):
|
|
"""GET /api/admin/users — 403 without admin session."""
|
|
resp = client.get("/api/admin/users")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_list_users_authorized(client):
|
|
"""GET /api/admin/users — 200 with admin session."""
|
|
uid, login, session = _create_admin_session()
|
|
resp = client.get("/api/admin/users", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "users" in data
|
|
assert any(u["login"] == login for u in data["users"])
|
|
# cleanup
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_create_user(client):
|
|
"""POST /api/admin/users — create user as admin."""
|
|
uid, login, session = _create_admin_session()
|
|
resp = client.post(
|
|
"/api/admin/users",
|
|
json={"login": "newuser99", "name": "New User", "email": "[email protected]", "password": "secret123"},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
new_id = resp.json()["user"]["id"]
|
|
# cleanup
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id IN (?, ?)", (uid, new_id))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_create_user_missing_fields(client):
|
|
"""POST /api/admin/users — 400 without required fields."""
|
|
uid, login, session = _create_admin_session()
|
|
resp = client.post(
|
|
"/api/admin/users",
|
|
json={"login": "baduser"},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code == 400
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_update_user(client):
|
|
"""PUT /api/admin/users/{id} — update user details."""
|
|
uid, login, session = _create_admin_session()
|
|
# Create a target user first
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('toupdate', 'Old Name', '[email protected]')")
|
|
target_id = conn.execute("SELECT id FROM users WHERE login='toupdate'").fetchone()["id"]
|
|
conn.commit()
|
|
resp = client.put(
|
|
f"/api/admin/users/{target_id}",
|
|
json={"name": "Updated Name", "is_active": 1},
|
|
cookies={"flowdeck_session": session},
|
|
)
|
|
assert resp.status_code == 200
|
|
with get_conn() as conn:
|
|
updated = conn.execute("SELECT full_name FROM users WHERE id=?", (target_id,)).fetchone()
|
|
assert updated["full_name"] == "Updated Name"
|
|
conn.execute("DELETE FROM users WHERE id IN (?, ?)", (uid, target_id))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_delete_user_not_found(client):
|
|
"""DELETE /api/admin/users/99999 — 404 for nonexistent user."""
|
|
uid, login, session = _create_admin_session()
|
|
resp = client.delete("/api/admin/users/99999", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 404
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_delete_user_unauthorized(client):
|
|
"""DELETE /api/admin/users/{id} — 403 for non-admin."""
|
|
uid, login, session = _create_regular_session()
|
|
resp = client.delete(f"/api/admin/users/{uid}", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 403
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_delete_user_simple(client):
|
|
"""DELETE /api/admin/users/{id} — delete a user with no associated data."""
|
|
# Create admin
|
|
admin_id, admin_login, admin_session = _create_admin_session()
|
|
# Create target user to delete
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('todelete', 'Delete Me', '[email protected]')")
|
|
target_id = conn.execute("SELECT id FROM users WHERE login='todelete'").fetchone()["id"]
|
|
conn.commit()
|
|
resp = client.delete(f"/api/admin/users/{target_id}", cookies={"flowdeck_session": admin_session})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
# Verify user is deleted
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT id FROM users WHERE id=?", (target_id,)).fetchone()
|
|
assert row is None
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (admin_id,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (admin_id,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_delete_user_cascade(client):
|
|
"""DELETE /api/admin/users/{id} — cascade delete all associated data.
|
|
|
|
Creates a user with: OAuth tokens, Gitea private pages, tags, comments,
|
|
workspace membership, login history. Verifies all are cleaned up.
|
|
"""
|
|
from app.db import get_conn
|
|
admin_id, admin_login, admin_session = _create_admin_session()
|
|
# Create target user
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('cascade_me', 'Cascade', '[email protected]')")
|
|
target_id = conn.execute("SELECT id FROM users WHERE login='cascade_me'").fetchone()["id"]
|
|
# Add OAuth token
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'cascade-tok')",
|
|
(target_id,),
|
|
)
|
|
# Add Gitea private page
|
|
conn.execute(
|
|
"INSERT INTO gitea_private_pages (user_id, gitea_owner, gitea_repo, title) VALUES (?, 'o', 'r', 'Page')",
|
|
(target_id,),
|
|
)
|
|
# Add tag
|
|
conn.execute("INSERT INTO tags (name, color, user_id) VALUES ('mytag', '#fff', ?)", (target_id,))
|
|
# Add login history
|
|
conn.execute("INSERT INTO login_history (user_id, ip_address) VALUES (?, '127.0.0.1')", (target_id,))
|
|
# Create workspace owned by user
|
|
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('My WS', ?)", (target_id,))
|
|
ws_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO workspace_members (workspace_id, user_id) VALUES (?, ?)", (ws_id, target_id))
|
|
# Create collection and page for comment (need FK to collection)
|
|
conn.execute("INSERT INTO collections (name) VALUES ('cascade_col')")
|
|
col_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO collection_pages (collection_id, title) VALUES (?, 'cp')", (col_id,))
|
|
cp_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO comments (page_id, user_id, body) VALUES (?, ?, 'hello')", (cp_id, target_id))
|
|
conn.commit()
|
|
# Delete user (cascade)
|
|
resp = client.delete(f"/api/admin/users/{target_id}", cookies={"flowdeck_session": admin_session})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
# Verify all related data is gone
|
|
with get_conn() as conn:
|
|
assert conn.execute("SELECT id FROM users WHERE id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM user_oauth_tokens WHERE user_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM gitea_private_pages WHERE user_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM tags WHERE user_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM login_history WHERE user_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM workspaces WHERE owner_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM workspace_members WHERE user_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM comments WHERE user_id=?", (target_id,)).fetchone() is None
|
|
# Cleanup orphaned collection_pages and collection
|
|
conn.execute("DELETE FROM collection_pages WHERE id=?", (cp_id,))
|
|
conn.execute("DELETE FROM collections WHERE id=?", (col_id,))
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (admin_id,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (admin_id,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_delete_user_cascade_with_pages(client):
|
|
"""DELETE admin user cascade — also deletes workspace pages and favorites."""
|
|
from app.db import get_conn
|
|
admin_id, admin_login, admin_session = _create_admin_session()
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('caspage', 'CascadePage', '[email protected]')")
|
|
target_id = conn.execute("SELECT id FROM users WHERE login='caspage'").fetchone()["id"]
|
|
# Create workspace with pages
|
|
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('PageWS', ?)", (target_id,))
|
|
ws_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO pages (workspace, workspace_id, title) VALUES ('w', ?, 'Page1')", (ws_id,))
|
|
page_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO favorites (user_id, page_id) VALUES (?, ?)", (target_id, page_id))
|
|
# Create collection and page for page_history FK chain
|
|
conn.execute("INSERT INTO collections (name) VALUES ('cascade_col2')")
|
|
col_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO collection_pages (collection_id, title) VALUES (?, 'cp2')", (col_id,))
|
|
cp_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO page_history (page_id, user_id, change_type, snapshot_json) VALUES (?, ?, 'edited', '{}')", (cp_id, target_id))
|
|
conn.commit()
|
|
resp = client.delete(f"/api/admin/users/{target_id}", cookies={"flowdeck_session": admin_session})
|
|
assert resp.status_code == 200
|
|
with get_conn() as conn:
|
|
assert conn.execute("SELECT id FROM users WHERE id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM workspaces WHERE owner_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM pages WHERE id=?", (page_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM favorites WHERE user_id=?", (target_id,)).fetchone() is None
|
|
assert conn.execute("SELECT id FROM page_history WHERE user_id=?", (target_id,)).fetchone() is None
|
|
# Cleanup orphaned collection_pages and collection
|
|
conn.execute("DELETE FROM collection_pages WHERE id=?", (cp_id,))
|
|
conn.execute("DELETE FROM collections WHERE id=?", (col_id,))
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (admin_id,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (admin_id,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_stats(client):
|
|
"""GET /api/admin/stats — admin can see aggregate statistics."""
|
|
uid, login, session = _create_admin_session()
|
|
resp = client.get("/api/admin/stats", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
for key in ("total_users", "total_workspaces", "total_files"):
|
|
assert key in data
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_admin_stats_unauthorized(client):
|
|
"""GET /api/admin/stats — 403 for non-admin."""
|
|
resp = client.get("/api/admin/stats")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_audit(client):
|
|
"""GET /api/admin/audit — admin can view login history."""
|
|
uid, login, session = _create_admin_session()
|
|
resp = client.get("/api/admin/audit", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
assert "entries" in resp.json()
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: Gitea Status ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_gitea_status_unlinked(client):
|
|
"""GET /api/gitea/status — returns linked=false when no session/token."""
|
|
resp = client.get("/api/gitea/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["linked"] is False
|
|
|
|
|
|
def test_gitea_status_linked(client):
|
|
"""GET /api/gitea/status — returns linked=true when OAuth token exists."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('gstatus', 'GStatus', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='gstatus'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'status-token')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "gstatus", "is_admin": 0})
|
|
resp = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["linked"] is True
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_gitea_status_with_expired_token(client):
|
|
"""GET /api/gitea/status — returns linked=true even with old token (token existence is enough)."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('gexpired', 'GExp', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='gexpired'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token, expires_at) VALUES (?, 'gitea', 'old-token', '2020-01-01')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "gexpired", "is_admin": 0})
|
|
resp = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
# Token exists → linked=true (status endpoint only checks existence)
|
|
assert resp.json()["linked"] is True
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_gitea_disconnect_no_auth(client):
|
|
"""DELETE /api/gitea/disconnect — 401 without session."""
|
|
resp = client.delete("/api/gitea/disconnect")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_gitea_disconnect_with_auth(client):
|
|
"""DELETE /api/gitea/disconnect — removes OAuth tokens for authenticated user."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('disconn', 'Disconn', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='disconn'").fetchone()["id"]
|
|
conn.execute(
|
|
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'dc-token')",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "disconn", "is_admin": 0})
|
|
# Verify linked before
|
|
status_before = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
|
assert status_before.json()["linked"] is True
|
|
# Disconnect
|
|
resp = client.delete("/api/gitea/disconnect", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
# Verify unlinked after
|
|
status_after = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
|
assert status_after.json()["linked"] is False
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: OAuth Link Mode ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_oauth_login_local_page(client):
|
|
"""GET /auth/login?provider=local — renders local login HTML page."""
|
|
resp = client.get("/auth/login?provider=local")
|
|
assert resp.status_code == 200
|
|
assert "FlowDeck" in resp.text
|
|
assert "Login" in resp.text or "login" in resp.text.lower()
|
|
|
|
|
|
def test_oauth_login_gitea_redirect(client):
|
|
"""GET /auth/login?provider=gitea — redirects to Gitea OAuth (configured in test env)."""
|
|
resp = client.get("/auth/login?provider=gitea", follow_redirects=False)
|
|
# Gitea OAuth IS configured in test env → redirect to Gitea
|
|
assert resp.status_code == 302
|
|
assert "login/oauth" in resp.headers.get("location", "").lower()
|
|
|
|
|
|
def test_oauth_login_with_link_mode(client):
|
|
"""GET /auth/login?provider=gitea&mode=link — link mode redirects to Gitea OAuth."""
|
|
resp = client.get("/auth/login?provider=gitea&mode=link", follow_redirects=False)
|
|
# Gitea OAuth IS configured → redirect to Gitea with link mode set in session
|
|
assert resp.status_code == 302
|
|
assert "login/oauth" in resp.headers.get("location", "").lower()
|
|
|
|
|
|
def test_oauth_login_with_mode_link_and_provider_github(client):
|
|
"""GET /auth/login?provider=github&mode=link — sets link mode for GitHub."""
|
|
resp = client.get("/auth/login?provider=github&mode=link")
|
|
# GitHub OAuth not configured either → error page
|
|
assert resp.status_code == 200
|
|
assert "github" in resp.text.lower() or "not configured" in resp.text.lower()
|
|
|
|
|
|
def test_oauth_callback_invalid_state(client):
|
|
"""GET /auth/callback?code=test&state=invalid — 400 for invalid state."""
|
|
resp = client.get("/auth/callback?code=test_code&state=invalid_state")
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_oauth_callback_missing_code(client):
|
|
"""GET /auth/callback — 422 without required code param."""
|
|
resp = client.get("/auth/callback")
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_oauth_logout(client):
|
|
"""GET /auth/logout — redirects to local login page (follow_redirects=False)."""
|
|
resp = client.get("/auth/logout", follow_redirects=False)
|
|
assert resp.status_code == 302
|
|
assert "login" in resp.headers.get("location", "").lower()
|
|
|
|
|
|
def test_auth_register_missing_fields(client):
|
|
"""POST /auth/register — 400 without email/password."""
|
|
resp = client.post("/auth/register", json={})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_auth_register_short_password(client):
|
|
"""POST /auth/register — 400 with password < 6 chars."""
|
|
resp = client.post("/auth/register", json={"email": "[email protected]", "password": "ab"})
|
|
assert resp.status_code == 400
|
|
assert "6" in resp.json()["error"]
|
|
|
|
|
|
def test_auth_register_success(client):
|
|
"""POST /auth/register — successfully register a new user."""
|
|
resp = client.post("/auth/register", json={"email": "[email protected]", "password": "secret123", "name": "New User"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
assert resp.json()["user"]["login"] == "[email protected]"
|
|
# cleanup
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=(SELECT id FROM users WHERE login='[email protected]')")
|
|
conn.execute("DELETE FROM users WHERE login='[email protected]'")
|
|
conn.commit()
|
|
|
|
|
|
def test_auth_register_duplicate(client):
|
|
"""POST /auth/register — 409 for duplicate email."""
|
|
resp = client.post("/auth/register", json={"email": "[email protected]", "password": "secret123"})
|
|
assert resp.status_code == 200
|
|
# Try again with same email
|
|
resp2 = client.post("/auth/register", json={"email": "[email protected]", "password": "another1"})
|
|
assert resp2.status_code == 409
|
|
# cleanup
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM login_history WHERE user_id=(SELECT id FROM users WHERE login='[email protected]')")
|
|
conn.execute("DELETE FROM users WHERE login='[email protected]'")
|
|
conn.commit()
|
|
|
|
|
|
def test_auth_local_login_invalid_credentials(client):
|
|
"""POST /auth/local-login — 401 with wrong password."""
|
|
resp = client.post("/auth/local-login", json={"email": "[email protected]", "password": "wrong"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_auth_local_login_missing_fields(client):
|
|
"""POST /auth/local-login — 400 without email/password."""
|
|
resp = client.post("/auth/local-login", json={})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_auth_user_authenticated(client):
|
|
"""GET /auth/user — returns authenticated=true with valid session."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('authuser', 'Auth', '[email protected]')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='authuser'").fetchone()["id"]
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
session = SessionManager.create_session({"id": uid, "login": "authuser", "is_admin": 0})
|
|
resp = client.get("/auth/user", cookies={"flowdeck_session": session})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["authenticated"] is True
|
|
assert data["user"]["login"] == "authuser"
|
|
with get_conn() as conn:
|
|
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
|
conn.commit()
|
|
|
|
|
|
def test_auth_user_unauthenticated(client):
|
|
"""GET /auth/user — returns authenticated=false without session."""
|
|
resp = client.get("/auth/user")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["authenticated"] is False
|
|
|
|
|
|
# ══════════════════════════════════════════════════════
|
|
# ── v3.0.0: Gitea API Edge Cases ──
|
|
# ══════════════════════════════════════════════════════
|
|
|
|
def test_gitea_labels_no_auth(client):
|
|
"""GET /api/gitea/projects/owner/repo/labels — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/labels")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_gitea_tree_no_auth(client):
|
|
"""GET /api/gitea/projects/owner/repo/tree — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/tree")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_gitea_file_get_no_auth(client):
|
|
"""GET /api/gitea/projects/owner/repo/file?path=x — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/file?path=README.md")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_gitea_orgs_no_auth(client):
|
|
"""GET /api/gitea/orgs — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/orgs")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_gitea_projects_no_auth(client):
|
|
"""GET /api/gitea/projects — 401 without Gitea token."""
|
|
resp = client.get("/api/gitea/projects")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_gitea_file_delete_no_auth(client):
|
|
"""DELETE /api/gitea/projects/owner/repo/file — 401 without Gitea token."""
|
|
resp = client.delete("/api/gitea/projects/owner/repo/file?path=test.md&sha=abc")
|
|
assert resp.status_code == 401 # no Gitea token → 401
|
|
|
|
|
|
def test_gitea_file_delete_missing_params(client):
|
|
"""DELETE file without path+sha → 400 even without auth."""
|
|
resp = client.delete("/api/gitea/projects/owner/repo/file")
|
|
assert resp.status_code in (400, 401)
|
|
|
|
|
|
def test_gitea_private_pages_list_no_auth(client):
|
|
"""GET private-pages — returns empty without auth."""
|
|
resp = client.get("/api/gitea/projects/owner/repo/private-pages")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["pages"] == []
|
|
|
|
|
|
def test_gitea_private_pages_create_no_auth(client):
|
|
"""POST private-pages — 401 without session."""
|
|
resp = client.post("/api/gitea/projects/owner/repo/private-pages", json={"title": "Test"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ── v2.6.0: Notion-style breadcrumb navigation ──
|
|
|
|
def test_nav_menu_endpoint_responds(client):
|
|
"""GET /api/nav/menu — always returns an items list (200)."""
|
|
resp = client.get("/api/nav/menu")
|
|
assert resp.status_code == 200
|
|
assert "items" in resp.json()
|
|
|
|
|
|
def test_nav_menu_workspace_pages(client):
|
|
"""/api/nav/menu returns root pages and nested children for a workspace."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
u = conn.execute("INSERT OR IGNORE INTO users (login) VALUES ('navuser')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='navuser'").fetchone()["id"]
|
|
cur = conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('NavWS', ?)", (uid,))
|
|
ws_id = cur.lastrowid
|
|
p = conn.execute(
|
|
"INSERT INTO pages (workspace, workspace_id, title, content, content_format, "
|
|
"parent_section, parent_id) VALUES ('NavWS', ?, 'Parent Page', '', 'blocks', 'Private', NULL)",
|
|
(ws_id,),
|
|
)
|
|
parent_id = p.lastrowid
|
|
conn.execute(
|
|
"INSERT INTO pages (workspace, workspace_id, title, content, content_format, "
|
|
"parent_section, parent_id) VALUES ('NavWS', ?, 'Child Page', '', 'blocks', 'Private', ?)",
|
|
(ws_id, parent_id),
|
|
)
|
|
conn.commit()
|
|
|
|
# Root level should list the parent and flag it as having children
|
|
resp = client.get(f"/api/nav/menu?workspace_id={ws_id}")
|
|
assert resp.status_code == 200
|
|
items = resp.json()["items"]
|
|
names = {i["name"]: i for i in items}
|
|
assert "Parent Page" in names
|
|
assert names["Parent Page"]["has_children"] is True
|
|
|
|
# Children of the parent should include the child page
|
|
resp = client.get(f"/api/nav/menu?workspace_id={ws_id}&parent_id={parent_id}")
|
|
child_names = {i["name"] for i in resp.json()["items"]}
|
|
assert "Child Page" in child_names
|
|
|
|
|
|
def test_page_renders_breadcrumb_data(client):
|
|
"""Rendered page includes the breadcrumb JSON payload with a Home crumb."""
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT OR IGNORE INTO users (login) VALUES ('crumbuser')")
|
|
uid = conn.execute("SELECT id FROM users WHERE login='crumbuser'").fetchone()["id"]
|
|
cur = conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('CrumbWS', ?)", (uid,))
|
|
ws_id = cur.lastrowid
|
|
p = conn.execute(
|
|
"INSERT INTO pages (workspace, workspace_id, title, content, content_format, "
|
|
"parent_section, parent_id) VALUES ('CrumbWS', ?, 'Crumb Page', '', 'blocks', 'Private', NULL)",
|
|
(ws_id,),
|
|
)
|
|
page_id = p.lastrowid
|
|
conn.commit()
|
|
|
|
resp = client.get(f"/pages/{page_id}")
|
|
assert resp.status_code == 200
|
|
assert 'id="fd-breadcrumb-data"' in resp.text
|
|
assert '"Home"' in resp.text
|
|
assert "fdBreadcrumb" in resp.text
|
|
|
|
|
|
# ═══════════ v4.0.2 — Regression tests (critical paths) ═══════════
|
|
|
|
|
|
def test_landing_page_no_auth(client):
|
|
"""Visiting / without auth shows the landing page."""
|
|
resp = client.get("/", follow_redirects=False)
|
|
assert resp.status_code == 200
|
|
assert "FlowDeck" in resp.text
|
|
assert "Get started free" in resp.text or "Get started" in resp.text
|
|
|
|
|
|
def test_register_page_get(client):
|
|
"""GET /auth/register shows the registration form."""
|
|
resp = client.get("/auth/register", follow_redirects=False)
|
|
assert resp.status_code == 200
|
|
assert "register" in resp.text.lower()
|
|
|
|
|
|
def test_login_page_shows_expired_banner(client):
|
|
"""Login page with ?expired=1 shows session expired message."""
|
|
resp = client.get("/auth/login?provider=local&expired=1", follow_redirects=False)
|
|
assert resp.status_code == 200
|
|
assert "expired" in resp.text.lower()
|
|
|
|
|
|
def test_create_page_defaults_to_untitled(client):
|
|
"""Creating a page with empty title defaults to 'Untitled'."""
|
|
resp = client.post("/board/api/pages?title=§ion=Private", follow_redirects=False)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["title"] == "Untitled"
|
|
|
|
|
|
def test_styled_404_page(client):
|
|
"""Unknown routes return a styled 404 HTML page."""
|
|
resp = client.get("/this-does-not-exist-xyz", follow_redirects=False)
|
|
assert resp.status_code == 404
|
|
assert "404" in resp.text
|
|
assert "FlowDeck" in resp.text
|
|
|
|
|
|
def test_api_404_returns_json(client):
|
|
"""Unknown API routes return JSON, not HTML."""
|
|
resp = client.get("/api/does-not-exist", follow_redirects=False)
|
|
assert resp.status_code == 404
|
|
data = resp.json()
|
|
assert "detail" in data
|
|
|
|
|
|
def test_login_validation_empty_fields(client):
|
|
"""Login with empty fields returns 400 error."""
|
|
resp = client.post("/auth/local-login", json={"email": "", "password": ""})
|
|
assert resp.status_code == 400
|
|
data = resp.json()
|
|
assert "error" in data
|
|
|
|
|
|
def test_register_validation_short_password(client):
|
|
"""Registration with short password returns 400."""
|
|
resp = client.post("/auth/register", json={
|
|
"email": "[email protected]", "password": "ab", "name": "Test"
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_register_duplicate_rejected(client):
|
|
"""Duplicate registration returns 409."""
|
|
# Register first time
|
|
r1 = client.post("/auth/register", json={
|
|
"email": "duptest2", "password": "password123", "name": "Dup"
|
|
})
|
|
assert r1.status_code == 200
|
|
# Second registration with same email should fail
|
|
r2 = client.post("/auth/register", json={
|
|
"email": "duptest2", "password": "password123", "name": "Dup2"
|
|
})
|
|
assert r2.status_code == 409
|
|
assert "already exists" in r2.json().get("error", "").lower()
|
|
|
|
|
|
def test_session_expired_redirect(client):
|
|
"""Unauthenticated access to protected page redirects with expired param."""
|
|
resp = client.get("/workspaces", follow_redirects=False)
|
|
assert resp.status_code == 302
|
|
location = resp.headers.get("location", "")
|
|
assert "login" in location
|
|
assert "expired=1" in location
|