bruno
0112a5b5d8
- Correctif : la fenêtre Share rechargait ses partages à chaque ouverture (before: liste vide après refresh) - Sidebar « Shared » scindée en sous-groupes « Par moi » (partages nominatifs + liens) et « Avec moi » (partages reçus), sans doublons - Bibliothèque : filtre « Tous / Par moi / Avec moi » sur l'onglet Shared ; /api/library/shared?dir=made|received|all + share_dir par élément - Document : bouton barre affiche « 👥 Shared ▾ » quand la page est partagée (membre, lien ou publiée), recalculé à la volée - Workspace local : emoji 👥 juste avant le nom des fichiers partagés (is_shared exposé par /api/local-workspace/tree) - Tests +6 (tests/test_sharing.py) : direction made/received/all, fallback dir invalide, rendu pageIsShared ; suite 325 verte (+3 PDF pré-existants)
291 lines
9.6 KiB
Python
291 lines
9.6 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 json
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
db_path = db_file.name
|
|
db_file.close()
|
|
|
|
os.environ["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.main import app
|
|
from app.db import init_db, get_conn
|
|
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 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):
|
|
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_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 get_conn_ctx():
|
|
from app.db import get_conn
|
|
return get_conn() |