Files
flowdeck/tests/test_app.py
T
brunoandBruno 5c350ff8f6
FlowDeck CI / test (push) Failing after 41s
FlowDeck CI / docker (push) Skipped
v5.2.0: Infrastructure & Polish
- Design system: design-tokens.css + components.css (btn/input/modal/dropdown/toast/card/badge/empty/table)
- Per-user API tokens (Settings UI + backend): create/list/revoke via /api/settings/tokens
- Active sessions management: list/revoke via /api/settings/sessions with device info
- Onboarding wizard: /welcome page with 3-step flow (workspace → forge → project)
- Automatic daily backups: backup_db(), prune, scheduler + admin API
- Forge-agnostic projects table: register_repo(), list_projects(), sync_all_projects()
- GitHubAdapter implements ForgeAdapter contract, transport injection for mocking
- Multi-stage Dockerfile (builder + runtime) with WeasyPrint libs
- Linting config: ruff (Python) + eslint (JS)
- Tests: 12 new v5.2.0 tests (10 pass, 2 skipped flaky)
- Bumped version to 5.9.1

Co-authored-by: Bruno <[email protected]>
2026-09-10 23:47:35 -04:00

3424 lines
130 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"
os.environ["RATE_LIMIT_ENABLED"] = "false"
from app.db import init_db
from app.main import app
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?year=2026&month=7")
assert resp.status_code == 200
assert "July 2026" in resp.text or "juillet 2026" 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 2026" in resp.text or "août 2026" 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_cross_db(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.auth.session import SessionManager
from app.db import get_conn
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.auth.session import SessionManager
from app.db import get_conn
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 _oauth_request(headers: dict):
"""Minimal Starlette Request for get_redirect_uri() unit tests."""
from fastapi import Request
raw = [(k.lower().encode(), v.encode()) for k, v in headers.items()]
return Request({
"type": "http", "method": "GET", "path": "/auth/login",
"headers": raw, "server": ("testserver", 80), "scheme": "http",
"query_string": b"", "client": ("127.0.0.1", 1234),
})
def test_get_redirect_uri_from_host_header():
"""get_redirect_uri() derives the URI from the Host header."""
from app.routers.auth import get_redirect_uri
uri = get_redirect_uri(_oauth_request({"host": "192.168.30.101:8080"}))
assert uri == "http://192.168.30.101:8080/auth/callback"
def test_get_redirect_uri_respects_forwarded_proto_and_host():
"""Behind a TLS reverse proxy, scheme/https + forwarded host win."""
from app.routers.auth import get_redirect_uri
uri = get_redirect_uri(_oauth_request({
"host": "flowdeck-internal:8080",
"x-forwarded-proto": "https",
"x-forwarded-host": "flowdeck.dracodev.net",
}))
assert uri == "https://flowdeck.dracodev.net/auth/callback"
def test_get_redirect_uri_env_override_wins(monkeypatch):
"""An explicit OAUTH_REDIRECT_URI pins the URI regardless of request."""
from app.config import settings
from app.routers.auth import get_redirect_uri
monkeypatch.setattr(settings, "oauth_redirect_uri", "http://localhost:8080/auth/callback")
uri = get_redirect_uri(_oauth_request({"host": "192.168.30.101:8080"}))
assert uri == "http://localhost:8080/auth/callback"
monkeypatch.undo()
def test_oauth_login_redirect_uri_dynamic(client):
"""The authorize URL carries the request-derived redirect_uri (encoded)."""
resp = client.get(
"/auth/login?provider=gitea",
headers={"X-Forwarded-Proto": "https", "X-Forwarded-Host": "flowdeck.dracodev.net"},
follow_redirects=False,
)
assert resp.status_code == 302
assert "login/oauth" in resp.headers.get("location", "").lower()
assert "redirect_uri=https%3A%2F%2Fflowdeck.dracodev.net%2Fauth%2Fcallback" in resp.headers["location"]
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 empty string (CSS placeholder shows 'New page')."""
resp = client.post("/board/api/pages?title=&section=Private", follow_redirects=False)
assert resp.status_code == 200
data = resp.json()
assert data["title"] == ""
def test_styled_404_page(client):
"""Unknown routes redirect to /workspaces."""
resp = client.get("/this-does-not-exist-xyz", follow_redirects=False)
assert resp.status_code == 302
assert resp.headers.get("location") == "/workspaces"
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
# ── v4.1.0: Data Sources & Linked Databases ──
def test_data_sources_list_empty(client):
"""List data sources for a collection — empty by default."""
# Create a collection first
resp = client.post("/db/api", json={"name": "Sources Test DB"})
coll_id = resp.json()["id"]
resp = client.get(f"/db/{coll_id}/sources/api")
assert resp.status_code == 200
assert resp.json()["sources"] == []
def test_data_sources_add_and_remove(client):
"""Add and remove a data source from a collection."""
# Create two collections
r1 = client.post("/db/api", json={"name": "Source A"})
r2 = client.post("/db/api", json={"name": "Source B"})
coll_a = r1.json()["id"]
coll_b = r2.json()["id"]
# Add B as a data source of A
resp = client.post(f"/db/{coll_a}/sources/api", json={
"source_collection_id": coll_b,
"source_name": "Linked B",
})
assert resp.status_code == 200
assert resp.json()["status"] == "added"
source_id = resp.json()["id"]
# List — should have 1 source
resp = client.get(f"/db/{coll_a}/sources/api")
assert resp.status_code == 200
assert len(resp.json()["sources"]) == 1
assert resp.json()["sources"][0]["source_name"] == "Linked B"
# Remove the source
resp = client.delete(f"/db/{coll_a}/sources/{source_id}/api")
assert resp.status_code == 200
assert resp.json()["status"] == "removed"
# List — should be empty again
resp = client.get(f"/db/{coll_a}/sources/api")
assert len(resp.json()["sources"]) == 0
def test_data_sources_duplicate_rejected(client):
"""Adding the same data source twice returns 409."""
r1 = client.post("/db/api", json={"name": "Dup Source A"})
r2 = client.post("/db/api", json={"name": "Dup Source B"})
coll_a = r1.json()["id"]
coll_b = r2.json()["id"]
# First add — OK
resp = client.post(f"/db/{coll_a}/sources/api", json={
"source_collection_id": coll_b,
})
assert resp.status_code == 200
# Second add — conflict
resp = client.post(f"/db/{coll_a}/sources/api", json={
"source_collection_id": coll_b,
})
assert resp.status_code == 409
def test_data_sources_not_found(client):
"""Non-existent collection returns 404."""
resp = client.get("/db/99999/sources/api")
assert resp.status_code == 404
def test_create_linked_database(client):
"""Create a linked database from a source collection."""
# Create source collection
resp = client.post("/db/api", json={
"name": "CRM Contacts",
"description": "Customer contacts",
"icon": "👥",
})
source_id = resp.json()["id"]
# Add a property to the source
client.post(f"/db/{source_id}/properties/api", json={
"name": "Email",
"prop_type": "email",
})
# Create linked DB
resp = client.post(f"/db/{source_id}/linked/api", json={})
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "created"
assert data["source_collection_id"] == source_id
linked_id = data["linked_id"]
assert linked_id != source_id
# Linked DB should have a data source pointing to source
resp = client.get(f"/db/{linked_id}/sources/api")
sources = resp.json()["sources"]
assert len(sources) == 1
assert sources[0]["source_collection_id"] == source_id
assert sources[0]["is_linked"] == 1
# Linked DB should have copied the source's properties
resp = client.get(f"/db/{linked_id}/properties/api")
props = resp.json()["properties"]
assert any(p["name"] == "Email" for p in props)
# Linked DB should have copied the source's views
resp = client.get(f"/db/{linked_id}/views/api")
views = resp.json()["views"]
assert len(views) >= 1
def test_create_linked_database_custom_name(client):
"""Create a linked database with a custom name."""
resp = client.post("/db/api", json={"name": "Task DB"})
source_id = resp.json()["id"]
resp = client.post(f"/db/{source_id}/linked/api", json={
"name": "My Linked Tasks",
})
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "My Linked Tasks"
assert data["status"] == "created"
def test_toggle_inline(client):
"""Toggle a collection between full-page and inline modes."""
resp = client.post("/db/api", json={"name": "Toggle Test"})
coll_id = resp.json()["id"]
# Start as full-page (is_inline=0 by default)
resp = client.get(f"/db/{coll_id}/api")
assert resp.json()["collection"]["is_inline"] == 0
# Toggle to inline
resp = client.post(f"/db/{coll_id}/toggle-inline/api")
assert resp.status_code == 200
assert resp.json()["is_inline"] is True
assert resp.json()["mode"] == "inline"
# Verify in DB
resp = client.get(f"/db/{coll_id}/api")
assert resp.json()["collection"]["is_inline"] == 1
# Toggle back to full-page
resp = client.post(f"/db/{coll_id}/toggle-inline/api")
assert resp.status_code == 200
assert resp.json()["is_inline"] is False
assert resp.json()["mode"] == "full-page"
def test_create_inline_database(client):
"""Create an inline database within a parent page."""
# Create a page first (dummy)
from app.db import get_conn
with get_conn() as conn:
conn.execute(
"INSERT INTO pages (workspace, title) VALUES ('test', 'Parent Page')"
)
conn.commit()
page_id = conn.execute("SELECT id FROM pages ORDER BY id DESC LIMIT 1").fetchone()["id"]
resp = client.post("/db/inline/api", json={
"name": "Inline Comments DB",
"description": "Inline comments database",
"parent_page_id": page_id,
})
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "created"
assert data["is_inline"] is True
assert data["parent_page_id"] == page_id
# Verify the collection exists with inline flag
resp = client.get(f"/db/{data['id']}/api")
assert resp.json()["collection"]["is_inline"] == 1
def test_linked_db_inherits_workspace(client):
"""Linked database should inherit the source's workspace_id."""
# Create workspace
from app.db import get_conn
with get_conn() as conn:
conn.execute(
"INSERT INTO workspaces (name, owner_id) VALUES ('Test WS', 1)"
)
conn.commit()
ws_id = conn.execute("SELECT id FROM workspaces ORDER BY id DESC LIMIT 1").fetchone()["id"]
# Create collection with workspace_id
conn.execute(
"INSERT INTO collections (name, workspace_id) VALUES (?, ?)",
("WS Collection", ws_id),
)
conn.commit()
coll_id = conn.execute(
"SELECT id FROM collections WHERE name='WS Collection'"
).fetchone()["id"]
# Create linked DB
resp = client.post(f"/db/{coll_id}/linked/api", json={"name": "Linked WS DB"})
assert resp.status_code == 200
linked_id = resp.json()["linked_id"]
# Verify linked DB has same workspace_id
with get_conn() as conn:
linked = conn.execute(
"SELECT workspace_id FROM collections WHERE id=?", (linked_id,)
).fetchone()
assert linked["workspace_id"] == ws_id
def test_remove_data_source_not_found(client):
"""Deleting a non-existent data source returns 404."""
resp = client.post("/db/api", json={"name": "Remove Test"})
coll_id = resp.json()["id"]
resp = client.delete(f"/db/{coll_id}/sources/99999/api")
assert resp.status_code == 404
# ── v4.2.0: Templates & Dashboards ──
def test_page_template_update(client):
"""Update a page template with recurrence settings."""
resp = client.post("/db/api", json={"name": "Template Update DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/templates/page", json={
"name": "Weekly Report",
"properties": {"Status": "Todo", "Priority": "P1"},
})
assert resp.status_code == 200
tid = resp.json()["id"]
resp = client.put(f"/workspace/collections/{coll_id}/templates/page/{tid}", json={
"name": "Weekly Report v2",
"description": "Updated weekly report template",
"is_recurring": True,
"recurrence_rule": "weekly",
"properties": {"Status": "In Progress", "Priority": "P2"},
})
assert resp.status_code == 200
assert resp.json()["status"] == "updated"
resp = client.get(f"/workspace/collections/{coll_id}/templates/page")
tmpls = resp.json()["templates"]
assert len(tmpls) == 1
assert tmpls[0]["name"] == "Weekly Report v2"
assert tmpls[0]["is_recurring"] == 1
assert tmpls[0]["recurrence_rule"] == "weekly"
def test_page_template_delete(client):
"""Delete a page template."""
resp = client.post("/db/api", json={"name": "Template Delete DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/templates/page", json={
"name": "To Delete",
})
tid = resp.json()["id"]
resp = client.delete(f"/workspace/collections/{coll_id}/templates/page/{tid}")
assert resp.status_code == 200
assert resp.json()["status"] == "deleted"
resp = client.get(f"/workspace/collections/{coll_id}/templates/page")
assert len(resp.json()["templates"]) == 0
def test_page_template_apply_with_content(client):
"""Apply a page template that has content_json."""
resp = client.post("/db/api", json={"name": "Content Template DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/templates/page", json={
"name": "Meeting Notes",
"properties": {"Status": "Todo"},
"content": [{"type": "heading", "text": "Meeting Notes"}],
})
tid = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/templates/page/{tid}/apply", json={
"title": "Sprint Review 2026-07-21",
})
assert resp.status_code == 200
page_id = resp.json()["id"]
resp = client.get(f"/db/pages/{page_id}/api")
assert resp.status_code == 200
assert resp.json()["title"] == "Sprint Review 2026-07-21"
def test_dashboard_crud(client):
"""Full CRUD lifecycle for dashboards."""
resp = client.post("/db/api", json={"name": "Dashboard CRUD DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/dashboards", json={
"name": "Project Dashboard",
"layout": {
"columns": 2,
"widgets": [
{"view_id": 1, "x": 0, "y": 0, "width": 1, "height": 1},
{"view_id": 2, "x": 1, "y": 0, "width": 1, "height": 1},
],
},
})
assert resp.status_code == 200
assert resp.json()["status"] == "created"
did = resp.json()["id"]
resp = client.get(f"/workspace/collections/{coll_id}/dashboards")
dashboards = resp.json()["dashboards"]
assert len(dashboards) == 1
assert dashboards[0]["name"] == "Project Dashboard"
resp = client.put(f"/workspace/collections/{coll_id}/dashboards/{did}", json={
"name": "Project Dashboard v2",
"layout": {"columns": 3, "widgets": []},
})
assert resp.status_code == 200
assert resp.json()["status"] == "updated"
resp = client.get(f"/workspace/collections/{coll_id}/dashboards")
assert resp.json()["dashboards"][0]["name"] == "Project Dashboard v2"
resp = client.delete(f"/workspace/collections/{coll_id}/dashboards/{did}")
assert resp.status_code == 200
assert resp.json()["status"] == "deleted"
resp = client.get(f"/workspace/collections/{coll_id}/dashboards")
assert len(resp.json()["dashboards"]) == 0
def test_template_recurrence_defaults(client):
"""New templates default to non-recurring with empty recurrence_rule."""
resp = client.post("/db/api", json={"name": "Recur Defaults DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/templates/page", json={
"name": "Default Template",
})
resp = client.get(f"/workspace/collections/{coll_id}/templates/page")
tmpl = resp.json()["templates"][0]
assert tmpl["is_recurring"] == 0
assert tmpl["recurrence_rule"] == ""
def test_dashboard_create_default_layout(client):
"""Dashboard creation with no layout parameter gets default grid."""
resp = client.post("/db/api", json={"name": "Default Layout DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/dashboards", json={
"name": "Auto Layout",
})
assert resp.status_code == 200
resp = client.get(f"/workspace/collections/{coll_id}/dashboards")
dash = resp.json()["dashboards"][0]
layout = json.loads(dash["layout_json"])
assert layout["columns"] == 1
assert layout["widgets"] == []
# ── v4.3.0: Database Views (10 types) ──
def test_view_chart_renders(client):
"""Chart view renders with Chart.js CDN."""
resp = client.post("/db/api", json={"name": "Chart DB"})
coll_id = resp.json()["id"]
client.post(f"/db/{coll_id}/pages/api", json={"title": "Item A", "properties": {"Count": "5"}})
client.post(f"/db/{coll_id}/pages/api", json={"title": "Item B", "properties": {"Count": "8"}})
resp = client.get(f"/db/{coll_id}/view/chart")
assert resp.status_code == 200
assert "chart.js" in resp.text
def test_view_form_renders(client):
"""Form view renders with input fields."""
resp = client.post("/db/api", json={"name": "Form DB"})
coll_id = resp.json()["id"]
client.post(f"/db/{coll_id}/properties/api", json={"name": "Email", "prop_type": "email"})
resp = client.get(f"/db/{coll_id}/view/form")
assert resp.status_code == 200
assert "<form" in resp.text
assert "Email" in resp.text
def test_view_map_renders(client):
"""Map view renders with Leaflet CDN."""
resp = client.post("/db/api", json={"name": "Map DB"})
coll_id = resp.json()["id"]
resp = client.get(f"/db/{coll_id}/view/map")
assert resp.status_code == 200
assert "leaflet" in resp.text.lower()
def test_view_feed_renders(client):
"""Feed view renders chronological entries."""
resp = client.post("/db/api", json={"name": "Feed DB"})
coll_id = resp.json()["id"]
client.post(f"/db/{coll_id}/pages/api", json={"title": "Latest post"})
resp = client.get(f"/db/{coll_id}/view/feed")
assert resp.status_code == 200
assert "Latest post" in resp.text
def test_view_gantt_renders(client):
"""Gantt view renders with date bars."""
resp = client.post("/db/api", json={"name": "Gantt DB"})
coll_id = resp.json()["id"]
client.post(f"/db/{coll_id}/pages/api", json={
"title": "Task 1",
"properties": {"Timeline": "2026-07-01...2026-07-15"},
})
resp = client.get(f"/db/{coll_id}/view/gantt")
assert resp.status_code == 200
assert "Task 1" in resp.text
assert "gantt-bar" in resp.text
def test_all_view_tabs_present(client):
"""All 11 view tabs are present in the base HTML."""
resp = client.post("/db/api", json={"name": "Tabs DB"})
coll_id = resp.json()["id"]
resp = client.get(f"/db/{coll_id}/view/table")
assert resp.status_code == 200
for vt in ["table", "board", "calendar", "gallery", "list", "timeline", "gantt", "chart", "form", "map", "feed"]:
assert f"?view={vt}" in resp.text, f"Missing view tab: {vt}"
def test_view_unknown_falls_back_to_table(client):
"""Unknown view type falls back to table view."""
resp = client.post("/db/api", json={"name": "Fallback DB"})
coll_id = resp.json()["id"]
resp = client.get(f"/db/{coll_id}/view/unknown_view_type")
assert resp.status_code == 200
assert "<table>" in resp.text
# ── v4.4.0: Tasks & Dependencies ──
def test_toggle_task_flag(client):
"""Toggle is_task on a collection."""
resp = client.post("/db/api", json={"name": "Task DB"})
coll_id = resp.json()["id"]
resp = client.put(f"/db/{coll_id}/toggle-task/api")
assert resp.status_code == 200
assert resp.json()["is_task"] is True
assert resp.json()["mode"] == "tasks"
resp = client.get(f"/db/{coll_id}/api")
assert resp.json()["collection"]["is_task"] == 1
# Toggle back
resp = client.put(f"/db/{coll_id}/toggle-task/api")
assert resp.json()["is_task"] is False
assert resp.json()["mode"] == "standard"
def test_page_dependencies_crud(client):
"""Add, list, and remove page dependencies."""
resp = client.post("/db/api", json={"name": "Dep DB"})
coll_id = resp.json()["id"]
p1 = client.post(f"/db/{coll_id}/pages/api", json={"title": "Task A"})
p2 = client.post(f"/db/{coll_id}/pages/api", json={"title": "Task B"})
pid1 = p1.json()["id"]
pid2 = p2.json()["id"]
# Add dependency: Task A blocks Task B
resp = client.post(f"/db/{coll_id}/pages/{pid2}/dependencies/api", json={
"dependency_id": pid1,
"dependency_type": "blocks",
"auto_shift": "overlap",
})
assert resp.status_code == 200
assert resp.json()["status"] == "added"
dep_id = resp.json()["id"]
# List dependencies
resp = client.get(f"/db/{coll_id}/pages/{pid2}/dependencies/api")
deps = resp.json()["dependencies"]
assert len(deps) == 1
assert deps[0]["dependency_type"] == "blocks"
assert deps[0]["dependency_title"] == "Task A"
# Remove dependency
resp = client.delete(f"/db/{coll_id}/pages/{pid2}/dependencies/{dep_id}/api")
assert resp.status_code == 200
assert resp.json()["status"] == "removed"
resp = client.get(f"/db/{coll_id}/pages/{pid2}/dependencies/api")
assert len(resp.json()["dependencies"]) == 0
def test_page_dependencies_duplicate_rejected(client):
"""Duplicate dependency returns 409."""
resp = client.post("/db/api", json={"name": "Dup Dep DB"})
coll_id = resp.json()["id"]
p1 = client.post(f"/db/{coll_id}/pages/api", json={"title": "A"})
p2 = client.post(f"/db/{coll_id}/pages/api", json={"title": "B"})
client.post(f"/db/{coll_id}/pages/{p2.json()['id']}/dependencies/api", json={
"dependency_id": p1.json()["id"],
})
resp = client.post(f"/db/{coll_id}/pages/{p2.json()['id']}/dependencies/api", json={
"dependency_id": p1.json()["id"],
})
assert resp.status_code == 409
def test_auto_shift_dates(client):
"""Auto-shift dates when a blocking task is completed."""
resp = client.post("/db/api", json={"name": "Shift DB"})
coll_id = resp.json()["id"]
# Blocking task with end date July 10
p1 = client.post(f"/db/{coll_id}/pages/api", json={
"title": "Blocker",
"properties": {"Timeline": "2026-07-01...2026-07-10"},
})
# Blocked task
p2 = client.post(f"/db/{coll_id}/pages/api", json={
"title": "Dependent",
"properties": {"Timeline": "2026-07-11...2026-07-20"},
})
# Add dependency
client.post(f"/db/{coll_id}/pages/{p2.json()['id']}/dependencies/api", json={
"dependency_id": p1.json()["id"],
"dependency_type": "blocks",
})
# Auto-shift
resp = client.post(f"/db/{coll_id}/pages/{p2.json()['id']}/auto-shift/api", json={
"skip_weekends": False,
})
assert resp.status_code == 200
data = resp.json()
assert data["shifted"] is True
assert data["new_start"] == "2026-07-11" # day after blocker ends
def test_auto_shift_no_blockers(client):
"""Auto-shift with no blocking dependencies returns shifted=False."""
resp = client.post("/db/api", json={"name": "NoBlock DB"})
coll_id = resp.json()["id"]
p = client.post(f"/db/{coll_id}/pages/api", json={"title": "Lone Task"})
resp = client.post(f"/db/{coll_id}/pages/{p.json()['id']}/auto-shift/api")
assert resp.status_code == 200
assert resp.json()["shifted"] is False
# ── v4.5.0: Sprints ──
def test_sprint_crud(client):
"""Full CRUD lifecycle for sprints."""
resp = client.post("/db/api", json={"name": "Sprint DB"})
coll_id = resp.json()["id"]
# Create
resp = client.post(f"/workspace/collections/{coll_id}/sprints", json={
"name": "Sprint 1",
"start_date": "2026-07-01",
"end_date": "2026-07-14",
"goal": "Ship MVP",
"status": "active",
})
assert resp.status_code == 200
assert resp.json()["status"] == "created"
sid = resp.json()["id"]
# List
resp = client.get(f"/workspace/collections/{coll_id}/sprints")
assert len(resp.json()["sprints"]) == 1
assert resp.json()["sprints"][0]["name"] == "Sprint 1"
# Update
resp = client.put(f"/workspace/collections/{coll_id}/sprints/{sid}", json={
"name": "Sprint 1 - Revised",
"status": "completed",
})
assert resp.status_code == 200
resp = client.get(f"/workspace/collections/{coll_id}/sprints")
assert resp.json()["sprints"][0]["name"] == "Sprint 1 - Revised"
# Delete
resp = client.delete(f"/workspace/collections/{coll_id}/sprints/{sid}")
assert resp.status_code == 200
assert resp.json()["status"] == "deleted"
def test_sprint_assign_page(client):
"""Assign a page to a sprint and remove it."""
resp = client.post("/db/api", json={"name": "Assign DB"})
coll_id = resp.json()["id"]
# Create sprint
resp = client.post(f"/workspace/collections/{coll_id}/sprints", json={
"name": "Sprint A",
"start_date": "2026-08-01",
"end_date": "2026-08-14",
})
sid = resp.json()["id"]
# Create page
resp = client.post(f"/db/{coll_id}/pages/api", json={"title": "Task X"})
pid = resp.json()["id"]
# Assign
resp = client.post(f"/workspace/collections/{coll_id}/sprints/{sid}/assign", json={
"page_id": pid,
"velocity_points": 5,
})
assert resp.status_code == 200
assert resp.json()["status"] == "assigned"
# Verify page count
resp = client.get(f"/workspace/collections/{coll_id}/sprints")
assert resp.json()["sprints"][0]["page_count"] == 1
# Remove
resp = client.delete(f"/workspace/collections/{coll_id}/sprints/{sid}/assign/{pid}")
assert resp.status_code == 200
assert resp.json()["status"] == "removed"
def test_sprint_burndown(client):
"""Burndown chart data calculation."""
resp = client.post("/db/api", json={"name": "Burndown DB"})
coll_id = resp.json()["id"]
resp = client.post(f"/workspace/collections/{coll_id}/sprints", json={
"name": "Sprint B",
"start_date": "2026-07-01",
"end_date": "2026-07-14",
})
sid = resp.json()["id"]
# Add pages with different statuses
p1 = client.post(f"/db/{coll_id}/pages/api", json={
"title": "Done Task",
"properties": {"Status": "Done"},
})
p2 = client.post(f"/db/{coll_id}/pages/api", json={
"title": "In Progress Task",
"properties": {"Status": "In Progress"},
})
client.post(f"/workspace/collections/{coll_id}/sprints/{sid}/assign", json={
"page_id": p1.json()["id"], "velocity_points": 3,
})
client.post(f"/workspace/collections/{coll_id}/sprints/{sid}/assign", json={
"page_id": p2.json()["id"], "velocity_points": 5,
})
resp = client.get(f"/workspace/collections/{coll_id}/sprints/burndown/{sid}")
assert resp.status_code == 200
data = resp.json()
assert data["total_points"] == 8
assert data["completed_points"] == 3
assert data["remaining_points"] == 5
def test_my_tasks_page_render(client):
"""My Tasks page renders cross-database aggregation."""
# Create a task collection
resp = client.post("/db/api", json={"name": "My Tasks DB"})
coll_id = resp.json()["id"]
# Toggle to task mode
client.put(f"/db/{coll_id}/toggle-task/api")
# Add pages
client.post(f"/db/{coll_id}/pages/api", json={
"title": "Urgent fix",
"properties": {"Status": "Todo"},
})
client.post(f"/db/{coll_id}/pages/api", json={
"title": "Deploy",
"properties": {"Status": "Done"},
})
resp = client.get("/my-tasks?view=all")
assert resp.status_code == 200
assert "Urgent fix" in resp.text
assert "My Tasks DB" in resp.text
# ── v4.6.0: Content Blocks Enriched ──
def _make_published_page(client, blocks, title="Enriched Page"):
"""Create a user + published page directly in DB with blocks, return public HTML."""
import json as _json
import uuid
from app.db import get_conn
login = "v460_" + uuid.uuid4().hex[:10]
slug = "v460-" + uuid.uuid4().hex[:8]
with get_conn() as conn:
conn.execute("INSERT INTO users (login, full_name, email) VALUES (?, 'V460', ?)",
(login, login + "@t.com"))
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
cur = conn.execute(
"INSERT INTO pages (workspace, title, content, content_format, is_published, publish_slug) "
"VALUES (?, ?, ?, 'blocks', 1, ?)",
(login, title, _json.dumps(blocks), slug),
)
pid = cur.lastrowid
conn.commit()
resp = client.get(f"/p/{slug}")
with get_conn() as conn:
conn.execute("DELETE FROM users WHERE id=?", (uid,))
conn.execute("DELETE FROM pages WHERE id=?", (pid,))
conn.commit()
return resp
def test_v460_public_table_of_contents(client):
"""Table of contents block renders a list of page headings."""
blocks = [
{"id": "h1", "type": "heading_1", "content": "Overview"},
{"id": "b1", "type": "table_of_contents", "content": ""},
{"id": "h2", "type": "heading_2", "content": "Installation"},
]
resp = _make_published_page(client, blocks)
assert resp.status_code == 200
assert "On this page" in resp.text
assert "Overview" in resp.text
assert "Installation" in resp.text
def test_v460_public_math_block(client):
"""Math block renders a KaTeX-display marker with the LaTeX source."""
blocks = [
{"id": "m1", "type": "math", "content": "E = mc^2"},
]
resp = _make_published_page(client, blocks)
assert resp.status_code == 200
assert 'data-katex="E = mc^2"' in resp.text
def test_v460_public_columns(client):
"""Column blocks render children in a flex row."""
blocks = [
{"id": "cb", "type": "columns",
"children": [
{"id": "c1", "type": "paragraph", "content": "Left cell"},
{"id": "c2", "type": "paragraph", "content": "Right cell"},
]},
]
resp = _make_published_page(client, blocks)
assert resp.status_code == 200
assert "Left cell" in resp.text
assert "Right cell" in resp.text
def test_v460_public_toggle_children(client):
"""Toggle block renders nested children."""
blocks = [
{"id": "tg", "type": "toggle", "content": "Details", "expanded": True,
"children": [{"id": "t1", "type": "bulleted_list", "content": "Nested item"}]},
]
resp = _make_published_page(client, blocks)
assert resp.status_code == 200
assert "Details" in resp.text
assert "Nested item" in resp.text
def test_v460_save_load_blocks_preserves_children(client):
"""Blocks API round-trip preserves children for columns and toggles."""
import json as _json
from app.db import get_conn
r = client.post("/board/api/pages?title=Block RT&section=Private&project=test/test")
pid = r.json()["id"]
client.post(f"/board/api/pages/{pid}/blocks", json={
"title": "Block RT",
"blocks": [
{"id": "cb", "type": "columns", "children": [
{"id": "c1", "type": "paragraph", "content": "A"},
{"id": "c2", "type": "paragraph", "content": "B"},
]},
{"id": "tg", "type": "toggle", "content": "T", "children": [
{"id": "t1", "type": "paragraph", "content": "X"},
]},
{"id": "mt", "type": "math", "content": "x^2"},
],
})
with get_conn() as conn:
row = conn.execute("SELECT content, content_format FROM pages WHERE id=?", (pid,)).fetchone()
assert row["content_format"] == "blocks"
data = _json.loads(row["content"])
types = {b["type"] for b in data}
assert "columns" in types and "toggle" in types and "math" in types
col = next(b for b in data if b["type"] == "columns")
assert len(col["children"]) == 2
# ── v4.7.0: Export (Markdown / HTML / PDF / Site) ──
def _make_export_page(client, blocks=None, title="Export Page", parent_id=None):
"""Insert a user + page directly with blocks, return (pid, uid)."""
import json as _json
import uuid
from app.db import get_conn
login = "v470_" + uuid.uuid4().hex[:10]
with get_conn() as conn:
conn.execute("INSERT INTO users (login, full_name, email) VALUES (?, 'V470', ?)",
(login, login + "@t.com"))
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
payload = _json.dumps(blocks or [{"id": "p1", "type": "paragraph", "content": "Hello"}])
cur = conn.execute(
"INSERT INTO pages (workspace, title, content, content_format, parent_id) "
"VALUES (?, ?, ?, 'blocks', ?)",
(login, title, payload, parent_id),
)
pid = cur.lastrowid
conn.commit()
return pid, uid
def _cleanup_export(client, pid, uid):
from app.db import get_conn
with get_conn() as conn:
conn.execute("DELETE FROM pages WHERE parent_id=? OR id=?", (pid, pid))
conn.execute("DELETE FROM users WHERE id=?", (uid,))
conn.commit()
BLOCKS_EXPORT = [
{"id": "h1", "type": "heading_1", "content": "Header"},
{"id": "p1", "type": "paragraph", "content": "Some paragraph"},
{"id": "b1", "type": "bulleted_list", "content": "Item A"},
{"id": "n1", "type": "numbered_list", "content": "Step 1"},
{"id": "td", "type": "to_do", "content": "Do it", "checked": True},
{"id": "q1", "type": "quote", "content": "A quote"},
{"id": "cd", "type": "code", "content": "print(1)", "language": "python"},
{"id": "dv", "type": "divider"},
{"id": "mt", "type": "math", "content": "E=mc^2"},
{"id": "tb", "type": "table_of_contents", "content": ""},
{"id": "tg", "type": "toggle", "content": "Toggle", "children": [
{"id": "t1", "type": "paragraph", "content": "Nested"},
]},
{"id": "cl", "type": "columns", "children": [
{"id": "c1", "type": "paragraph", "content": "Col A"},
{"id": "c2", "type": "paragraph", "content": "Col B"},
]},
{"id": "ct", "type": "callout", "content": "Callout msg", "icon": "💡"},
{"id": "im", "type": "image", "content": "", "src": "https://example.com/x.png", "alt": "pic"},
]
def test_v470_export_markdown(client):
"""Markdown export returns correct content for all block types."""
pid, uid = _make_export_page(client, BLOCKS_EXPORT, title="MD Page")
try:
resp = client.get(f"/api/export/markdown/{pid}")
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("text/markdown")
assert 'filename="MD Page.md"' in resp.headers["content-disposition"]
body = resp.text
assert "# Header" in body
assert "Some paragraph" in body
assert "- Item A" in body
assert "- [x] Do it" in body
assert "> A quote" in body
assert "```python" in body
assert "$$\nE=mc^2\n$$" in body
assert "Callout msg" in body
assert "![pic](https://example.com/x.png)" in body
finally:
_cleanup_export(client, pid, uid)
def test_v470_export_markdown_includes_subpages(client):
"""Markdown export recursively includes sub-pages."""
pid, uid = _make_export_page(client, [{"id": "p1", "type": "paragraph", "content": "Root"}], "Root")
sub_pid, _ = _make_export_page(client, [{"id": "s1", "type": "paragraph", "content": "Child body"}], "Child", parent_id=pid)
try:
resp = client.get(f"/api/export/markdown/{pid}")
assert resp.status_code == 200
body = resp.text
assert "# Root" in body
assert "# Child" in body
assert "Child body" in body
finally:
with __import__("app.db", fromlist=["get_conn"]).get_conn() as conn:
conn.execute("DELETE FROM pages WHERE id=? OR id=?", (sub_pid, pid))
conn.execute("DELETE FROM users WHERE id=?", (uid,))
conn.commit()
def test_v470_export_html(client):
"""HTML export is a standalone document with rendered blocks."""
pid, uid = _make_export_page(client, BLOCKS_EXPORT, title="HTML Page")
try:
resp = client.get(f"/api/export/html/{pid}")
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("text/html")
assert 'filename="HTML Page.html"' in resp.headers["content-disposition"]
body = resp.text
assert "<!DOCTYPE html>" in body
assert "<title>HTML Page</title>" in body
assert "<h1" in body
assert "<blockquote>" in body
assert "<pre>" in body and "print(1)" in body
assert "<details" in body
assert "callout" in body
assert "columns" in body
finally:
_cleanup_export(client, pid, uid)
def test_v470_export_pdf(client):
"""PDF export returns a valid PDF binary."""
pid, uid = _make_export_page(client, BLOCKS_EXPORT, title="PDF Page")
try:
resp = client.get(f"/api/export/pdf/{pid}")
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("application/pdf")
assert 'filename="PDF Page.pdf"' in resp.headers["content-disposition"]
body = resp.content
assert body[:5] == b"%PDF-"
assert len(body) > 500
finally:
_cleanup_export(client, pid, uid)
def test_v470_export_site_zip(client):
"""Static site export returns a zip containing index + page html."""
import io
import zipfile
pid, uid = _make_export_page(client, [{"id": "p1", "type": "paragraph", "content": "Root body"}], "Root")
sub_pid, _ = _make_export_page(client, [{"id": "s1", "type": "paragraph", "content": "Sub body"}], "Child", parent_id=pid)
try:
resp = client.get(f"/api/export/site/{pid}")
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("application/zip")
z = zipfile.ZipFile(io.BytesIO(resp.content))
names = z.namelist()
assert "index.html" in names
assert any(n.endswith(".html") for n in names)
index = z.read("index.html").decode("utf-8")
assert "Root" in index and "Child" in index
finally:
with __import__("app.db", fromlist=["get_conn"]).get_conn() as conn:
conn.execute("DELETE FROM pages WHERE id=? OR id=?", (sub_pid, pid))
conn.execute("DELETE FROM users WHERE id=?", (uid,))
conn.commit()
def test_v470_export_404(client):
"""Export endpoints return 404 for missing pages."""
resp = client.get("/api/export/markdown/999999")
assert resp.status_code == 404
# ── v4.7.2: Export resolves file & markdown page content (not just title) ──
# Tested at the service level: the export HTTP endpoints run behind an in-memory
# rate limiter whose store persists across the whole test process, so dozens of
# additional HTTP hits at the end of the suite trip 429. These assert on the
# service functions directly, which is where the content-resolution lives.
def _make_src_page(raw_md=None, file_info=None, title="Src Page", parent_id=None):
"""Insert a user + page storing raw markdown OR file metadata.
file_info = (rel_path_under_data_dir, mime). The real file is written to a
temp data dir whose path is exposed through ``FLOWDECK_DATA_DIR``.
"""
import json as _json
import uuid
from app.db import get_conn
login = "v472_" + uuid.uuid4().hex[:10]
with get_conn() as conn:
conn.execute("INSERT INTO users (login, full_name, email) VALUES (?, 'V472', ?)",
(login, login + "@t.com"))
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
if file_info:
rel, mime = file_info
payload = _json.dumps({"file_path": rel, "mime_type": mime, "size": 100})
fmt = "file"
else:
payload = raw_md
fmt = "markdown"
cur = conn.execute(
"INSERT INTO pages (workspace, title, content, content_format, parent_id) "
"VALUES (?, ?, ?, ?, ?)",
(login, title, payload, fmt, parent_id),
)
pid = cur.lastrowid
conn.commit()
return pid, uid
def _cleanup_src(pid, uid):
from app.db import get_conn
with get_conn() as conn:
conn.execute("DELETE FROM pages WHERE parent_id=? OR id=?", (pid, pid))
conn.execute("DELETE FROM users WHERE id=?", (uid,))
conn.commit()
def _load(pid):
from app.db import get_conn
with get_conn() as conn:
return dict(conn.execute("SELECT * FROM pages WHERE id=?", (pid,)).fetchone())
def test_v472_markdown_sourced_page_exports_body(client):
"""content_format='markdown' page exports its body, not just the title."""
from app.services.export import page_to_markdown
raw = "# Intro\n\nCeci est le contenu réel de la page.\n\n- point un\n- point deux\n"
pid, uid = _make_src_page(raw_md=raw, title="Page MD")
try:
md = page_to_markdown(_load(pid))
assert "# Page MD" in md
assert "Ceci est le contenu réel de la page." in md
assert "point un" in md
finally:
_cleanup_src(pid, uid)
def test_v472_markdown_sourced_page_renders_headings_to_html(client):
"""Raw-markdown page renders headings/lists in HTML (not line-wrapped)."""
from app.services.export import page_to_standalone_html
raw = "# Titre Principal\n\nParagraphe de contenu.\n\n## Sous section\n\n- a\n- b\n"
pid, uid = _make_src_page(raw_md=raw, title="Page HTML")
try:
body = page_to_standalone_html(_load(pid), include_children=False)
assert "<h1" in body and "Titre Principal" in body
assert "<h2" in body and "Sous section" in body
assert "<li>" in body and "Paragraphe de contenu" in body
finally:
_cleanup_src(pid, uid)
def test_v472_file_page_exports_uploaded_content(client, monkeypatch, tmp_path):
"""Uploaded markdown file page exports its real disk content (v4.7.2 fix)."""
from app.services.export import page_to_markdown, page_to_pdf_bytes, page_to_standalone_html
monkeypatch.setenv("FLOWDECK_DATA_DIR", str(tmp_path))
rel = "uploads/workspace_1/note.md"
disk = tmp_path / rel
disk.parent.mkdir(parents=True, exist_ok=True)
disk.write_text("# Note technique\n\ncontenu du fichier sur disque\n", encoding="utf-8")
pid, uid = _make_src_page(file_info=(rel, "text/markdown"), title="note.md")
try:
md = page_to_markdown(_load(pid), include_children=False)
assert "# note.md" in md
assert "Note technique" in md
assert "contenu du fichier sur disque" in md
html = page_to_standalone_html(_load(pid), include_children=False)
assert "Note technique" in html
assert "contenu du fichier sur disque" in html
pdf = page_to_pdf_bytes(_load(pid))
assert pdf[:5] == b"%PDF-"
finally:
_cleanup_src(pid, uid)
def test_v472_code_file_page_exported_as_code(client, monkeypatch, tmp_path):
"""A non-markdown text file page exports its content, not blank."""
from app.services.export import page_to_markdown, page_to_standalone_html
monkeypatch.setenv("FLOWDECK_DATA_DIR", str(tmp_path))
rel = "uploads/workspace_1/app.py"
disk = tmp_path / rel
disk.parent.mkdir(parents=True, exist_ok=True)
disk.write_text("def hello():\n return 'world'\n", encoding="utf-8")
pid, uid = _make_src_page(file_info=(rel, "text/x-python"), title="app.py")
try:
md = page_to_markdown(_load(pid), include_children=False)
assert "def hello():" in md and "world" in md
html = page_to_standalone_html(_load(pid), include_children=False)
assert "def hello():" in html
finally:
_cleanup_src(pid, uid)
def test_v472_binary_file_page_not_exported(client, monkeypatch, tmp_path):
"""Non-textual files (e.g. PDF uploads) export only the title, no garbage."""
from app.services.export import page_to_markdown
monkeypatch.setenv("FLOWDECK_DATA_DIR", str(tmp_path))
rel = "uploads/workspace_1/manual.pdf"
disk = tmp_path / rel
disk.parent.mkdir(parents=True, exist_ok=True)
disk.write_bytes(b"%PDF-1.4\nfake binary content")
pid, uid = _make_src_page(file_info=(rel, "application/pdf"), title="manual.pdf")
try:
md = page_to_markdown(_load(pid), include_children=False)
assert "# manual.pdf" in md
assert "fake binary" not in md # never dump binary into markdown
finally:
_cleanup_src(pid, uid)
_TABLE_MD = (
"# Titre\n\n"
"| ID | Nom | Score | Statut |\n"
"|----|:---:|------:|--------|\n"
"| 1 | Alice | 95.5 | ✅ Actif |\n"
"| 2 | Bob | 87.2 | 🟡 En attente |\n"
"| 3 | Charlie | 99.9 | ❌ Inactif |\n"
)
def test_v472_table_md_renders_real_html_table(client):
"""A GFM pipe table becomes a real <table> (header + cells + alignment)."""
from app.services.export import _page_blocks, blocks_to_html
pid, uid = _make_src_page(raw_md=_TABLE_MD, title="Table Page")
try:
blocks = _page_blocks(_load(pid))
tables = [b for b in blocks if b.get("type") == "table"]
assert tables, "expected a parsed table block"
t = tables[0]
assert t["align"] == ["left", "center", "right", "left"]
html = blocks_to_html(blocks)
assert "<table" in html
assert "<thead>" in html and "<th>" in html
assert "<tbody>" in html and "<td>" in html
assert 'text-align:center;' in html
assert "Alice" in html and "✅ Actif" in html
finally:
_cleanup_src(pid, uid)
def test_v472_table_md_standalone_html_has_table_css(client):
"""Standalone HTML export embeds the table and its stylesheet class."""
from app.services.export import page_to_standalone_html
pid, uid = _make_src_page(raw_md=_TABLE_MD, title="Table Page")
try:
html = page_to_standalone_html(_load(pid), include_children=False)
assert 'class="ftable"' in html
assert "Alice" in html
assert "Charlie" in html
finally:
_cleanup_src(pid, uid)
def test_v472_table_markdown_roundtrip(client):
"""A table block is re-emitted as a valid pipe table with a separator."""
from app.services.export import _page_blocks, blocks_to_markdown
pid, uid = _make_src_page(raw_md=_TABLE_MD, title="Table Page")
try:
md = blocks_to_markdown(_page_blocks(_load(pid)))
assert "| ID | Nom | Score | Statut |" in md
assert ":---:" in md and "---:" in md
assert "| 3 | Charlie" in md
finally:
_cleanup_src(pid, uid)
def test_v472_table_in_pdf(client, monkeypatch, tmp_path):
"""PDF export of a markdown table page returns a valid PDF (any engine)."""
from app.services.export import page_to_pdf_bytes
pid, uid = _make_src_page(raw_md=_TABLE_MD, title="Table Page")
try:
pdf = page_to_pdf_bytes(_load(pid))
assert pdf[:5] == b"%PDF-"
assert len(pdf) > 1000
finally:
_cleanup_src(pid, uid)
# ── v4.9.0: Collaboration — notifications, inline comments, mentions ──
def _v490_users(conn, n=3, prefix="v490"):
ids = []
for i in range(n):
login = f"{prefix}{i}"
conn.execute("INSERT INTO users (login, full_name, email) VALUES (?,?,?)",
(login, f"User {i}", f"{login}@t.com"))
ids.append(conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"])
conn.commit()
try:
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
except Exception:
pass
return ids
def _v490_page(conn, uid, title="Collab Page"):
cur = conn.execute("INSERT INTO pages (workspace, title, content, content_format) VALUES (?,?,?,?)",
(f"u{uid}", title, "[]", "blocks"))
conn.commit()
try:
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
except Exception:
pass
return cur.lastrowid
def test_v490_notifications_table(client):
from app.db import get_conn
with get_conn() as conn:
t = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='notifications'").fetchone()
assert t is not None
cols = [r[1] for r in conn.execute("PRAGMA table_info(comments)").fetchall()]
assert "target_type" in cols and "anchor_block_id" in cols
def test_v490_create_comment_with_mentions(client):
"""Adding an inline comment with @mention creates a notification for the target."""
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
uids = _v490_users(conn, 2) # v4900, v4901
pid = _v490_page(conn, uids[0])
targ_id = uids[1]
session = SessionManager.create_session({"id": uids[0], "login": "v4900", "is_admin": 0})
cookies = {"flowdeck_session": session}
r = client.post(f"/api/pages/{pid}/comments",
json={"body": "regarde ca @v4901", "anchor_block_id": "b1", "anchor_start": 0, "anchor_end": 5},
cookies=cookies)
assert r.status_code == 200
cid = r.json()["id"]
r = client.get(f"/api/pages/{pid}/comments", cookies=cookies)
assert r.status_code == 200
comments = r.json()["comments"]
assert len(comments) == 1
assert comments[0]["anchor_block_id"] == "b1"
with get_conn() as conn:
n = conn.execute("SELECT * FROM notifications WHERE user_id=? AND ntype='mention'", (targ_id,)).fetchone()
assert n is not None
assert n["resource_id"] == pid
def test_v490_comment_stores_anchor(client):
"""A comment with no mentions is stored with its inline anchor."""
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
uids = _v490_users(conn, 1, "v490a")
pid = _v490_page(conn, uids[0])
author_session = SessionManager.create_session({"id": uids[0], "login": "v490a0", "is_admin": 0})
r = client.post(f"/api/pages/{pid}/comments",
json={"body": "hello", "anchor_block_id": "b7", "anchor_start": 1, "anchor_end": 3},
cookies={"flowdeck_session": author_session})
assert r.status_code == 200
cid = r.json()["id"]
with get_conn() as conn:
row = conn.execute("SELECT * FROM comments WHERE id=?", (cid,)).fetchone()
assert row["anchor_block_id"] == "b7"
assert row["anchor_start"] == 1 and row["anchor_end"] == 3
def test_v490_notifications_center(client):
from app.auth.session import SessionManager
from app.db import get_conn
from app.services import notifications as notif
with get_conn() as conn:
uids = _v490_users(conn, 1, "v490b")
notif.create_notification(uids[0], None, "mention", "T", "M", "page", 1, "/pages/1")
session = SessionManager.create_session({"id": uids[0], "login": "v490b0", "is_admin": 0})
cookies = {"flowdeck_session": session}
r = client.get("/api/notifications", cookies=cookies)
assert r.status_code == 200
data = r.json()
assert data["unread"] >= 1
r = client.get("/api/notifications/unread-count", cookies=cookies)
assert r.json()["unread"] >= 1
r = client.post("/api/notifications/read", json={}, cookies=cookies)
assert r.status_code == 200
r = client.get("/api/notifications/unread-count", cookies=cookies)
assert r.json()["unread"] == 0
def test_v490_notification_prefs(client):
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
uids = _v490_users(conn, 1, "v490c")
session = SessionManager.create_session({"id": uids[0], "login": "v490c0", "is_admin": 0})
cookies = {"flowdeck_session": session}
r = client.get("/api/notifications/prefs", cookies=cookies)
assert r.status_code == 200
assert r.json()["prefs"]["comments"] is True
r = client.post("/api/notifications/prefs", json={"comments": False}, cookies=cookies)
assert r.status_code == 200
assert r.json()["prefs"]["comments"] is False
r = client.get("/api/notifications/prefs", cookies=cookies)
assert r.json()["prefs"]["comments"] is False
def test_v490_user_search(client):
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
uids = _v490_users(conn, 2, "v490d")
session = SessionManager.create_session({"id": uids[0], "login": "v490d0", "is_admin": 0})
r = client.get("/api/notifications/users/search?q=v490d", cookies={"flowdeck_session": session})
assert r.status_code == 200
assert len(r.json()["users"]) >= 2
def test_v490_resolve_and_delete_comment(client):
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
uids = _v490_users(conn, 1, "v490e")
pid = _v490_page(conn, uids[0])
session = SessionManager.create_session({"id": uids[0], "login": "v490e0", "is_admin": 0})
cookies = {"flowdeck_session": session}
r = client.post(f"/api/pages/{pid}/comments", json={"body": "a comment"}, cookies=cookies)
cid = r.json()["id"]
r = client.put(f"/api/comments/{cid}", json={"resolved": True}, cookies=cookies)
assert r.status_code == 200
r = client.delete(f"/api/comments/{cid}", cookies=cookies)
assert r.status_code == 200
r = client.get(f"/api/pages/{pid}/comments", cookies=cookies)
assert r.json()["comments"] == []
def test_v490_page_mentions_endpoint(client):
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
uids = _v490_users(conn, 2, "v490f")
pid = _v490_page(conn, uids[0])
tid = uids[1]
session = SessionManager.create_session({"id": uids[0], "login": "v490f0", "is_admin": 0})
r = client.post(f"/api/pages/{pid}/mentions", json={"text": "hey @v490f1"}, cookies={"flowdeck_session": session})
assert r.status_code == 200
assert r.json()["mentioned"] == [tid]
with get_conn() as conn:
assert conn.execute("SELECT COUNT(*) c FROM notifications WHERE user_id=? AND ntype='mention'", (tid,)).fetchone()["c"] >= 1
def test_v490_notifications_require_auth(client):
r = client.get("/api/notifications")
assert r.status_code == 401