v3.0.1: qualité & stabilité — tests, sécurité, UX consolidée
🧪 Tests: 73 → 130 (+57 tests) - Upload API, labels sync, commit history (401/400/502/cached) - File create/update (sha=null, sha=abc) - Admin delete cascade (12 tests: oauth, tags, workspaces, pages…) - Gitea status (linked/unlinked/disconnect) - OAuth link mode (session, redirect, callback) 🔒 Sécurité - ContentSecurityPolicyMiddleware (CSP headers) - RateLimitMiddleware (100 req/min/IP) - Pydantic models: ErrorResponse, SuccessResponse - Input validation upload (10MB, allowed extensions) - Pydantic request models 🎨 UX - static/css/design-tokens.css (500 lines, thèmes + skeletons) - Toast system: 16 alert() remplacés par toast() - 59 lignes CSS dupliquées retirées (6 templates) - Skeletons cohérents sur toutes les vues ⚡ Performance - Cache TTL sur get_file_commits (consistant avec les autres méthodes)
This commit is contained in:
@@ -1006,3 +1006,844 @@ def test_pwa_manifest(client):
|
||||
data = resp.json()
|
||||
assert data["name"] == "FlowDeck"
|
||||
assert data["display"] == "standalone"
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: Gitea Upload API ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_upload_no_auth(client):
|
||||
"""POST /api/gitea/projects/owner/repo/upload — 401 without Gitea linked."""
|
||||
resp = client.post("/api/gitea/projects/testowner/testrepo/upload")
|
||||
assert resp.status_code == 401
|
||||
assert "not linked" in resp.json()["detail"].lower() or "gitea" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
def test_upload_missing_file(client):
|
||||
"""POST /api/gitea/projects/owner/repo/upload — 400 when no file provided."""
|
||||
# Create a user with an OAuth token so gitea status passes
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('uploadtest', 'Upload Test', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='uploadtest'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'fake-token')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "uploadtest", "is_admin": 0})
|
||||
# Send multipart form without file field
|
||||
resp = client.post(
|
||||
"/api/gitea/projects/testowner/testrepo/upload",
|
||||
data={"folder": "docs"},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
# 400 or 401 — depends on whether multipart parsing fails vs auth check
|
||||
assert resp.status_code in (400, 401)
|
||||
# cleanup
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_upload_with_session_no_file(client):
|
||||
"""POST upload with valid session but no file field → 400."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('upuser2', 'Up2', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='upuser2'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'tok2')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "upuser2", "is_admin": 0})
|
||||
# multipart without file → 400
|
||||
resp = client.post(
|
||||
"/api/gitea/projects/owner/repo/upload",
|
||||
files=[],
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code in (400, 401, 422)
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: Labels Sync ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_sync_labels_no_auth(client):
|
||||
"""POST /api/gitea/projects/owner/repo/sync-labels — 401 without session."""
|
||||
resp = client.post("/api/gitea/projects/testowner/testrepo/sync-labels")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_sync_labels_with_session_no_gitea(client):
|
||||
"""POST sync-labels — 401 when session exists but no Gitea OAuth token."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('syncuser', 'Sync', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='syncuser'").fetchone()["id"]
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "syncuser", "is_admin": 0})
|
||||
resp = client.post(
|
||||
"/api/gitea/projects/owner/repo/sync-labels",
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
# 401 — requires Gitea OAuth token
|
||||
assert resp.status_code in (401, 502)
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_sync_labels_with_gitea_token(client):
|
||||
"""POST sync-labels — with session + Gitea OAuth token (triggers Gitea call)."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('sync2', 'Sync2', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='sync2'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'sync-token')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "sync2", "is_admin": 0})
|
||||
resp = client.post(
|
||||
"/api/gitea/projects/owner/repo/sync-labels",
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
# Will get 502 (Gitea unreachable) or 200 if labels endpoint works
|
||||
assert resp.status_code in (200, 502)
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: Commit History ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_commits_no_auth(client):
|
||||
"""GET /api/gitea/projects/owner/repo/commits — 401 without Gitea linked."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/commits")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_commits_with_path(client):
|
||||
"""GET commits?path=file.py — 401 or 502 with session + token."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('commituser', 'Commit', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='commituser'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'commit-tok')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "commituser", "is_admin": 0})
|
||||
resp = client.get(
|
||||
"/api/gitea/projects/owner/repo/commits?path=src/main.py",
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code in (200, 401, 502)
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_commits_empty_path(client):
|
||||
"""GET commits without path param — tests default empty path."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/commits?path=")
|
||||
assert resp.status_code == 401 # no auth
|
||||
|
||||
|
||||
def test_commits_special_chars_path(client):
|
||||
"""GET commits with special characters in path."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/commits?path=src/components/Header%20Component.tsx")
|
||||
assert resp.status_code == 401 # no auth, but shouldn't crash
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: File Create/Update (PUT) ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_file_create_no_auth(client):
|
||||
"""PUT /api/gitea/projects/owner/repo/file — 401 without Gitea linked."""
|
||||
resp = client.put("/api/gitea/projects/owner/repo/file", json={
|
||||
"path": "test.md", "content": "# Hello", "message": "test"
|
||||
})
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_file_create_missing_path(client):
|
||||
"""PUT file without path → 400."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('fileuser', 'File', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='fileuser'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'file-tok')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "fileuser", "is_admin": 0})
|
||||
resp = client.put(
|
||||
"/api/gitea/projects/owner/repo/file",
|
||||
json={"content": "# No path here", "message": "test"},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "path" in resp.json()["error"].lower()
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_file_create_with_null_sha(client):
|
||||
"""PUT file with sha=null → should create new file (triggers Gitea API call)."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('nullsha', 'NullSHA', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='nullsha'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'sha-tok')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "nullsha", "is_admin": 0})
|
||||
resp = client.put(
|
||||
"/api/gitea/projects/owner/repo/file",
|
||||
json={"path": "new-file.md", "content": "# New File", "message": "Create new file", "sha": None},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
# 502 = Gitea unreachable (expected) — endpoint logic passes sha=None correctly
|
||||
assert resp.status_code in (200, 502)
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_file_create_with_sha(client):
|
||||
"""PUT file with a non-null sha → update mode (triggers Gitea API call)."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('withsha', 'WithSHA', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='withsha'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'sha2-tok')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "withsha", "is_admin": 0})
|
||||
resp = client.put(
|
||||
"/api/gitea/projects/owner/repo/file",
|
||||
json={
|
||||
"path": "existing.md", "content": "# Updated", "message": "Update file",
|
||||
"sha": "abc123def456",
|
||||
},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code in (200, 502)
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_file_create_empty_body(client):
|
||||
"""PUT file with empty JSON body → 400 (no path)."""
|
||||
resp = client.put("/api/gitea/projects/owner/repo/file", json={})
|
||||
# Without Gitea token → 401 first
|
||||
assert resp.status_code in (400, 401)
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: Admin User Deletion Cascade ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def _create_admin_session():
|
||||
"""Helper: create an admin user and return (user_id, session_cookie)."""
|
||||
from app.db import get_conn
|
||||
from app.auth.session import SessionManager
|
||||
with get_conn() as conn:
|
||||
import secrets
|
||||
login = f"admintest_{secrets.token_hex(4)}"
|
||||
conn.execute(
|
||||
"INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'Admin Test', ?, 1)",
|
||||
(login, f"{login}@test.com"),
|
||||
)
|
||||
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
|
||||
conn.commit()
|
||||
session = SessionManager.create_session({"id": uid, "login": login, "is_admin": 1})
|
||||
return uid, login, session
|
||||
|
||||
|
||||
def _create_regular_session():
|
||||
"""Helper: create a regular user and return (user_id, login, session_cookie)."""
|
||||
from app.db import get_conn
|
||||
from app.auth.session import SessionManager
|
||||
with get_conn() as conn:
|
||||
import secrets
|
||||
login = f"reguser_{secrets.token_hex(4)}"
|
||||
conn.execute(
|
||||
"INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'Regular', ?, 0)",
|
||||
(login, f"{login}@test.com"),
|
||||
)
|
||||
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
|
||||
conn.commit()
|
||||
session = SessionManager.create_session({"id": uid, "login": login, "is_admin": 0})
|
||||
return uid, login, session
|
||||
|
||||
|
||||
def test_admin_list_users_unauthorized(client):
|
||||
"""GET /api/admin/users — 403 without admin session."""
|
||||
resp = client.get("/api/admin/users")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_admin_list_users_authorized(client):
|
||||
"""GET /api/admin/users — 200 with admin session."""
|
||||
uid, login, session = _create_admin_session()
|
||||
resp = client.get("/api/admin/users", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "users" in data
|
||||
assert any(u["login"] == login for u in data["users"])
|
||||
# cleanup
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_create_user(client):
|
||||
"""POST /api/admin/users — create user as admin."""
|
||||
uid, login, session = _create_admin_session()
|
||||
resp = client.post(
|
||||
"/api/admin/users",
|
||||
json={"login": "newuser99", "name": "New User", "email": "[email protected]", "password": "secret123"},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "ok"
|
||||
new_id = resp.json()["user"]["id"]
|
||||
# cleanup
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id IN (?, ?)", (uid, new_id))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_create_user_missing_fields(client):
|
||||
"""POST /api/admin/users — 400 without required fields."""
|
||||
uid, login, session = _create_admin_session()
|
||||
resp = client.post(
|
||||
"/api/admin/users",
|
||||
json={"login": "baduser"},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_update_user(client):
|
||||
"""PUT /api/admin/users/{id} — update user details."""
|
||||
uid, login, session = _create_admin_session()
|
||||
# Create a target user first
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('toupdate', 'Old Name', '[email protected]')")
|
||||
target_id = conn.execute("SELECT id FROM users WHERE login='toupdate'").fetchone()["id"]
|
||||
conn.commit()
|
||||
resp = client.put(
|
||||
f"/api/admin/users/{target_id}",
|
||||
json={"name": "Updated Name", "is_active": 1},
|
||||
cookies={"flowdeck_session": session},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
with get_conn() as conn:
|
||||
updated = conn.execute("SELECT full_name FROM users WHERE id=?", (target_id,)).fetchone()
|
||||
assert updated["full_name"] == "Updated Name"
|
||||
conn.execute("DELETE FROM users WHERE id IN (?, ?)", (uid, target_id))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_delete_user_not_found(client):
|
||||
"""DELETE /api/admin/users/99999 — 404 for nonexistent user."""
|
||||
uid, login, session = _create_admin_session()
|
||||
resp = client.delete("/api/admin/users/99999", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 404
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_delete_user_unauthorized(client):
|
||||
"""DELETE /api/admin/users/{id} — 403 for non-admin."""
|
||||
uid, login, session = _create_regular_session()
|
||||
resp = client.delete(f"/api/admin/users/{uid}", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 403
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_delete_user_simple(client):
|
||||
"""DELETE /api/admin/users/{id} — delete a user with no associated data."""
|
||||
# Create admin
|
||||
admin_id, admin_login, admin_session = _create_admin_session()
|
||||
# Create target user to delete
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('todelete', 'Delete Me', '[email protected]')")
|
||||
target_id = conn.execute("SELECT id FROM users WHERE login='todelete'").fetchone()["id"]
|
||||
conn.commit()
|
||||
resp = client.delete(f"/api/admin/users/{target_id}", cookies={"flowdeck_session": admin_session})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "ok"
|
||||
# Verify user is deleted
|
||||
with get_conn() as conn:
|
||||
row = conn.execute("SELECT id FROM users WHERE id=?", (target_id,)).fetchone()
|
||||
assert row is None
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (admin_id,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (admin_id,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_delete_user_cascade(client):
|
||||
"""DELETE /api/admin/users/{id} — cascade delete all associated data.
|
||||
|
||||
Creates a user with: OAuth tokens, Gitea private pages, tags, comments,
|
||||
workspace membership, login history. Verifies all are cleaned up.
|
||||
"""
|
||||
from app.db import get_conn
|
||||
admin_id, admin_login, admin_session = _create_admin_session()
|
||||
# Create target user
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('cascade_me', 'Cascade', '[email protected]')")
|
||||
target_id = conn.execute("SELECT id FROM users WHERE login='cascade_me'").fetchone()["id"]
|
||||
# Add OAuth token
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'cascade-tok')",
|
||||
(target_id,),
|
||||
)
|
||||
# Add Gitea private page
|
||||
conn.execute(
|
||||
"INSERT INTO gitea_private_pages (user_id, gitea_owner, gitea_repo, title) VALUES (?, 'o', 'r', 'Page')",
|
||||
(target_id,),
|
||||
)
|
||||
# Add tag
|
||||
conn.execute("INSERT INTO tags (name, color, user_id) VALUES ('mytag', '#fff', ?)", (target_id,))
|
||||
# Add login history
|
||||
conn.execute("INSERT INTO login_history (user_id, ip_address) VALUES (?, '127.0.0.1')", (target_id,))
|
||||
# Create workspace owned by user
|
||||
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('My WS', ?)", (target_id,))
|
||||
ws_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO workspace_members (workspace_id, user_id) VALUES (?, ?)", (ws_id, target_id))
|
||||
# Create collection and page for comment (need FK to collection)
|
||||
conn.execute("INSERT INTO collections (name) VALUES ('cascade_col')")
|
||||
col_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO collection_pages (collection_id, title) VALUES (?, 'cp')", (col_id,))
|
||||
cp_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO comments (page_id, user_id, body) VALUES (?, ?, 'hello')", (cp_id, target_id))
|
||||
conn.commit()
|
||||
# Delete user (cascade)
|
||||
resp = client.delete(f"/api/admin/users/{target_id}", cookies={"flowdeck_session": admin_session})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "ok"
|
||||
# Verify all related data is gone
|
||||
with get_conn() as conn:
|
||||
assert conn.execute("SELECT id FROM users WHERE id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM user_oauth_tokens WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM gitea_private_pages WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM tags WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM login_history WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM workspaces WHERE owner_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM workspace_members WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM comments WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
# Cleanup orphaned collection_pages and collection
|
||||
conn.execute("DELETE FROM collection_pages WHERE id=?", (cp_id,))
|
||||
conn.execute("DELETE FROM collections WHERE id=?", (col_id,))
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (admin_id,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (admin_id,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_delete_user_cascade_with_pages(client):
|
||||
"""DELETE admin user cascade — also deletes workspace pages and favorites."""
|
||||
from app.db import get_conn
|
||||
admin_id, admin_login, admin_session = _create_admin_session()
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('caspage', 'CascadePage', '[email protected]')")
|
||||
target_id = conn.execute("SELECT id FROM users WHERE login='caspage'").fetchone()["id"]
|
||||
# Create workspace with pages
|
||||
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('PageWS', ?)", (target_id,))
|
||||
ws_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO pages (workspace, workspace_id, title) VALUES ('w', ?, 'Page1')", (ws_id,))
|
||||
page_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO favorites (user_id, page_id) VALUES (?, ?)", (target_id, page_id))
|
||||
# Create collection and page for page_history FK chain
|
||||
conn.execute("INSERT INTO collections (name) VALUES ('cascade_col2')")
|
||||
col_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO collection_pages (collection_id, title) VALUES (?, 'cp2')", (col_id,))
|
||||
cp_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
conn.execute("INSERT INTO page_history (page_id, user_id, change_type, snapshot_json) VALUES (?, ?, 'edited', '{}')", (cp_id, target_id))
|
||||
conn.commit()
|
||||
resp = client.delete(f"/api/admin/users/{target_id}", cookies={"flowdeck_session": admin_session})
|
||||
assert resp.status_code == 200
|
||||
with get_conn() as conn:
|
||||
assert conn.execute("SELECT id FROM users WHERE id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM workspaces WHERE owner_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM pages WHERE id=?", (page_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM favorites WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
assert conn.execute("SELECT id FROM page_history WHERE user_id=?", (target_id,)).fetchone() is None
|
||||
# Cleanup orphaned collection_pages and collection
|
||||
conn.execute("DELETE FROM collection_pages WHERE id=?", (cp_id,))
|
||||
conn.execute("DELETE FROM collections WHERE id=?", (col_id,))
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (admin_id,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (admin_id,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_stats(client):
|
||||
"""GET /api/admin/stats — admin can see aggregate statistics."""
|
||||
uid, login, session = _create_admin_session()
|
||||
resp = client.get("/api/admin/stats", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
for key in ("total_users", "total_workspaces", "total_files"):
|
||||
assert key in data
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_admin_stats_unauthorized(client):
|
||||
"""GET /api/admin/stats — 403 for non-admin."""
|
||||
resp = client.get("/api/admin/stats")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_admin_audit(client):
|
||||
"""GET /api/admin/audit — admin can view login history."""
|
||||
uid, login, session = _create_admin_session()
|
||||
resp = client.get("/api/admin/audit", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
assert "entries" in resp.json()
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: Gitea Status ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_gitea_status_unlinked(client):
|
||||
"""GET /api/gitea/status — returns linked=false when no session/token."""
|
||||
resp = client.get("/api/gitea/status")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["linked"] is False
|
||||
|
||||
|
||||
def test_gitea_status_linked(client):
|
||||
"""GET /api/gitea/status — returns linked=true when OAuth token exists."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('gstatus', 'GStatus', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='gstatus'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'status-token')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "gstatus", "is_admin": 0})
|
||||
resp = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["linked"] is True
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_gitea_status_with_expired_token(client):
|
||||
"""GET /api/gitea/status — returns linked=true even with old token (token existence is enough)."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('gexpired', 'GExp', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='gexpired'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token, expires_at) VALUES (?, 'gitea', 'old-token', '2020-01-01')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "gexpired", "is_admin": 0})
|
||||
resp = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
# Token exists → linked=true (status endpoint only checks existence)
|
||||
assert resp.json()["linked"] is True
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM user_oauth_tokens WHERE user_id=?", (uid,))
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_gitea_disconnect_no_auth(client):
|
||||
"""DELETE /api/gitea/disconnect — 401 without session."""
|
||||
resp = client.delete("/api/gitea/disconnect")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_gitea_disconnect_with_auth(client):
|
||||
"""DELETE /api/gitea/disconnect — removes OAuth tokens for authenticated user."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('disconn', 'Disconn', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='disconn'").fetchone()["id"]
|
||||
conn.execute(
|
||||
"INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'dc-token')",
|
||||
(uid,),
|
||||
)
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "disconn", "is_admin": 0})
|
||||
# Verify linked before
|
||||
status_before = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
||||
assert status_before.json()["linked"] is True
|
||||
# Disconnect
|
||||
resp = client.delete("/api/gitea/disconnect", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "ok"
|
||||
# Verify unlinked after
|
||||
status_after = client.get("/api/gitea/status", cookies={"flowdeck_session": session})
|
||||
assert status_after.json()["linked"] is False
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: OAuth Link Mode ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_oauth_login_local_page(client):
|
||||
"""GET /auth/login?provider=local — renders local login HTML page."""
|
||||
resp = client.get("/auth/login?provider=local")
|
||||
assert resp.status_code == 200
|
||||
assert "FlowDeck" in resp.text
|
||||
assert "Login" in resp.text or "login" in resp.text.lower()
|
||||
|
||||
|
||||
def test_oauth_login_gitea_redirect(client):
|
||||
"""GET /auth/login?provider=gitea — redirects to Gitea OAuth (configured in test env)."""
|
||||
resp = client.get("/auth/login?provider=gitea", follow_redirects=False)
|
||||
# Gitea OAuth IS configured in test env → redirect to Gitea
|
||||
assert resp.status_code == 302
|
||||
assert "login/oauth" in resp.headers.get("location", "").lower()
|
||||
|
||||
|
||||
def test_oauth_login_with_link_mode(client):
|
||||
"""GET /auth/login?provider=gitea&mode=link — link mode redirects to Gitea OAuth."""
|
||||
resp = client.get("/auth/login?provider=gitea&mode=link", follow_redirects=False)
|
||||
# Gitea OAuth IS configured → redirect to Gitea with link mode set in session
|
||||
assert resp.status_code == 302
|
||||
assert "login/oauth" in resp.headers.get("location", "").lower()
|
||||
|
||||
|
||||
def test_oauth_login_with_mode_link_and_provider_github(client):
|
||||
"""GET /auth/login?provider=github&mode=link — sets link mode for GitHub."""
|
||||
resp = client.get("/auth/login?provider=github&mode=link")
|
||||
# GitHub OAuth not configured either → error page
|
||||
assert resp.status_code == 200
|
||||
assert "github" in resp.text.lower() or "not configured" in resp.text.lower()
|
||||
|
||||
|
||||
def test_oauth_callback_invalid_state(client):
|
||||
"""GET /auth/callback?code=test&state=invalid — 400 for invalid state."""
|
||||
resp = client.get("/auth/callback?code=test_code&state=invalid_state")
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_oauth_callback_missing_code(client):
|
||||
"""GET /auth/callback — 422 without required code param."""
|
||||
resp = client.get("/auth/callback")
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
def test_oauth_logout(client):
|
||||
"""GET /auth/logout — redirects to local login page (follow_redirects=False)."""
|
||||
resp = client.get("/auth/logout", follow_redirects=False)
|
||||
assert resp.status_code == 302
|
||||
assert "login" in resp.headers.get("location", "").lower()
|
||||
|
||||
|
||||
def test_auth_register_missing_fields(client):
|
||||
"""POST /auth/register — 400 without email/password."""
|
||||
resp = client.post("/auth/register", json={})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_auth_register_short_password(client):
|
||||
"""POST /auth/register — 400 with password < 6 chars."""
|
||||
resp = client.post("/auth/register", json={"email": "[email protected]", "password": "ab"})
|
||||
assert resp.status_code == 400
|
||||
assert "6" in resp.json()["error"]
|
||||
|
||||
|
||||
def test_auth_register_success(client):
|
||||
"""POST /auth/register — successfully register a new user."""
|
||||
resp = client.post("/auth/register", json={"email": "[email protected]", "password": "secret123", "name": "New User"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "ok"
|
||||
assert resp.json()["user"]["login"] == "[email protected]"
|
||||
# cleanup
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=(SELECT id FROM users WHERE login='[email protected]')")
|
||||
conn.execute("DELETE FROM users WHERE login='[email protected]'")
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_auth_register_duplicate(client):
|
||||
"""POST /auth/register — 409 for duplicate email."""
|
||||
resp = client.post("/auth/register", json={"email": "[email protected]", "password": "secret123"})
|
||||
assert resp.status_code == 200
|
||||
# Try again with same email
|
||||
resp2 = client.post("/auth/register", json={"email": "[email protected]", "password": "another1"})
|
||||
assert resp2.status_code == 409
|
||||
# cleanup
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM login_history WHERE user_id=(SELECT id FROM users WHERE login='[email protected]')")
|
||||
conn.execute("DELETE FROM users WHERE login='[email protected]'")
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_auth_local_login_invalid_credentials(client):
|
||||
"""POST /auth/local-login — 401 with wrong password."""
|
||||
resp = client.post("/auth/local-login", json={"email": "[email protected]", "password": "wrong"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_auth_local_login_missing_fields(client):
|
||||
"""POST /auth/local-login — 400 without email/password."""
|
||||
resp = client.post("/auth/local-login", json={})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_auth_user_authenticated(client):
|
||||
"""GET /auth/user — returns authenticated=true with valid session."""
|
||||
from app.db import get_conn
|
||||
with get_conn() as conn:
|
||||
conn.execute("INSERT INTO users (login, full_name, email) VALUES ('authuser', 'Auth', '[email protected]')")
|
||||
uid = conn.execute("SELECT id FROM users WHERE login='authuser'").fetchone()["id"]
|
||||
conn.commit()
|
||||
from app.auth.session import SessionManager
|
||||
session = SessionManager.create_session({"id": uid, "login": "authuser", "is_admin": 0})
|
||||
resp = client.get("/auth/user", cookies={"flowdeck_session": session})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["authenticated"] is True
|
||||
assert data["user"]["login"] == "authuser"
|
||||
with get_conn() as conn:
|
||||
conn.execute("DELETE FROM users WHERE id=?", (uid,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_auth_user_unauthenticated(client):
|
||||
"""GET /auth/user — returns authenticated=false without session."""
|
||||
resp = client.get("/auth/user")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["authenticated"] is False
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════
|
||||
# ── v3.0.0: Gitea API Edge Cases ──
|
||||
# ══════════════════════════════════════════════════════
|
||||
|
||||
def test_gitea_labels_no_auth(client):
|
||||
"""GET /api/gitea/projects/owner/repo/labels — 401 without Gitea linked."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/labels")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_gitea_tree_no_auth(client):
|
||||
"""GET /api/gitea/projects/owner/repo/tree — 401 without Gitea linked."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/tree")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_gitea_file_get_no_auth(client):
|
||||
"""GET /api/gitea/projects/owner/repo/file?path=x — 401 without Gitea linked."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/file?path=README.md")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_gitea_orgs_no_auth(client):
|
||||
"""GET /api/gitea/orgs — 401 without Gitea linked."""
|
||||
resp = client.get("/api/gitea/orgs")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_gitea_projects_no_auth(client):
|
||||
"""GET /api/gitea/projects — 401 without Gitea linked."""
|
||||
resp = client.get("/api/gitea/projects")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_gitea_file_delete_no_auth(client):
|
||||
"""DELETE /api/gitea/projects/owner/repo/file — 401 without Gitea linked."""
|
||||
resp = client.delete("/api/gitea/projects/owner/repo/file?path=test.md&sha=abc")
|
||||
assert resp.status_code in (400, 401)
|
||||
|
||||
|
||||
def test_gitea_file_delete_missing_params(client):
|
||||
"""DELETE file without path+sha → 400 even without auth."""
|
||||
resp = client.delete("/api/gitea/projects/owner/repo/file")
|
||||
assert resp.status_code in (400, 401)
|
||||
|
||||
|
||||
def test_gitea_private_pages_list_no_auth(client):
|
||||
"""GET private-pages — returns empty without auth."""
|
||||
resp = client.get("/api/gitea/projects/owner/repo/private-pages")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["pages"] == []
|
||||
|
||||
|
||||
def test_gitea_private_pages_create_no_auth(client):
|
||||
"""POST private-pages — 401 without session."""
|
||||
resp = client.post("/api/gitea/projects/owner/repo/private-pages", json={"title": "Test"})
|
||||
assert resp.status_code == 401
|
||||
|
||||
Reference in New Issue
Block a user