bruno
d125eb399e
- A3 : helper `_require_user_id()` (401 sans session) sur profile/password/token/forge ; `PUT /api/user/password` exige `current_password` vérifié ; `/api/user` sorti de la liste CSRF exemptée - A4 : `POST /api/v1/token` → 401 sans session, chemin legacy `user_id=0` supprimé - A5 : CRUD membres d'espace (POST/PUT/DELETE) : session + rôle admin de l'espace (ou admin global), placeholder user créé en `is_admin=0`, GET membres sans session → 401 - A6 : `_require_view` → 404 et `_require_edit` → 401 quand il n'y a pas de session (fin du legacy single-user sur les collections) - A7 : création ET lecture de page → 401 sans session (PermissionManager conservé) ; `/board/api/pages` sorti de `EXCLUDED_PATHS` ; header CSRF manquant ajouté sur setItemIcon (local_workspace) - A8 : seed admin sans mot de passe codé en dur — aléatoire au premier boot loggé une fois, ou `FLOWDECK_ADMIN_PASSWORD` ; re-seed seulement si absent - tests : client connecté par défaut via `_TestSessionAuth` (session + CSRF injectés à la volée, jamais dans le cookie jar → plus de CookieConflict), helper `anon()` sur les 40 tests d'anonymat ; 1016/1016 verts, `ruff check app tests` OK
1078 lines
42 KiB
Python
1078 lines
42 KiB
Python
"""FlowDeck — v7.3.0 Teamspaces + Verified pages + collab polish.
|
|
|
|
Covers migration 29, teamspace CRUD + private 404 + members/roles, verified
|
|
badge with expiry + notification sweep, wiki home, follows, comment
|
|
reactions, guest shares (expiry/revocation) and page views, plus the
|
|
mermaid/equation/progress block renderers.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import secrets
|
|
|
|
from conftest import anon
|
|
|
|
from app.db import get_conn
|
|
|
|
# ── helpers ────────────────────────────────────────────────────────────────
|
|
|
|
def _user(login=None, is_admin=0):
|
|
login = login or f"v73_{secrets.token_hex(4)}"
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, email, is_admin) VALUES (?,?,?,?)",
|
|
(login, login, f"{login}@test.com", is_admin))
|
|
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
|
|
conn.commit()
|
|
from app.auth.session import SessionManager
|
|
return uid, login, {"flowdeck_session": SessionManager.create_session(
|
|
{"id": uid, "login": login})}
|
|
|
|
|
|
def _workspace(owner_id, name="WS"):
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO workspaces (name, owner_id) VALUES (?,?)",
|
|
(name, owner_id))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
def _member(ws_id, uid, role="editor"):
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT OR IGNORE INTO workspace_members (workspace_id, user_id, role)"
|
|
" VALUES (?,?,?)", (ws_id, uid, role))
|
|
conn.commit()
|
|
|
|
|
|
def _page(ws_id, title="Doc", teamspace_id=None, owner_id=None, content="hello"):
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO pages (workspace, title, workspace_id, content,"
|
|
" content_format, teamspace_id) VALUES ('',?,?,?,'blocks',?)",
|
|
(title, ws_id, content, teamspace_id))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
def _collection(ws_id, name="Tasks"):
|
|
"""``ws_id=None`` creates a throwaway workspace (blocks tests need only the FK)."""
|
|
if not ws_id:
|
|
ws_id = _workspace(_user()[0], f"blkw{secrets.token_hex(3)}")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO collections (name, workspace_id) VALUES (?,?)",
|
|
(name, ws_id))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
def _add_rows(cid, values_by_pid):
|
|
import json
|
|
with get_conn() as conn:
|
|
for vals in values_by_pid:
|
|
conn.execute("INSERT INTO collection_pages (collection_id, title, position,"
|
|
" property_values_json) VALUES (?,?,0,?)",
|
|
(cid, f"r{secrets.token_hex(2)}", json.dumps(vals)))
|
|
conn.commit()
|
|
|
|
|
|
# ── migration 29 ───────────────────────────────────────────────────────────
|
|
|
|
def test_migration_29_tables_exist(client):
|
|
with get_conn() as conn:
|
|
names = {r["name"] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
|
|
for t in ("teamspaces", "teamspace_members", "page_verifications",
|
|
"comment_reactions", "page_follows", "guest_shares", "page_views"):
|
|
assert t in names, f"missing {t}"
|
|
|
|
|
|
def test_migration_29_teamspace_columns(client):
|
|
with get_conn() as conn:
|
|
for table in ("pages", "collections"):
|
|
cols = {r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()}
|
|
assert "teamspace_id" in cols, table
|
|
|
|
|
|
# ── teamspaces ─────────────────────────────────────────────────────────────
|
|
|
|
def test_teamspace_create_and_list(client):
|
|
uid, login, c = _user()
|
|
wid = _workspace(uid)
|
|
r = client.post("/api/v2/wiki/teamspaces", cookies=c,
|
|
json={"workspace_id": wid, "name": "Engineering"})
|
|
assert r.status_code == 201, r.text
|
|
assert r.json()["name"] == "Engineering"
|
|
lst = client.get(f"/api/v2/wiki/teamspaces?workspace_id={wid}", cookies=c)
|
|
assert lst.status_code == 200
|
|
spaces = lst.json()["teamspaces"]
|
|
assert len(spaces) == 1
|
|
assert spaces[0]["role"] == "owner"
|
|
|
|
|
|
def test_teamspace_creator_is_owner(client):
|
|
from app.services.wiki import get_teamspace_role
|
|
uid, _, _ = _user()
|
|
wid = _workspace(uid)
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by)"
|
|
" VALUES (?,?,?)", (wid, "Docs", uid))
|
|
conn.commit()
|
|
assert get_teamspace_role(uid, cur.lastrowid) == "owner"
|
|
|
|
|
|
def test_teamspace_duplicate_name_conflict(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
assert client.post("/api/v2/wiki/teamspaces", cookies=c,
|
|
json={"workspace_id": wid, "name": "Dup"}).status_code == 201
|
|
assert client.post("/api/v2/wiki/teamspaces", cookies=c,
|
|
json={"workspace_id": wid, "name": "Dup"}).status_code == 409
|
|
|
|
|
|
def test_teamspace_requires_name(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
assert client.post("/api/v2/wiki/teamspaces", cookies=c,
|
|
json={"workspace_id": wid, "name": " "}).status_code == 400
|
|
|
|
|
|
def test_teamspace_requires_auth(client):
|
|
anon(client)
|
|
assert client.get("/api/v2/wiki/teamspaces?workspace_id=1").status_code == 401
|
|
assert client.post("/api/v2/wiki/teamspaces", json={}).status_code == 401
|
|
|
|
|
|
def test_teamspace_private_is_404_for_outsider(client):
|
|
owner, _, oc = _user()
|
|
other, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "Secret",
|
|
"private": True}).json()["id"]
|
|
# owner sees it
|
|
assert client.get(f"/api/v2/wiki/teamspaces/{tsid}", cookies=oc).status_code == 200
|
|
# non-member of a private teamspace gets 404 (not 403)
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role)"
|
|
" VALUES (?,?,'viewer')", (wid, other))
|
|
conn.commit()
|
|
_, _, otherc = _user()
|
|
r = client.get(f"/api/v2/wiki/teamspaces/{tsid}", cookies=otherc)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_teamspace_public_visible_to_workspace_members(client):
|
|
owner, _, oc = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "Open"}).json()["id"]
|
|
_member(wid, 999999 - 999999 + 0 or 0) if False else None
|
|
other_uid, _, otherc = _user()
|
|
_member(wid, other_uid, "viewer")
|
|
assert client.get(f"/api/v2/wiki/teamspaces/{tsid}", cookies=otherc).status_code == 200
|
|
lst = client.get(f"/api/v2/wiki/teamspaces?workspace_id={wid}", cookies=otherc)
|
|
assert [t["name"] for t in lst.json()["teamspaces"]] == ["Open"]
|
|
|
|
|
|
def test_teamspace_list_empty_for_non_member(client):
|
|
owner, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
_, _, c = _user()
|
|
assert client.get(f"/api/v2/wiki/teamspaces?workspace_id={wid}", cookies=c).json()[
|
|
"teamspaces"] == []
|
|
|
|
|
|
def test_teamspace_members_add_update_remove(client):
|
|
owner, _, oc = _user()
|
|
member, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "Team"}).json()["id"]
|
|
r = client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}", cookies=oc,
|
|
json={"role": "viewer"})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["role"] == "viewer"
|
|
lst = client.get(f"/api/v2/wiki/teamspaces/{tsid}/members", cookies=oc)
|
|
assert {m["user_id"] for m in lst.json()["members"]} == {owner, member}
|
|
# update
|
|
assert client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}", cookies=oc,
|
|
json={"role": "editor"}).json()["role"] == "editor"
|
|
# remove
|
|
assert client.delete(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}",
|
|
cookies=oc).status_code == 200
|
|
assert client.delete(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}",
|
|
cookies=oc).status_code == 404
|
|
|
|
|
|
def test_teamspace_member_bad_role(client):
|
|
owner, _, oc = _user()
|
|
member, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "T"}).json()["id"]
|
|
assert client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}", cookies=oc,
|
|
json={"role": "godmode"}).status_code == 400
|
|
|
|
|
|
def test_teamspace_member_unknown_user(client):
|
|
owner, _, oc = _user()
|
|
wid = _workspace(owner)
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "T"}).json()["id"]
|
|
assert client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/999999", cookies=oc,
|
|
json={"role": "editor"}).status_code == 404
|
|
|
|
|
|
def test_teamspace_role_helpers(client):
|
|
from app.services import wiki
|
|
owner, _, _ = _user()
|
|
member, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, member, "viewer")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO teamspaces (workspace_id, name, private, created_by)"
|
|
" VALUES (?,?,0,?)", (wid, "Pub", owner))
|
|
conn.commit()
|
|
tsid = cur.lastrowid
|
|
assert wiki.can_read_teamspace(member, tsid) is True
|
|
assert wiki.can_write_teamspace(member, tsid) is False
|
|
assert wiki.can_write_teamspace(owner, tsid) is True
|
|
assert wiki.get_teamspace_role(999999, tsid) is None
|
|
assert wiki.get_teamspace_role(None, tsid) is None
|
|
|
|
|
|
# ── verified pages ─────────────────────────────────────────────────────────
|
|
|
|
def test_verify_page_badge(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
r = client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=c, json={"note": "reviewed"})
|
|
assert r.status_code == 200, r.text
|
|
v = r.json()["verification"]
|
|
assert v["page_id"] == pid
|
|
assert v["active"] is True
|
|
assert v["note"] == "reviewed"
|
|
got = client.get(f"/api/v2/wiki/pages/{pid}/verification", cookies=c)
|
|
assert got.json()["verification"]["verified_by"] == uid
|
|
|
|
|
|
def test_verify_page_requires_editor(client):
|
|
owner, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
pid = _page(wid)
|
|
viewer, _, vc = _user()
|
|
_member(wid, viewer, "viewer")
|
|
assert client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=vc).status_code == 403
|
|
|
|
|
|
def test_verify_page_admin_override(client):
|
|
admin, _, ac = _user(is_admin=1)
|
|
owner, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
pid = _page(wid)
|
|
assert client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=ac).status_code == 200
|
|
|
|
|
|
def test_verify_page_in_private_teamspace(client):
|
|
owner, _, oc = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "Priv",
|
|
"private": True}).json()["id"]
|
|
pid = _page(wid, "Secret", teamspace_id=tsid)
|
|
# owner is teamspace owner → can verify
|
|
assert client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=oc).status_code == 200
|
|
|
|
|
|
def test_verify_page_unknown(client):
|
|
uid, _, c = _user()
|
|
assert client.post("/api/v2/wiki/pages/999999/verify", cookies=c).status_code == 404
|
|
|
|
|
|
def test_unverify_page(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=c)
|
|
assert client.delete(f"/api/v2/wiki/pages/{pid}/verify", cookies=c).status_code == 200
|
|
assert client.delete(f"/api/v2/wiki/pages/{pid}/verify", cookies=c).status_code == 404
|
|
assert client.get(f"/api/v2/wiki/pages/{pid}/verification",
|
|
cookies=c).json()["verification"] is None
|
|
|
|
|
|
def test_verified_list_excludes_expired(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
fresh = _page(wid, "Fresh")
|
|
stale = _page(wid, "Stale")
|
|
client.post(f"/api/v2/wiki/pages/{fresh}/verify", cookies=c, json={"days": 30})
|
|
client.post(f"/api/v2/wiki/pages/{stale}/verify", cookies=c, json={"days": 30})
|
|
past = (datetime.datetime.now(datetime.UTC)
|
|
- datetime.timedelta(days=1)).replace(microsecond=0).isoformat()
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE page_verifications SET expires_at=? WHERE page_id=?", (past, stale))
|
|
conn.commit()
|
|
r = client.get(f"/api/v2/wiki/verified?workspace_id={wid}", cookies=c)
|
|
titles = [p["title"] for p in r.json()["pages"]]
|
|
assert "Fresh" in titles
|
|
assert "Stale" not in titles
|
|
|
|
|
|
def test_verify_days_clamped(client):
|
|
from app.services.wiki import verify_page
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
# 0 and 9999 are both clamped into 1..365
|
|
assert verify_page(pid, uid, days=9999)["expires_at"] is not None
|
|
assert verify_page(pid, uid, days=0)["expires_at"] is not None
|
|
with get_conn() as conn:
|
|
assert conn.execute("SELECT COUNT(*) AS n FROM page_verifications"
|
|
" WHERE page_id=?", (pid,)).fetchone()["n"] == 1
|
|
|
|
|
|
def test_verify_replaces_previous_verification(client):
|
|
from app.services.wiki import verification, verify_page
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
first = verify_page(pid, uid, days=10, note="first")
|
|
second = verify_page(pid, uid, days=20, note="second")
|
|
assert first["note"] == "first"
|
|
assert verification(pid)["note"] == "second"
|
|
assert second["expires_at"] != first["expires_at"]
|
|
|
|
|
|
def test_verified_list_hides_private_teamspace(client):
|
|
owner, _, oc = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
|
|
json={"workspace_id": wid, "name": "P",
|
|
"private": True}).json()["id"]
|
|
pid = _page(wid, "Hidden", teamspace_id=tsid)
|
|
client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=oc)
|
|
other, _, oc2 = _user()
|
|
_member(wid, other, "viewer")
|
|
r = client.get(f"/api/v2/wiki/verified?workspace_id={wid}", cookies=oc2)
|
|
assert "Hidden" not in [p["title"] for p in r.json()["pages"]]
|
|
|
|
|
|
def test_expiry_sweep_notifies(client):
|
|
admin, _, ac = _user(is_admin=1)
|
|
owner, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
_member(wid, owner, "admin")
|
|
pid = _page(wid, "About to expire")
|
|
client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=ac, json={"days": 1})
|
|
soon = (datetime.datetime.now(datetime.UTC)
|
|
+ datetime.timedelta(days=3)).replace(microsecond=0).isoformat()
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE page_verifications SET expires_at=? WHERE page_id=?", (soon, pid))
|
|
conn.commit()
|
|
r = client.post("/api/v2/wiki/verify-expiry-sweep", cookies=ac)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["notified"] == 1
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT ntype FROM notifications WHERE user_id=? AND"
|
|
" resource_id=?", (admin, pid)).fetchone()
|
|
assert row and row["ntype"] == "page.verification_expiring"
|
|
|
|
|
|
def test_expiry_sweep_requires_admin(client):
|
|
uid, _, c = _user()
|
|
assert client.post("/api/v2/wiki/verify-expiry-sweep", cookies=c).status_code == 403
|
|
|
|
|
|
# ── follows ────────────────────────────────────────────────────────────────
|
|
|
|
def test_toggle_follow(client):
|
|
from app.services.wiki import is_following
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid)
|
|
r = client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=c)
|
|
assert r.json()["following"] is True
|
|
assert is_following(pid, uid) is True
|
|
r = client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=c)
|
|
assert r.json()["following"] is False
|
|
assert is_following(pid, uid) is False
|
|
|
|
|
|
def test_list_followers(client):
|
|
a, _, ac = _user()
|
|
b, _, bc = _user()
|
|
wid = _workspace(a)
|
|
pid = _page(wid)
|
|
client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=ac)
|
|
client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=bc)
|
|
r = client.get(f"/api/v2/wiki/pages/{pid}/followers", cookies=ac)
|
|
assert {f["id"] for f in r.json()["followers"]} == {a, b}
|
|
|
|
|
|
def test_follow_unknown_page(client):
|
|
_, _, c = _user()
|
|
assert client.post("/api/v2/wiki/pages/999999/follow", cookies=c).status_code == 404
|
|
|
|
|
|
# ── comment reactions ──────────────────────────────────────────────────────
|
|
|
|
def _comment(page_id, uid, body="hi"):
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO comments (page_id, user_id, body) VALUES (?,?,?)",
|
|
(page_id, uid, body))
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
def test_comment_reaction_toggle(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid)
|
|
cid = _comment(pid, uid)
|
|
r = client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=c, json={"emoji": "👍"})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["reactions"]["👍"]["count"] == 1
|
|
# same emoji again removes it
|
|
r = client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=c, json={"emoji": "👍"})
|
|
assert r.json()["reactions"] == {}
|
|
|
|
|
|
def test_comment_reaction_multiple_users(client):
|
|
a, _, ac = _user()
|
|
b, _, bc = _user()
|
|
wid = _workspace(a)
|
|
pid = _page(wid)
|
|
cid = _comment(pid, a)
|
|
client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=ac, json={"emoji": "🎉"})
|
|
client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=bc, json={"emoji": "🎉"})
|
|
got = client.get(f"/api/v2/wiki/comments/{cid}/reactions", cookies=ac)
|
|
data = got.json()["reactions"]
|
|
assert data["🎉"]["count"] == 2
|
|
assert set(data["🎉"]["users"]) == {a, b}
|
|
|
|
|
|
def test_comment_reaction_requires_emoji(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid)
|
|
cid = _comment(pid, uid)
|
|
assert client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=c,
|
|
json={}).status_code == 400
|
|
|
|
|
|
def test_comment_reaction_unknown_comment(client):
|
|
uid, _, c = _user()
|
|
assert client.post("/api/v2/wiki/comments/999999/reactions", cookies=c,
|
|
json={"emoji": "👍"}).status_code == 404
|
|
|
|
|
|
# ── guest shares ───────────────────────────────────────────────────────────
|
|
|
|
def test_guest_share_flow(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid, "Shared doc", content="the body")
|
|
r = client.post(f"/api/v2/wiki/pages/{pid}/guests", cookies=c,
|
|
json={"email": "[email protected]", "role": "viewer"})
|
|
assert r.status_code == 201, r.text
|
|
body = r.json()
|
|
assert body["url"] == f"/g/{body['token']}"
|
|
# the guest page is reachable with no session at all
|
|
guest = client.get(f"/g/{body['token']}")
|
|
assert guest.status_code == 200
|
|
assert "Shared doc" in guest.text
|
|
assert "the body" in guest.text
|
|
# a view was recorded
|
|
with get_conn() as conn:
|
|
assert conn.execute("SELECT views FROM page_views WHERE page_id=?", (pid,)).fetchone()[
|
|
"views"] == 1
|
|
|
|
|
|
def test_guest_share_unknown_token(client):
|
|
assert client.get("/g/g_totally_bogus").status_code == 404
|
|
|
|
|
|
def test_guest_share_expired(client):
|
|
from app.services.wiki import create_guest_share, resolve_guest_share
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid)
|
|
share = create_guest_share(pid, "[email protected]", "viewer", uid, days=30)
|
|
assert resolve_guest_share(share["token"]) is not None
|
|
past = (datetime.datetime.now(datetime.UTC)
|
|
- datetime.timedelta(days=1)).replace(microsecond=0).isoformat()
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE guest_shares SET expires_at=? WHERE id=?",
|
|
(past, share["id"]))
|
|
conn.commit()
|
|
assert resolve_guest_share(share["token"]) is None
|
|
assert client.get(f"/g/{share['token']}").status_code == 404
|
|
|
|
|
|
def test_guest_share_revoked(client):
|
|
from app.services.wiki import resolve_guest_share
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
created = client.post(f"/api/v2/wiki/pages/{pid}/guests", cookies=c,
|
|
json={"email": "[email protected]"}).json()
|
|
share_id = created["id"]
|
|
assert resolve_guest_share(created["token"]) is not None
|
|
assert client.delete(f"/api/v2/wiki/guests/{share_id}", cookies=c).status_code == 200
|
|
# revocation is idempotent, and the token stops working
|
|
assert client.delete(f"/api/v2/wiki/guests/{share_id}", cookies=c).status_code == 200
|
|
assert resolve_guest_share(created["token"]) is None
|
|
assert client.get(f"/g/{created['token']}").status_code == 404
|
|
lst = client.get(f"/api/v2/wiki/pages/{pid}/guests", cookies=c)
|
|
assert lst.json()["guests"][0]["revoked"] == 1
|
|
# unknown id → 404
|
|
assert client.delete("/api/v2/wiki/guests/999999", cookies=c).status_code == 404
|
|
|
|
|
|
def test_guest_share_bad_role(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
assert client.post(f"/api/v2/wiki/pages/{pid}/guests", cookies=c,
|
|
json={"email": "[email protected]", "role": "admin"}).status_code == 400
|
|
|
|
|
|
def test_guest_share_requires_session(client):
|
|
anon(client)
|
|
assert client.post("/api/v2/wiki/pages/1/guests", json={}).status_code == 401
|
|
|
|
|
|
# ── page views ─────────────────────────────────────────────────────────────
|
|
|
|
def test_page_view_recording_and_stats(client):
|
|
from app.services.wiki import record_view, view_stats
|
|
uid, _, _ = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid)
|
|
today = datetime.datetime.now(datetime.UTC)
|
|
d0 = today.strftime("%Y-%m-%d")
|
|
d1 = (today - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
|
|
d2 = (today - datetime.timedelta(days=2)).strftime("%Y-%m-%d")
|
|
assert record_view(pid, day=d0) == 1
|
|
assert record_view(pid, day=d0) == 2
|
|
assert record_view(pid, day=d1) == 1
|
|
stats = view_stats(pid, days=5)
|
|
assert stats["total"] == 3
|
|
assert len(stats["series"]) == 5
|
|
day_map = {d["day"]: d["views"] for d in stats["series"]}
|
|
assert day_map[d0] == 2
|
|
assert day_map[d1] == 1
|
|
assert day_map[d2] == 0 # gap filled
|
|
# outside the window
|
|
assert view_stats(pid, days=1)["total"] == 2
|
|
|
|
|
|
def test_page_view_stats_outside_window(client):
|
|
from app.services.wiki import record_view, view_stats
|
|
uid, _, _ = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid)
|
|
record_view(pid, day="2020-01-01")
|
|
assert view_stats(pid, days=30)["total"] == 0
|
|
assert view_stats(pid, days=30)["total"] == 0
|
|
|
|
|
|
def test_page_views_endpoint(client):
|
|
from app.services.wiki import record_view
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
pid = _page(wid)
|
|
record_view(pid)
|
|
r = client.get(f"/api/v2/wiki/pages/{pid}/views?days=7", cookies=c)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["total"] == 1
|
|
|
|
|
|
# ── wiki home ──────────────────────────────────────────────────────────────
|
|
|
|
def test_wiki_home(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
_member(wid, uid, "admin")
|
|
client.post("/api/v2/wiki/teamspaces", cookies=c,
|
|
json={"workspace_id": wid, "name": "Eng"})
|
|
pid = _page(wid, "Runbook")
|
|
client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=c)
|
|
r = client.get(f"/api/v2/wiki/home?workspace_id={wid}", cookies=c)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["workspace_id"] == wid
|
|
assert [t["name"] for t in body["teamspaces"]] == ["Eng"]
|
|
assert body["verified"][0]["page_id"] == pid
|
|
assert any(p["title"] == "Runbook" for p in body["recents"])
|
|
|
|
|
|
def test_wiki_home_requires_workspace(client):
|
|
_, _, c = _user()
|
|
assert client.get("/api/v2/wiki/home", cookies=c).status_code == 400
|
|
assert client.get("/api/v2/wiki/home?workspace_id=abc", cookies=c).status_code == 400
|
|
assert client.get("/api/v2/wiki/home?workspace_id=999999", cookies=c).status_code == 404
|
|
|
|
|
|
# ── blocks: mermaid / equation / progress ──────────────────────────────────
|
|
|
|
def test_mermaid_fallback_without_mmdc(monkeypatch):
|
|
from app.services import wiki_blocks
|
|
monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: False)
|
|
out = wiki_blocks.render_mermaid("graph TD; A-->B;")
|
|
assert 'class="mermaid"' in out
|
|
assert "graph TD" in out
|
|
|
|
|
|
def test_mermaid_uses_svg_when_available(monkeypatch):
|
|
from app.services import wiki_blocks
|
|
monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: True)
|
|
monkeypatch.setattr(wiki_blocks, "mermaid_to_svg", lambda src, timeout=20: "<svg><g/></svg>")
|
|
out = wiki_blocks.render_mermaid("graph TD; A-->B;")
|
|
assert out.startswith("<figure")
|
|
assert "<svg>" in out
|
|
|
|
|
|
def test_mermaid_escapes_source(monkeypatch):
|
|
from app.services import wiki_blocks
|
|
monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: False)
|
|
out = wiki_blocks.render_mermaid("<script>alert(1)</script>")
|
|
assert "<script>" not in out
|
|
assert "<script>" in out
|
|
|
|
|
|
def test_equation_render():
|
|
from app.services.wiki_blocks import render_equation
|
|
out = render_equation("E = mc^2")
|
|
assert "fd-equation" in out
|
|
assert "mc^2" in out
|
|
assert "\\(" in out and "\\)" in out
|
|
|
|
|
|
def test_equation_sanitizes_delimiter_breakout():
|
|
"""`<`, `>` and `\\` must be stripped so the source cannot close the
|
|
KaTeX delimiter early or inject markup into the exported HTML."""
|
|
from app.services.wiki_blocks import render_equation, sanitize_equation
|
|
out = render_equation("</span><img src=x onerror=alert(1)>")
|
|
assert "<img" not in out
|
|
assert sanitize_equation("</span>") == "/span" # tags neutralised, no `<`/`>`
|
|
assert "<" not in sanitize_equation("a<b>c")
|
|
assert ">" not in sanitize_equation("a<b>c")
|
|
assert sanitize_equation("a\\)</span>b") == "a)/spanb" # backslash + tags stripped
|
|
|
|
|
|
def test_equation_empty_returns_blank():
|
|
from app.services.wiki_blocks import render_equation
|
|
assert render_equation(" ") == ""
|
|
|
|
|
|
def test_progress_block_checkbox(client):
|
|
from app.services.wiki_blocks import progress_value
|
|
cid = _collection(None, "Tasks")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO collection_properties (collection_id, name,"
|
|
" prop_type) VALUES (?,?,?)", (cid, "Done", "checkbox"))
|
|
conn.commit()
|
|
pid = cur.lastrowid
|
|
_add_rows(cid, [{str(pid): True}, {str(pid): True}, {str(pid): False}, {str(pid): False}])
|
|
stats = progress_value({"collection_id": cid, "done_property_id": pid})
|
|
assert stats["done"] == 2
|
|
assert stats["total"] == 4
|
|
assert stats["percent"] == 50.0
|
|
|
|
|
|
def test_progress_block_number_sum(client):
|
|
from app.services.wiki_blocks import progress_value
|
|
cid = _collection(None, "Budget")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO collection_properties (collection_id, name,"
|
|
" prop_type) VALUES (?,?,?)", (cid, "Amount", "number"))
|
|
conn.commit()
|
|
pid = cur.lastrowid
|
|
_add_rows(cid, [{str(pid): 50}, {str(pid): 50}])
|
|
stats = progress_value({"collection_id": cid, "property_id": pid, "func": "sum"})
|
|
assert stats["percent"] == 100.0
|
|
|
|
|
|
def test_progress_block_empty_collection(client):
|
|
from app.services.wiki_blocks import progress_value
|
|
cid = _collection(None, "Empty")
|
|
stats = progress_value({"collection_id": cid, "property_id": 1})
|
|
assert stats["percent"] is None
|
|
|
|
|
|
def test_progress_block_no_ref():
|
|
from app.services.wiki_blocks import progress_value, render_progress
|
|
assert progress_value(None)["percent"] is None
|
|
assert progress_value({})["percent"] is None
|
|
out = render_progress({"type": "progress"})
|
|
assert "fd-progress" in out
|
|
|
|
|
|
def test_progress_block_render(client):
|
|
from app.services.wiki_blocks import render_progress
|
|
cid = _collection(None, "Tasks2")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO collection_properties (collection_id, name,"
|
|
" prop_type) VALUES (?,?,?)", (cid, "Done", "checkbox"))
|
|
conn.commit()
|
|
pid = cur.lastrowid
|
|
_add_rows(cid, [{str(pid): True}, {str(pid): False}])
|
|
out = render_progress({"type": "progress", "label": "Sprint",
|
|
"rollup_ref": {"collection_id": cid, "done_property_id": pid}})
|
|
assert "50.0%" in out
|
|
assert "width:50" in out
|
|
assert "Sprint" in out
|
|
|
|
|
|
def test_blocks_preview_endpoint(client):
|
|
uid, _, c = _user()
|
|
r = client.post("/api/v2/wiki/blocks/preview", cookies=c, json={"blocks": [
|
|
{"type": "mermaid", "content": "graph TD; A-->B;"},
|
|
{"type": "equation_inline", "content": "a^2+b^2=c^2"},
|
|
{"type": "paragraph", "content": "ignored"},
|
|
]})
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
types = [b["type"] for b in body["rendered"]]
|
|
assert types == ["mermaid", "equation_inline"] # unknown types skipped
|
|
|
|
|
|
def test_blocks_preview_validation(client):
|
|
anon(client)
|
|
_, _, c = _user()
|
|
assert client.post("/api/v2/wiki/blocks/preview", cookies=c, json={}).status_code == 400
|
|
assert client.post("/api/v2/wiki/blocks/preview", cookies=c,
|
|
json={"blocks": [{"type": "mermaid"}] * 201}).status_code == 400
|
|
assert client.post("/api/v2/wiki/blocks/preview", json={}).status_code == 401
|
|
|
|
|
|
def test_export_includes_new_blocks(client, monkeypatch):
|
|
"""blocks_to_html must render the v7.3 block types (HTML/PDF export)."""
|
|
from app.services import export, wiki_blocks
|
|
monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: False)
|
|
html = export.blocks_to_html([
|
|
{"type": "mermaid", "content": "graph TD; A-->B;"},
|
|
{"type": "equation_inline", "content": "x^2"},
|
|
{"type": "progress", "label": "Done",
|
|
"rollup_ref": {"collection_id": 1}},
|
|
])
|
|
assert 'class="mermaid"' in html
|
|
assert "fd-equation" in html
|
|
assert "fd-progress" in html
|
|
|
|
|
|
# ── v7.3.0 follow-up: sidebar teamspaces ───────────────────────────────────
|
|
|
|
def test_teamspaces_list_cross_workspace(client):
|
|
"""Without workspace_id the sidebar list spans every teamspace the user
|
|
can see (workspace_name disambiguates the sections)."""
|
|
a, _, _ = _user()
|
|
b, _, _ = _user()
|
|
w1 = _workspace(a, "Alpha")
|
|
w2 = _workspace(b, "Beta")
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by) VALUES (?,?,?)",
|
|
(w1, "A1", a))
|
|
conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by) VALUES (?,?,?)",
|
|
(w2, "B1", b))
|
|
conn.commit()
|
|
uid, _, c = _user()
|
|
_member(w1, uid, "viewer")
|
|
_member(w2, uid, "viewer")
|
|
r = client.get("/api/v2/wiki/teamspaces", cookies=c)
|
|
assert r.status_code == 200, r.text
|
|
by_name = {s["name"]: s for s in r.json()["teamspaces"]}
|
|
assert set(by_name) == {"A1", "B1"}
|
|
assert by_name["A1"]["workspace_name"] == "Alpha"
|
|
assert by_name["B1"]["workspace_name"] == "Beta"
|
|
|
|
|
|
def test_teamspace_page_html(client):
|
|
uid, _, c = _user()
|
|
wid = _workspace(uid)
|
|
pid = _page(wid, "Doc")
|
|
with get_conn() as conn:
|
|
cur = conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by) VALUES (?,?,?)",
|
|
(wid, "Eng", uid))
|
|
conn.commit()
|
|
tid = cur.lastrowid
|
|
conn.execute("UPDATE pages SET teamspace_id=? WHERE id=?", (tid, pid))
|
|
conn.commit()
|
|
r = client.get(f"/wiki/teamspaces/{tid}", cookies=c)
|
|
assert r.status_code == 200, r.text
|
|
assert "Eng" in r.text
|
|
assert f"/pages/{pid}" in r.text
|
|
assert "Teamspace" in r.text
|
|
|
|
|
|
def test_teamspace_page_private_404(client):
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app as _app
|
|
|
|
owner, _, _ = _user()
|
|
wid = _workspace(owner)
|
|
with get_conn() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO teamspaces (workspace_id, name, created_by, private) VALUES (?,?,?,1)",
|
|
(wid, "Secret", owner))
|
|
conn.commit()
|
|
tid = cur.lastrowid
|
|
outsider, _, oc = _user()
|
|
# HTML pages that 404 redirect to /workspaces; the API answers a real 404.
|
|
c = TestClient(_app, follow_redirects=False)
|
|
r = c.get(f"/wiki/teamspaces/{tid}", cookies=oc)
|
|
assert r.status_code in (302, 403, 404)
|
|
r2 = client.get(f"/api/v2/wiki/teamspaces/{tid}", cookies=oc)
|
|
assert r2.status_code == 404
|
|
|
|
|
|
# ── v7.3.0 follow-up: page.updated notifications ───────────────────────────
|
|
|
|
def test_comment_auto_follows_author(client):
|
|
"""Commenting implies following (default ON), so the author receives
|
|
page.updated notifications from then on."""
|
|
from app.services.wiki import is_following
|
|
a, _, ac = _user()
|
|
wid = _workspace(a)
|
|
pid = _page(wid, "Page")
|
|
assert is_following(pid, a) is False
|
|
r = client.post(f"/api/pages/{pid}/comments", cookies=ac, json={"body": "lgtm"})
|
|
assert r.status_code == 200, r.text
|
|
assert is_following(pid, a) is True
|
|
|
|
|
|
def test_page_updated_notifies_followers_throttled(client):
|
|
"""Followers get an in-app page.updated notification, throttled to one per
|
|
10-minute window; the actor never notifies themselves."""
|
|
import asyncio
|
|
|
|
from app.services import automations
|
|
a, _, _ = _user()
|
|
b, _, _ = _user()
|
|
wid = _workspace(a)
|
|
pid = _page(wid, "Titre")
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO page_follows (page_id, user_id) VALUES (?,?)", (pid, b))
|
|
conn.commit()
|
|
payload = {"page_id": pid, "title": "Titre", "content_format": "blocks", "actor_id": a}
|
|
asyncio.run(automations.fire_event("page.updated", payload))
|
|
with get_conn() as conn:
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) AS c FROM notifications WHERE user_id=? AND"
|
|
" resource_type='page' AND resource_id=? AND ntype='page.updated'",
|
|
(b, pid)).fetchone()["c"]
|
|
assert n == 1
|
|
asyncio.run(automations.fire_event("page.updated", payload))
|
|
with get_conn() as conn:
|
|
n2 = conn.execute(
|
|
"SELECT COUNT(*) AS c FROM notifications WHERE user_id=? AND"
|
|
" resource_type='page' AND resource_id=? AND ntype='page.updated'",
|
|
(b, pid)).fetchone()["c"]
|
|
assert n2 == 1 # autosave spamming → still just one
|
|
with get_conn() as conn:
|
|
na = conn.execute(
|
|
"SELECT COUNT(*) AS c FROM notifications WHERE user_id=? AND"
|
|
" resource_type='page' AND resource_id=? AND ntype='page.updated'",
|
|
(a, pid)).fetchone()["c"]
|
|
assert na == 0 # actor excluded
|
|
|
|
|
|
def test_ensure_follow_idempotent(client):
|
|
from app.services.wiki import ensure_follow, is_following
|
|
a, _, _ = _user()
|
|
wid = _workspace(a)
|
|
pid = _page(wid)
|
|
assert ensure_follow(pid, a) is True
|
|
assert ensure_follow(pid, a) is False
|
|
assert is_following(pid, a) is True
|
|
|
|
|
|
# ── v7.3.0 follow-up: gitea:/github: unfurl ────────────────────────────────
|
|
|
|
def test_unfurl_repo_ref_regex():
|
|
from app.routers.board import _REPO_REF_RE
|
|
assert _REPO_REF_RE.match("gitea:flowdeck/flowdeck")
|
|
assert _REPO_REF_RE.match("github:torvalds/linux")
|
|
assert not _REPO_REF_RE.match("https://github.com/foo/bar")
|
|
assert not _REPO_REF_RE.match("gitea:owner")
|
|
assert not _REPO_REF_RE.match("evil:owner/repo")
|
|
|
|
|
|
def test_unfurl_repo_gitea(monkeypatch):
|
|
import asyncio
|
|
|
|
from app.routers.board import _unfurl_repo
|
|
from app.services import gitea_client
|
|
|
|
async def fake(self, owner, repo):
|
|
return {"full_name": "flowdeck/flowdeck", "description": "A great deck",
|
|
"default_branch": "main",
|
|
"html_url": "https://gitea.local/flowdeck/flowdeck",
|
|
"language": "Python"}
|
|
|
|
monkeypatch.setattr(gitea_client.GiteaClient, "get_repo_info", fake)
|
|
data = asyncio.run(_unfurl_repo("gitea", "flowdeck", "flowdeck"))
|
|
assert data["title"] == "flowdeck/flowdeck"
|
|
assert data["description"] == "A great deck"
|
|
assert data["site_name"] == "Gitea"
|
|
|
|
|
|
def test_unfurl_repo_github_unauthenticated(monkeypatch):
|
|
"""Without settings.github_token the forge API is called unauthenticated
|
|
(own http client), which is the production default."""
|
|
import asyncio
|
|
|
|
import httpx
|
|
|
|
from app.routers.board import _unfurl_repo
|
|
|
|
class _Resp:
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def json(self):
|
|
return {"full_name": "torvalds/linux", "default_branch": "master",
|
|
"html_url": "https://github.com/torvalds/linux", "language": "C"}
|
|
|
|
class _Client:
|
|
def __init__(self, *a, **k):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def get(self, url, headers=None):
|
|
return _Resp()
|
|
|
|
monkeypatch.setattr(httpx, "AsyncClient", _Client)
|
|
data = asyncio.run(_unfurl_repo("github", "torvalds", "linux"))
|
|
assert data["site_name"] == "GitHub"
|
|
assert data["language"] == "C"
|
|
|
|
|
|
def test_og_metadata_gitea_ref(client, monkeypatch):
|
|
import secrets
|
|
|
|
from app.services import gitea_client
|
|
|
|
async def fake(self, owner, repo):
|
|
return {"full_name": "flowdeck/flowdeck", "description": "A great deck",
|
|
"default_branch": "main"}
|
|
|
|
monkeypatch.setattr(gitea_client.GiteaClient, "get_repo_info", fake)
|
|
uid, _, c = _user()
|
|
csrf = secrets.token_hex(32)
|
|
r = client.post(
|
|
"/board/api/og/metadata",
|
|
cookies={**c, "csrf_token": csrf},
|
|
headers={"X-CSRF-Token": csrf},
|
|
json={"url": "gitea:flowdeck/flowdeck"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["ok"] is True
|
|
assert r.json()["title"] == "flowdeck/flowdeck"
|
|
|
|
|
|
# ── v7.3.0 follow-up: number KPI charts + multi-DB dashboards ──────────────
|
|
|
|
def test_chart_values_preserves_zero():
|
|
"""A 0 must stay 0 (it used to be coerced to 1 for the bar count)."""
|
|
from app.routers.collections import _chart_values
|
|
pages = [
|
|
{"property_values_json": '{"n": 0}'},
|
|
{"property_values_json": '{"n": 5}'},
|
|
{"property_values_json": '{"n": ""}'},
|
|
]
|
|
assert _chart_values(pages, "n") == [0.0, 5.0]
|
|
|
|
|
|
def test_chart_number_kpi():
|
|
import json
|
|
|
|
from app.routers.collections import _chart_aggregate, _fmt_number, _render_chart
|
|
pages = [
|
|
{"title": "a", "property_values_json": json.dumps({"prio": 3})},
|
|
{"title": "b", "property_values_json": json.dumps({"prio": 5})},
|
|
{"title": "c", "property_values_json": json.dumps({"prio": 0})},
|
|
]
|
|
assert _chart_aggregate(pages, "prio", "count") == 3.0
|
|
assert _chart_aggregate(pages, "prio", "sum") == 8.0
|
|
assert _chart_aggregate(pages, "prio", "avg") == 8.0 / 3.0
|
|
assert _chart_aggregate(pages, "prio", "min") == 0.0
|
|
assert _chart_aggregate(pages, "prio", "max") == 5.0
|
|
assert _fmt_number(350) == "350"
|
|
assert _fmt_number(1234) == "1.2K"
|
|
html = _render_chart("chart", {"name": "B", "icon": "X"}, pages,
|
|
{"chart_type": "number", "chart_property": "prio",
|
|
"aggregate": "sum"})
|
|
assert "kpi-value" in html
|
|
assert ">8<" in html
|
|
|
|
|
|
def test_dashboard_render_multi_db(client):
|
|
import json as _json
|
|
|
|
a, _, ac = _user()
|
|
w1 = _workspace(a, "W1")
|
|
w2 = _workspace(a, "W2")
|
|
c1 = _collection(w1, "Sales")
|
|
c2 = _collection(w2, "Issues")
|
|
_add_rows(c1, [{"amount": 100}, {"amount": 250}])
|
|
_add_rows(c2, [{"prio": 1}, {"prio": 3}])
|
|
with get_conn() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO collection_dashboards (collection_id, name, layout_json)"
|
|
" VALUES (?,?,?)",
|
|
(c1, "Board", _json.dumps({
|
|
"columns": 2,
|
|
"widgets": [
|
|
{"collection_id": c1, "width": 1, "chart_type": "number",
|
|
"chart_property": "amount", "aggregate": "sum",
|
|
"title": "Revenue"},
|
|
{"collection_id": c2, "width": 1, "chart_type": "bar",
|
|
"title": "Issues"},
|
|
],
|
|
})))
|
|
conn.commit()
|
|
did = cur.lastrowid
|
|
r = client.get(f"/db/{c1}/dashboards/{did}", cookies=ac)
|
|
assert r.status_code == 200, r.text
|
|
assert "Board" in r.text
|
|
assert "Revenue" in r.text
|
|
assert ">350<" in r.text # widget sums across its own database
|
|
assert "Issues" in r.text
|
|
|
|
|
|
def test_dashboard_render_404(client):
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app as _app
|
|
|
|
a, _, ac = _user()
|
|
wid = _workspace(a)
|
|
cid = _collection(wid, "X")
|
|
# Missing dashboard on an HTML path → 404 raised, app HTTP-404 handler
|
|
# redirects non-API paths to /workspaces.
|
|
c = TestClient(_app, follow_redirects=False)
|
|
r = c.get(f"/db/{cid}/dashboards/999999", cookies=ac)
|
|
assert r.status_code in (302, 404)
|