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
273 lines
10 KiB
Python
273 lines
10 KiB
Python
"""FlowDeck — v5.13.0 Realtime: WebSocket gateway, présence, curseurs live,
|
|
merge LWW des opérations de blocs + version de page + persistance debounce.
|
|
|
|
Covers: la route WS /ws/pages/{page_id} (auth via cookie, page introuvable),
|
|
le protocole hello/sync/op/ack/sel/title/peer_*, le merge last-write-wins et
|
|
la resynchronisation des clients périmés.
|
|
"""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from conftest import anon, login_test_client
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
|
|
@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-realtime"
|
|
os.environ["RATE_LIMIT_ENABLED"] = "false"
|
|
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, is_admin) VALUES (1, 'tester', 'Tester', 1)")
|
|
conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (2, 'other', 'Other', 0)")
|
|
conn.commit()
|
|
|
|
from app.services.realtime_server import manager
|
|
manager._rooms = {}
|
|
|
|
tc = TestClient(app, raise_server_exceptions=False)
|
|
yield login_test_client(tc)
|
|
|
|
os.unlink(db_path)
|
|
|
|
|
|
def _make_page(title="Realtime Page", blocks=None):
|
|
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, json.dumps(blocks or [], ensure_ascii=False)),
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
|
|
|
|
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(client, user_id=1, login="tester"):
|
|
client.cookies.set("flowdeck_session", _token(user_id, login))
|
|
|
|
|
|
# ── apply_op (service pur) ──
|
|
|
|
def test_apply_op_insert():
|
|
from app.services.realtime_server import apply_op
|
|
blocks = [{"id": "a", "content": "A"}]
|
|
out = apply_op(blocks, {"type": "insert", "index": 0,
|
|
"block": {"id": "b", "content": "B"}})
|
|
assert [b["id"] for b in out] == ["b", "a"]
|
|
assert apply_op(blocks, {"type": "insert", "index": None,
|
|
"block": {"content": "C"}})[-1]["content"] == "C"
|
|
# index clampé
|
|
assert apply_op(blocks, {"type": "insert", "index": 99,
|
|
"block": {"id": "z"}})[-1]["id"] == "z"
|
|
|
|
|
|
def test_apply_op_update_delete_move():
|
|
from app.services.realtime_server import apply_op
|
|
blocks = [{"id": "a", "content": "A"}, {"id": "b", "content": "B"},
|
|
{"id": "c", "content": "C"}]
|
|
out = apply_op(blocks, {"type": "update", "block": {"id": "b", "content": "B2"}})
|
|
assert next(x for x in out if x["id"] == "b")["content"] == "B2"
|
|
out = apply_op(blocks, {"type": "update", "block": {"id": "nope", "content": "X"}})
|
|
assert out == blocks
|
|
out = apply_op(blocks, {"type": "delete", "id": "a"})
|
|
assert [b["id"] for b in out] == ["b", "c"]
|
|
out = apply_op(blocks, {"type": "move", "id": "c", "index": 0})
|
|
assert [b["id"] for b in out] == ["c", "a", "b"]
|
|
out = apply_op(blocks, {"type": "move", "id": "nope", "index": 0})
|
|
assert out == blocks
|
|
assert apply_op(blocks, {"type": "unknown"}) == blocks
|
|
|
|
|
|
def test_merge_ops_sequential():
|
|
from app.services.realtime_server import merge_ops
|
|
blocks = []
|
|
ops = [
|
|
{"type": "insert", "index": 0, "block": {"id": "a", "content": "A"}},
|
|
{"type": "insert", "index": 1, "block": {"id": "b", "content": "B"}},
|
|
{"type": "update", "block": {"id": "a", "content": "A2"}},
|
|
{"type": "move", "id": "b", "index": 0},
|
|
]
|
|
assert [x["id"] for x in merge_ops(blocks, ops)] == ["b", "a"]
|
|
assert next(x for x in merge_ops(blocks, ops) if x["id"] == "a")["content"] == "A2"
|
|
|
|
|
|
# ── Auth & présence de page ──
|
|
|
|
def test_ws_requires_auth(client):
|
|
anon(client)
|
|
_make_page()
|
|
with pytest.raises(WebSocketDisconnect) as exc:
|
|
with client.websocket_connect("/ws/pages/1") as ws:
|
|
ws.receive_json()
|
|
assert exc.value.code == 4401
|
|
|
|
|
|
def test_ws_auth_rejected(client):
|
|
anon(client)
|
|
_make_page()
|
|
with pytest.raises(WebSocketDisconnect) as exc:
|
|
with client.websocket_connect("/ws/pages/1") as ws:
|
|
ws.receive_json()
|
|
assert exc.value.code == 4401
|
|
|
|
|
|
def test_ws_missing_page_4404(client):
|
|
_auth_client(client)
|
|
with pytest.raises(WebSocketDisconnect) as exc:
|
|
with client.websocket_connect("/ws/pages/9999") as ws:
|
|
ws.receive_json()
|
|
assert exc.value.code == 4404
|
|
|
|
|
|
def test_ws_hello_gets_sync(client):
|
|
pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "Hi"}])
|
|
_auth_client(client)
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as ws:
|
|
w = ws.receive_json()
|
|
assert w["t"] == "welcome"
|
|
assert w["self"]["id"] == 1
|
|
s = ws.receive_json()
|
|
assert s["t"] == "sync"
|
|
assert s["version"] == 0
|
|
assert s["blocks"][0]["id"] == "a"
|
|
assert s["title"] == "Realtime Page"
|
|
ws.send_json({"t": "hello"})
|
|
s2 = ws.receive_json()
|
|
assert s2["t"] == "sync" and s2["blocks"][0]["content"] == "Hi"
|
|
|
|
|
|
# ── merge LWW + broadcast ──
|
|
|
|
def test_ws_lww_merge(client):
|
|
pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"},
|
|
{"id": "b", "type": "paragraph", "content": "y"}])
|
|
_auth_client(client, 1, "tester")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wa:
|
|
wa.receive_json() # welcome
|
|
wa.receive_json() # sync
|
|
_auth_client(client, 2, "other")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wb:
|
|
wb.receive_json() # welcome (peer tester)
|
|
wb.receive_json() # sync
|
|
wa.receive_json() # peer_join other
|
|
wa.send_json({"t": "op", "v": 0, "op": {"type": "update",
|
|
"block": {"id": "a", "type": "paragraph", "content": "first"}}})
|
|
ack = wa.receive_json() # ack de son propre op
|
|
assert ack["t"] == "ack" and ack["v"] == 1
|
|
b_op = wb.receive_json()
|
|
assert b_op["t"] == "op" and b_op["from"] == 1
|
|
assert b_op["op"]["block"]["content"] == "first"
|
|
wb.send_json({"t": "op", "v": 1, "op": {"type": "update",
|
|
"block": {"id": "a", "type": "paragraph", "content": "second"}}})
|
|
wb.receive_json() # ack de son propre op
|
|
a_op = wa.receive_json()
|
|
assert a_op["t"] == "op" and a_op["from"] == 2
|
|
assert a_op["op"]["block"]["content"] == "second"
|
|
# dernier arrivé gagne (LWW) et persistance au disconnect
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT content FROM pages WHERE id=?", (pid,)).fetchone()
|
|
d = json.loads(row["content"])
|
|
assert d[0]["content"] == "second"
|
|
|
|
|
|
# ── présence ──
|
|
|
|
def test_ws_presence_join_leave(client):
|
|
pid = _make_page()
|
|
_auth_client(client, 1, "tester")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wa:
|
|
wa.receive_json()
|
|
wa.receive_json()
|
|
_auth_client(client, 2, "other")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wb:
|
|
w_f = wb.receive_json() # welcome
|
|
assert w_f["t"] == "welcome"
|
|
assert any(p["id"] == 1 for p in w_f["peers"])
|
|
wb.receive_json() # sync
|
|
pj = wa.receive_json() # peer_join other
|
|
assert pj["t"] == "peer_join" and pj["peer"]["id"] == 2
|
|
pl = wa.receive_json() # peer_leave other
|
|
assert pl["t"] == "peer_leave" and pl["id"] == 2
|
|
|
|
|
|
def test_ws_cursor_broadcast(client):
|
|
pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "l"}])
|
|
_auth_client(client, 1, "tester")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wa:
|
|
wa.receive_json()
|
|
wa.receive_json()
|
|
_auth_client(client, 2, "other")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wb:
|
|
wb.receive_json()
|
|
wb.receive_json()
|
|
wa.receive_json() # peer_join
|
|
wa.send_json({"t": "sel", "block": "a", "offset": 1})
|
|
m = wb.receive_json()
|
|
assert m["t"] == "sel" and m["from"] == 1
|
|
assert m["block"] == "a" and m["offset"] == 1
|
|
wb.send_json({"t": "sel", "block": None, "offset": 0})
|
|
m2 = wa.receive_json()
|
|
assert m2["t"] == "sel" and m2["block"] is None
|
|
|
|
|
|
def test_ws_title_broadcast(client):
|
|
pid = _make_page()
|
|
_auth_client(client, 1, "tester")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wa:
|
|
wa.receive_json()
|
|
wa.receive_json()
|
|
_auth_client(client, 2, "other")
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as wb:
|
|
wb.receive_json()
|
|
wb.receive_json()
|
|
wa.receive_json()
|
|
wa.send_json({"t": "title", "title": "Nouveau titre"})
|
|
m = wb.receive_json()
|
|
assert m["t"] == "title" and m["title"] == "Nouveau titre"
|
|
|
|
|
|
# ── resynchronisation des clients périmés ──
|
|
|
|
def test_ws_stale_client_gets_sync(client):
|
|
pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}])
|
|
_auth_client(client)
|
|
with client.websocket_connect(f"/ws/pages/{pid}") as ws:
|
|
ws.receive_json() # welcome
|
|
ws.receive_json() # sync
|
|
for i in range(3):
|
|
ws.send_json({"t": "op", "v": 0, "op": {"type": "update",
|
|
"block": {"id": "a", "type": "paragraph", "content": f"v{i}"}}})
|
|
# séquence déterministe : ack(v1), ack(v2,stale)+sync(v2), ack(v3,stale)+sync(v3)
|
|
got_stale = False
|
|
syncs = []
|
|
for _ in range(5):
|
|
m = ws.receive_json()
|
|
if m["t"] == "ack" and m.get("stale"):
|
|
got_stale = True
|
|
if m["t"] == "sync":
|
|
syncs.append(m)
|
|
assert got_stale
|
|
assert syncs and syncs[-1]["version"] >= 3
|
|
assert syncs[-1]["blocks"][0]["content"] == "v2"
|