"""FlowDeck — Comprehensive tests v1.3.0.""" import json import os import tempfile import pytest from conftest import anon, anon_csrf, login_test_client 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" os.environ["PUBLIC_API_INSECURE_OK"] = "true" # Point the process-wide settings singleton at OUR temp DB (xdist-safe). from app.config import settings settings.database_url = f"sqlite:///{db_path}" settings.rate_limit_enabled = False settings.public_api_insecure_ok = True from app.db import init_db from app.main import app init_db() # A default user id=1 so tests that reference the implicit user (favorites, # workspace ownership) satisfy foreign keys without relying on leaked state. from app.db import get_conn with get_conn() as conn: conn.execute( "INSERT OR IGNORE INTO users (id, login, full_name, email, is_admin) " "VALUES (1, 'tester', 'Tester', 'tester@test.dev', 1)" ) conn.commit() yield login_test_client(TestClient(app)) try: os.unlink(db_path) except PermissionError: pass # Windows: file may still be open in another thread # ── 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): anon(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): anon(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): anon(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): anon(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 "" 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" assert any(i.get("sizes") == "192x192" for i in data["icons"]) def test_pwa_manifest_static_assets(): """v6.0.0: the static manifest and every referenced icon must exist on disk.""" import json from pathlib import Path root = Path(__file__).resolve().parent.parent manifest_path = root / "static" / "manifest.json" assert manifest_path.is_file(), "static/manifest.json missing" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) assert manifest["name"] == "FlowDeck" assert "icons" in manifest and len(manifest["icons"]) >= 8 for icon in manifest["icons"]: src = icon["src"].lstrip("/") assert (root / src).is_file(), f"icon missing: {src}" # ══════════════════════════════════════════════════════ # ── 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', 'up@test.com')") 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', 'up2@t.com')") 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', 's@t.com')") 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', 's2@t.com')") 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', 'c@t.com')") 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', 'f@t.com')") 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', 'ns@t.com')") 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', 'ws@t.com')") 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): anon(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": "new@test.com", "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', 'old@t.com')") 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', 'del@t.com')") 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', 'cas@t.com')") 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', 'cp@t.com')") 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): anon(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', 'gs@t.com')") 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', 'ge@t.com')") 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): anon_csrf(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', 'dc@t.com')") 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(monkeypatch): """get_redirect_uri() derives the URI from the Host header when used directly. NOTE: an explicit ``OAUTH_REDIRECT_URI`` env override (as present in the project ``.env``) takes priority by design. This test pins that override to empty so it exercises the Host-header derivation path in isolation. """ from app.config import settings from app.routers.auth import get_redirect_uri monkeypatch.setattr(settings, "oauth_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": "test@test.com", "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": "new_user@test.com", "password": "secret123", "name": "New User"}) assert resp.status_code == 200 assert resp.json()["status"] == "ok" assert resp.json()["user"]["login"] == "new_user@test.com" # 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='new_user@test.com')") conn.execute("DELETE FROM users WHERE login='new_user@test.com'") conn.commit() def test_auth_register_duplicate(client): """POST /auth/register — 409 for duplicate email.""" resp = client.post("/auth/register", json={"email": "dup_user@test.com", "password": "secret123"}) assert resp.status_code == 200 # Try again with same email resp2 = client.post("/auth/register", json={"email": "dup_user@test.com", "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='dup_user@test.com')") conn.execute("DELETE FROM users WHERE login='dup_user@test.com'") 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": "nonexistent@test.com", "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', 'au@t.com')") 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): anon(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): anon_csrf(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: 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): anon(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=§ion=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": "test@test.com", "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): anon(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 vendored Chart.js (A20 : plus de 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 "/static/js/vendor/chart.umd.js" in resp.text assert "cdn.jsdelivr" not 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 "" 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§ion=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 "" in body assert "HTML Page" in body assert "" in body assert "
" in body and "print(1)" in body
        assert " 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 "" 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
(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 "" in html and "" in html and "
" in html assert "
" 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 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): anon(client) r = client.get("/api/notifications") assert r.status_code == 401