258 lines
11 KiB
Python
258 lines
11 KiB
Python
"""FlowDeck — Tests for v5.14.0 (Synced blocks)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import secrets
|
|
|
|
from app.db import get_conn
|
|
from app.services.synced_blocks import (
|
|
add_page_synced,
|
|
create_synced_block,
|
|
delete_synced_block,
|
|
get_page_synced,
|
|
get_synced_block,
|
|
list_synced_blocks,
|
|
remove_page_synced,
|
|
update_synced_block,
|
|
)
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────
|
|
|
|
def _login(client):
|
|
from app.auth.session import SessionManager
|
|
|
|
login = f"test_{secrets.token_hex(4)}"
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO users (login, full_name, email, is_admin) "
|
|
"VALUES (?, 'Test User', ?, 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})
|
|
r2 = client.get("/api/csrf-token", cookies={"flowdeck_session": session})
|
|
csrf = r2.json()["csrf_token"]
|
|
return session, csrf, uid
|
|
|
|
|
|
def _create_page(client, cookie, csrf, title="Test Page", content="[]"):
|
|
r = client.post(
|
|
f"/board/api/pages?title={title}",
|
|
headers={"X-CSRF-Token": csrf},
|
|
cookies={"flowdeck_session": cookie},
|
|
)
|
|
assert r.status_code == 200
|
|
return r.json()["id"]
|
|
|
|
|
|
# ── Service tests ────────────────────────────────────────────────
|
|
|
|
def test_create_synced_block(client):
|
|
"""Creating a synced block returns its id and stores it."""
|
|
session, _, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="My synced", content=[{"type": "paragraph", "content": "Hello"}], created_by=uid)
|
|
assert sid > 0
|
|
sb = get_synced_block(sid)
|
|
assert sb is not None
|
|
assert sb["title"] == "My synced"
|
|
assert json.loads(sb["content"]) == [{"type": "paragraph", "content": "Hello"}]
|
|
assert sb["workspace"] == "test"
|
|
|
|
|
|
def test_get_synced_block_not_found(client):
|
|
"""Getting a non-existent synced block returns None."""
|
|
sb = get_synced_block(99999)
|
|
assert sb is None
|
|
|
|
|
|
def test_update_synced_block(client):
|
|
"""Updating a synced block changes its content."""
|
|
session, _, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="Original", content=[{"type": "paragraph", "content": "Old"}], created_by=uid)
|
|
update_synced_block(sid, "Updated", [{"type": "heading_1", "content": "New"}])
|
|
sb = get_synced_block(sid)
|
|
assert sb["title"] == "Updated"
|
|
assert json.loads(sb["content"]) == [{"type": "heading_1", "content": "New"}]
|
|
|
|
|
|
def test_delete_synced_block(client):
|
|
"""Deleting a synced block removes it."""
|
|
session, _, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="To delete", content=[], created_by=uid)
|
|
delete_synced_block(sid)
|
|
assert get_synced_block(sid) is None
|
|
|
|
|
|
def test_list_synced_blocks(client):
|
|
"""Listing synced blocks returns all for a workspace."""
|
|
session, _, uid = _login(client)
|
|
create_synced_block(workspace="ws1", title="A", content=[], created_by=uid)
|
|
create_synced_block(workspace="ws1", title="B", content=[], created_by=uid)
|
|
create_synced_block(workspace="ws2", title="C", content=[], created_by=uid)
|
|
items = list_synced_blocks("ws1")
|
|
assert len(items) == 2
|
|
titles = {i["title"] for i in items}
|
|
assert titles == {"A", "B"}
|
|
|
|
|
|
# ── Page reference tests ─────────────────────────────────────────
|
|
|
|
def test_add_page_synced(client):
|
|
"""Adding a synced block reference to a page works."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="Sync1", content=[], created_by=uid)
|
|
page_id = _create_page(client, session, csrf, "Page with sync")
|
|
add_page_synced(page_id, sid, block_index=0)
|
|
refs = get_page_synced(page_id)
|
|
assert len(refs) == 1
|
|
assert refs[0]["synced_block_id"] == sid
|
|
|
|
|
|
def test_remove_page_synced(client):
|
|
"""Removing a synced block reference from a page works."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="Sync1", content=[], created_by=uid)
|
|
page_id = _create_page(client, session, csrf, "Page with sync")
|
|
add_page_synced(page_id, sid, block_index=0)
|
|
remove_page_synced(page_id, sid)
|
|
assert len(get_page_synced(page_id)) == 0
|
|
|
|
|
|
# ── API endpoint tests ───────────────────────────────────────────
|
|
|
|
def test_api_list_synced_blocks(client):
|
|
"""GET /board/api/synced-blocks returns synced blocks."""
|
|
session, _, uid = _login(client)
|
|
create_synced_block(workspace="test", title="API Test", content=[], created_by=uid)
|
|
r = client.get("/board/api/synced-blocks?workspace=test", cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert len(data["synced_blocks"]) >= 1
|
|
assert data["synced_blocks"][0]["title"] == "API Test"
|
|
|
|
|
|
def test_api_create_synced_block(client):
|
|
"""POST /board/api/synced-blocks creates a synced block."""
|
|
session, csrf, uid = _login(client)
|
|
r = client.post("/board/api/synced-blocks", json={
|
|
"title": "From API",
|
|
"content": [{"type": "paragraph", "content": "Hello from API"}],
|
|
"workspace": "test",
|
|
}, headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["status"] == "ok"
|
|
assert data["synced_block_id"] > 0
|
|
sb = get_synced_block(data["synced_block_id"])
|
|
assert sb["title"] == "From API"
|
|
|
|
|
|
def test_api_get_synced_block(client):
|
|
"""GET /board/api/synced-blocks/{sid} returns the block."""
|
|
session, _, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="GetTest", content=[], created_by=uid)
|
|
r = client.get(f"/board/api/synced-blocks/{sid}", cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
assert r.json()["title"] == "GetTest"
|
|
|
|
|
|
def test_api_get_synced_block_404(client):
|
|
"""GET /board/api/synced-blocks/{sid} returns 404 for missing block."""
|
|
session, _, uid = _login(client)
|
|
r = client.get("/board/api/synced-blocks/99999", cookies={"flowdeck_session": session})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_api_update_synced_block(client):
|
|
"""PUT /board/api/synced-blocks/{sid} updates the block."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="ToUpdate", content=[], created_by=uid)
|
|
r = client.put(f"/board/api/synced-blocks/{sid}", json={
|
|
"title": "Updated",
|
|
"content": [{"type": "heading_1", "content": "Updated content"}],
|
|
}, headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
sb = get_synced_block(sid)
|
|
assert sb["title"] == "Updated"
|
|
assert json.loads(sb["content"]) == [{"type": "heading_1", "content": "Updated content"}]
|
|
|
|
|
|
def test_api_delete_synced_block(client):
|
|
"""DELETE /board/api/synced-blocks/{sid} deletes the block."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="ToDelete", content=[], created_by=uid)
|
|
r = client.delete(f"/board/api/synced-blocks/{sid}", headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
assert get_synced_block(sid) is None
|
|
|
|
|
|
def test_api_add_synced_to_page(client):
|
|
"""POST /board/api/pages/{id}/synced adds a synced reference."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="PageSync", content=[], created_by=uid)
|
|
page_id = _create_page(client, session, csrf, "Page with sync")
|
|
r = client.post(f"/board/api/pages/{page_id}/synced", json={
|
|
"synced_block_id": sid,
|
|
"block_index": 0,
|
|
}, headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
refs = get_page_synced(page_id)
|
|
assert len(refs) == 1
|
|
|
|
|
|
def test_api_remove_synced_from_page(client):
|
|
"""DELETE /board/api/pages/{id}/synced/{sid} removes a synced reference."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="PageSync", content=[], created_by=uid)
|
|
page_id = _create_page(client, session, csrf, "Page with sync")
|
|
add_page_synced(page_id, sid, block_index=0)
|
|
r = client.delete(f"/board/api/pages/{page_id}/synced/{sid}", headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
assert len(get_page_synced(page_id)) == 0
|
|
|
|
|
|
def test_api_get_page_synced_refs(client):
|
|
"""GET /board/api/pages/{id}/synced returns synced refs for a page."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="RefTest", content=[], created_by=uid)
|
|
page_id = _create_page(client, session, csrf, "Page with sync")
|
|
add_page_synced(page_id, sid, block_index=0)
|
|
r = client.get(f"/board/api/pages/{page_id}/synced", cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert len(data["synced_blocks"]) == 1
|
|
assert data["synced_blocks"][0]["synced_block_id"] == sid
|
|
|
|
|
|
def test_save_blocks_with_synced(client):
|
|
"""POST /board/api/pages/{id}/blocks with synced blocks updates page_synced_blocks."""
|
|
session, csrf, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="BlockSync", content=[], created_by=uid)
|
|
page_id = _create_page(client, session, csrf, "Page with synced blocks")
|
|
r = client.post(f"/board/api/pages/{page_id}/blocks", json={
|
|
"blocks": [
|
|
{"type": "paragraph", "content": "Hello", "id": "b1"},
|
|
{"type": "synced", "synced_id": sid, "id": "b2"},
|
|
],
|
|
"title": "Updated",
|
|
}, headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": session})
|
|
assert r.status_code == 200
|
|
# Verify the reference exists
|
|
refs = get_page_synced(page_id)
|
|
assert any(r["synced_block_id"] == sid for r in refs)
|
|
|
|
|
|
# ── Resolve synced blocks tests ──────────────────────────────────
|
|
|
|
def test_resolve_synced_block(client):
|
|
"""resolve_synced_block replaces synced block references with content."""
|
|
from app.services.synced_blocks import resolve_synced_block
|
|
session, _, uid = _login(client)
|
|
sid = create_synced_block(workspace="test", title="ResolveTest", content=[{"type": "paragraph", "content": "Resolved!"}], created_by=uid)
|
|
blocks = [{"type": "synced", "synced_id": sid, "id": "b1"}]
|
|
resolved = resolve_synced_block(blocks)
|
|
assert len(resolved) == 1
|
|
assert resolved[0].get("_synced_id") == sid
|
|
assert resolved[0].get("_synced_content") == [{"type": "paragraph", "content": "Resolved!"}]
|