bruno
0861f1fdbf
- Scan de tous les appels `fetch` non-GET du front par préfixe : 12 préfixes n'ont AUCUN call site sans `X-CSRF-Token` → sortis de `EXCLUDED_PATHS` (`/db/`, `/workspace`, `/workspace/automations`, `/board/api/favorites`, `/api/pages`, `/api/recents`, `/api/notifications`, `/api/comments`, `/api/automations`, `/api/github`, `/api/admin`, `/api/onboarding`) - `welcome.html` : les 2 POST `/api/onboarding/*` reçoivent le header (`adminFetch` prouve que `/api/admin` était déjà couvert) - reste A19 (front à équiper) : `/api/workspace` (+`/api/workspaces`), `/api/local-workspace`, `/api/settings`, `/api/gitea`, `/api/agent` — 49 fetch - tests : helper `anon_csrf()` (anonyme + CSRF valide) pour isoler les 401 de route du 403 middleware — 4 tests d'anonymat ajustés - suite **1026/1026** · `ruff check app tests` OK
328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""FlowDeck — Partage de pages (v4.0 / fix partage membres).
|
|
|
|
Covers: partage par user_id ou email, validation de la permission
|
|
(view/comment/edit), upsert anti-doublon, mise à jour de permission (PUT),
|
|
retrait du partage et erreurs d'authentification.
|
|
"""
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from conftest import 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["DATABASE_URL"] = f"sqlite:///{db_path}"
|
|
os.environ["APP_SECRET_KEY"] = "test-secret-for-sharing"
|
|
os.environ["RATE_LIMIT_ENABLED"] = "false"
|
|
os.environ["GITEA_URL"] = "https://git.dracodev.net"
|
|
os.environ["GITEA_TOKEN"] = "test"
|
|
|
|
from app.config import settings
|
|
settings.database_url = f"sqlite:///{db_path}"
|
|
|
|
from app.db import get_conn, init_db
|
|
from app.main import app
|
|
init_db()
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO users (id, login, full_name, email, is_admin) "
|
|
"VALUES (1, 'owner', 'Owner', '[email protected]', 1)"
|
|
)
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO users (id, login, full_name, email, is_admin) "
|
|
"VALUES (2, 'alice', 'Alice', '[email protected]', 0)"
|
|
)
|
|
conn.commit()
|
|
|
|
tc = TestClient(app, raise_server_exceptions=False)
|
|
yield login_test_client(tc)
|
|
os.unlink(db_path)
|
|
|
|
|
|
def _token(user_id, login):
|
|
from app.auth.session import SessionManager
|
|
return SessionManager.create_session(
|
|
{"id": user_id, "login": login, "full_name": login.title(), "is_admin": 1}
|
|
)
|
|
|
|
|
|
def _auth(client, user_id=1, login="owner"):
|
|
client.cookies.set("flowdeck_session", _token(user_id, login))
|
|
|
|
|
|
def _make_page(_client, title="Share Page"):
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format, parent_section) "
|
|
"VALUES ('Private', ?, '[]', 'blocks', 'Private')",
|
|
(title,),
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
# ── Share création ──
|
|
|
|
|
|
def test_share_by_user_id(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
r = client.post(f"/api/pages/{pid}/share", json={"user_id": 2, "permission": "edit"})
|
|
assert r.status_code == 200, r.text
|
|
d = r.json()
|
|
assert d["status"] == "shared"
|
|
assert d["shared_with_user_id"] == 2
|
|
assert d["permission"] == "edit"
|
|
|
|
lst = client.get(f"/api/pages/{pid}/shares").json()["shares"]
|
|
assert len(lst) == 1
|
|
assert lst[0]["user_login"] == "alice"
|
|
assert lst[0]["permission"] == "edit"
|
|
|
|
|
|
def test_share_by_email_only(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
r = client.post(f"/api/pages/{pid}/share", json={"email": "[email protected]", "permission": "view"})
|
|
assert r.status_code == 200, r.text
|
|
d = r.json()
|
|
assert d["shared_with_email"] == "[email protected]"
|
|
assert d["shared_with_user_id"] is None
|
|
|
|
lst = client.get(f"/api/pages/{pid}/shares").json()["shares"]
|
|
assert lst[0]["shared_with_email"] == "[email protected]"
|
|
|
|
|
|
def test_share_invalid_permission_rejected(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
for bad in ("editor", "commenter", "viewer", "admin"):
|
|
r = client.post(f"/api/pages/{pid}/share", json={"email": "[email protected]", "permission": bad})
|
|
assert r.status_code == 400, bad
|
|
|
|
|
|
def test_share_requires_auth(client):
|
|
anon_csrf(client)
|
|
pid = _make_page(client)
|
|
r = client.post(f"/api/pages/{pid}/share", json={"email": "[email protected]", "permission": "view"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_share_without_target_rejected(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
r = client.post(f"/api/pages/{pid}/share", json={"permission": "view"})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_share_page_not_found(client):
|
|
_auth(client)
|
|
r = client.post("/api/pages/999999/share", json={"email": "[email protected]", "permission": "view"})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_share_target_user_missing(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
r = client.post(f"/api/pages/{pid}/share", json={"user_id": 999, "permission": "view"})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_share_upsert_same_user_updates_permission(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
first = client.post(f"/api/pages/{pid}/share", json={"user_id": 2, "permission": "view"}).json()
|
|
second = client.post(f"/api/pages/{pid}/share", json={"user_id": 2, "permission": "edit"}).json()
|
|
assert second["id"] == first["id"], "share should be upserted, not duplicated"
|
|
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT permission FROM page_shares WHERE page_id=? AND shared_with_user_id=2",
|
|
(pid,),
|
|
).fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0]["permission"] == "edit"
|
|
|
|
|
|
# ── Permission update (PUT) ──
|
|
|
|
|
|
def test_put_permission_updates(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
share_id = client.post(f"/api/pages/{pid}/share", json={"user_id": 2, "permission": "view"}).json()["id"]
|
|
r = client.put(f"/api/pages/{pid}/share/{share_id}", json={"permission": "comment"})
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "updated"
|
|
|
|
lst = client.get(f"/api/pages/{pid}/shares").json()["shares"]
|
|
assert lst[0]["permission"] == "comment"
|
|
|
|
|
|
def test_put_permission_invalid_rejected(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
share_id = client.post(f"/api/pages/{pid}/share", json={"user_id": 2, "permission": "view"}).json()["id"]
|
|
r = client.put(f"/api/pages/{pid}/share/{share_id}", json={"permission": "owner"})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_put_permission_missing_entry(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
r = client.put(f"/api/pages/{pid}/share/99999", json={"permission": "edit"})
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ── Remmove (DELETE) ──
|
|
|
|
|
|
def test_remove_share_unsets_is_shared(client):
|
|
_auth(client)
|
|
pid = _make_page(client)
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE pages SET is_shared=1 WHERE id=?", (pid,))
|
|
conn.commit()
|
|
|
|
share_id = client.post(f"/api/pages/{pid}/share", json={"user_id": 2, "permission": "view"}).json()["id"]
|
|
r = client.delete(f"/api/pages/{pid}/share/{share_id}")
|
|
assert r.status_code == 200
|
|
|
|
with get_conn() as conn:
|
|
is_shared = conn.execute("SELECT is_shared FROM pages WHERE id=?", (pid,)).fetchone()["is_shared"]
|
|
assert is_shared == 0
|
|
|
|
|
|
# ── Library : direction des partages (dir=made|received|all) ──
|
|
|
|
|
|
def _make_page_with(conn, title, share_mode="private", published=0):
|
|
cur = conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format, parent_section, share_mode, published) "
|
|
"VALUES ('Private', ?, '[]', 'blocks', 'Private', ?, ?)",
|
|
(title, share_mode, published),
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
def test_library_shared_dir_made_includes_nominal_and_link(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
with get_conn_ctx() as conn:
|
|
a = _make_page_with(conn, "Shared A")
|
|
b = _make_page_with(conn, "Link B", share_mode="anyone")
|
|
_make_page_with(conn, "Private C")
|
|
|
|
r = client.post(f"/api/pages/{a}/share", json={"user_id": 2, "permission": "edit"})
|
|
assert r.status_code == 200
|
|
|
|
made = client.get("/api/library/shared?dir=made").json()["items"]
|
|
ids = [it["id"] for it in made]
|
|
assert a in ids, "page shared to a user should appear in made"
|
|
assert b in ids, "link-mode page should appear in made"
|
|
assert all(it["share_dir"] == "made" for it in made if it["id"] in (a, b))
|
|
|
|
|
|
def test_library_shared_dir_received(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
with get_conn_ctx() as conn:
|
|
a = _make_page_with(conn, "Received A")
|
|
r = client.post(f"/api/pages/{a}/share", json={"user_id": 2, "permission": "view"})
|
|
assert r.status_code == 200
|
|
|
|
_auth(client, user_id=2, login="alice")
|
|
recv = client.get("/api/library/shared?dir=received").json()["items"]
|
|
ids = [it["id"] for it in recv]
|
|
assert a in ids
|
|
item = next(it for it in recv if it["id"] == a)
|
|
assert item["share_dir"] == "received"
|
|
|
|
|
|
def test_library_shared_dir_all_is_union(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
with get_conn_ctx() as conn:
|
|
a = _make_page_with(conn, "Union A")
|
|
client.post(f"/api/pages/{a}/share", json={"user_id": 2, "permission": "comment"})
|
|
|
|
_auth(client, user_id=2, login="alice")
|
|
all_items = client.get("/api/library/shared").json()["items"]
|
|
ids = [it["id"] for it in all_items]
|
|
assert a in ids
|
|
|
|
|
|
def test_library_shared_dir_all_includes_private_shared_made(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
with get_conn_ctx() as conn:
|
|
a = _make_page_with(conn, "Private Shared A")
|
|
client.post(f"/api/pages/{a}/share", json={"user_id": 2, "permission": "comment"})
|
|
|
|
all_items = client.get("/api/library/shared").json()["items"]
|
|
ids = [it["id"] for it in all_items]
|
|
assert a in ids, "privately-shared page must appear in 'all' (union made+received)"
|
|
|
|
|
|
def test_library_shared_invalid_dir_falls_back_to_all(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
with get_conn_ctx() as conn:
|
|
a = _make_page_with(conn, "Fallback A", share_mode="anyone")
|
|
r = client.get("/api/library/shared?dir=bogus")
|
|
assert r.status_code == 200
|
|
assert a in [it["id"] for it in r.json()["items"]]
|
|
|
|
|
|
def test_library_shared_received_none(client):
|
|
_auth(client, user_id=2, login="alice")
|
|
r = client.get("/api/library/shared?dir=received")
|
|
assert r.status_code == 200
|
|
assert r.json()["items"] == []
|
|
|
|
|
|
def test_page_editor_renders_page_is_shared(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
with get_conn_ctx() as conn:
|
|
a = _make_page_with(conn, "Marked A")
|
|
client.post(f"/api/pages/{a}/share", json={"user_id": 2, "permission": "view"})
|
|
|
|
r = client.get(f"/pages/{a}")
|
|
assert r.status_code == 200
|
|
assert "pageIsShared:true" in r.text
|
|
|
|
|
|
def test_tree_is_shared_includes_link_shared_pages(client):
|
|
_auth(client, user_id=1, login="owner")
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
ws = conn.execute("SELECT id FROM workspaces WHERE owner_id=1 LIMIT 1").fetchone()
|
|
if not ws:
|
|
return
|
|
ws_id = ws["id"]
|
|
cur = conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format, parent_section, share_mode, workspace_id) "
|
|
"VALUES ('WS', 'Link Shared WS', '[]', 'blocks', 'Workspace', 'anyone', ?)",
|
|
(ws_id,),
|
|
)
|
|
conn.commit()
|
|
pid = cur.lastrowid
|
|
|
|
r = client.get("/api/local-workspace/tree")
|
|
assert r.status_code == 200
|
|
items = r.json().get("children", [])
|
|
match = [c for c in items if c.get("id") == pid]
|
|
assert match, f"page {pid} not found in tree"
|
|
assert match[0]["is_shared"] is True, "link-shared page should show is_shared=true in tree"
|
|
|
|
|
|
def get_conn_ctx():
|
|
from app.db import get_conn
|
|
return get_conn()
|