feat(collab): v4.9.0 Collaboration - commentaires inline, mentions @, notifications in-app + email

tests/test_app.py

@@ -16,6 +16,7 @@ def client():
     os.environ["GITEA_TOKEN"] = "test"
     os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
     os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
+    os.environ["RATE_LIMIT_ENABLED"] = "false"
 
     from app.main import app
     from app.db import init_db
@@ -3234,4 +3235,184 @@ def test_v472_table_in_pdf(client, monkeypatch, tmp_path):
         assert pdf[:5] == b"%PDF-"
         assert len(pdf) > 1000
     finally:
-        _cleanup_src(pid, uid)
+        _cleanup_src(pid, uid)
+# ── v4.9.0: Collaboration — notifications, inline comments, mentions ──
+
+def _v490_users(conn, n=3, prefix="v490"):
+    ids = []
+    for i in range(n):
+        login = f"{prefix}{i}"
+        conn.execute("INSERT INTO users (login, full_name, email) VALUES (?,?,?)",
+                     (login, f"User {i}", f"{login}@t.com"))
+        ids.append(conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"])
+    conn.commit()
+    try:
+        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+    except Exception:
+        pass
+    return ids
+
+
+def _v490_page(conn, uid, title="Collab Page"):
+    cur = conn.execute("INSERT INTO pages (workspace, title, content, content_format) VALUES (?,?,?,?)",
+                       (f"u{uid}", title, "[]", "blocks"))
+    conn.commit()
+    try:
+        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+    except Exception:
+        pass
+    return cur.lastrowid
+
+
+def test_v490_notifications_table(client):
+    from app.db import get_conn
+    with get_conn() as conn:
+        t = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='notifications'").fetchone()
+        assert t is not None
+        cols = [r[1] for r in conn.execute("PRAGMA table_info(comments)").fetchall()]
+        assert "target_type" in cols and "anchor_block_id" in cols
+
+
+def test_v490_create_comment_with_mentions(client):
+    """Adding an inline comment with @mention creates a notification for the target."""
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    with get_conn() as conn:
+        uids = _v490_users(conn, 2)   # v4900, v4901
+        pid = _v490_page(conn, uids[0])
+    targ_id = uids[1]
+
+    session = SessionManager.create_session({"id": uids[0], "login": "v4900", "is_admin": 0})
+    cookies = {"flowdeck_session": session}
+
+    r = client.post(f"/api/pages/{pid}/comments",
+                    json={"body": "regarde ca @v4901", "anchor_block_id": "b1", "anchor_start": 0, "anchor_end": 5},
+                    cookies=cookies)
+    assert r.status_code == 200
+    cid = r.json()["id"]
+
+    r = client.get(f"/api/pages/{pid}/comments", cookies=cookies)
+    assert r.status_code == 200
+    comments = r.json()["comments"]
+    assert len(comments) == 1
+    assert comments[0]["anchor_block_id"] == "b1"
+
+    with get_conn() as conn:
+        n = conn.execute("SELECT * FROM notifications WHERE user_id=? AND ntype='mention'", (targ_id,)).fetchone()
+        assert n is not None
+        assert n["resource_id"] == pid
+
+
+def test_v490_comment_stores_anchor(client):
+    """A comment with no mentions is stored with its inline anchor."""
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    with get_conn() as conn:
+        uids = _v490_users(conn, 1, "v490a")
+        pid = _v490_page(conn, uids[0])
+    author_session = SessionManager.create_session({"id": uids[0], "login": "v490a0", "is_admin": 0})
+    r = client.post(f"/api/pages/{pid}/comments",
+                    json={"body": "hello", "anchor_block_id": "b7", "anchor_start": 1, "anchor_end": 3},
+                    cookies={"flowdeck_session": author_session})
+    assert r.status_code == 200
+    cid = r.json()["id"]
+    with get_conn() as conn:
+        row = conn.execute("SELECT * FROM comments WHERE id=?", (cid,)).fetchone()
+        assert row["anchor_block_id"] == "b7"
+        assert row["anchor_start"] == 1 and row["anchor_end"] == 3
+
+
+def test_v490_notifications_center(client):
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    from app.services import notifications as notif
+    with get_conn() as conn:
+        uids = _v490_users(conn, 1, "v490b")
+    notif.create_notification(uids[0], None, "mention", "T", "M", "page", 1, "/pages/1")
+
+    session = SessionManager.create_session({"id": uids[0], "login": "v490b0", "is_admin": 0})
+    cookies = {"flowdeck_session": session}
+
+    r = client.get("/api/notifications", cookies=cookies)
+    assert r.status_code == 200
+    data = r.json()
+    assert data["unread"] >= 1
+
+    r = client.get("/api/notifications/unread-count", cookies=cookies)
+    assert r.json()["unread"] >= 1
+
+    r = client.post("/api/notifications/read", json={}, cookies=cookies)
+    assert r.status_code == 200
+    r = client.get("/api/notifications/unread-count", cookies=cookies)
+    assert r.json()["unread"] == 0
+
+
+def test_v490_notification_prefs(client):
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    with get_conn() as conn:
+        uids = _v490_users(conn, 1, "v490c")
+    session = SessionManager.create_session({"id": uids[0], "login": "v490c0", "is_admin": 0})
+    cookies = {"flowdeck_session": session}
+
+    r = client.get("/api/notifications/prefs", cookies=cookies)
+    assert r.status_code == 200
+    assert r.json()["prefs"]["comments"] is True
+
+    r = client.post("/api/notifications/prefs", json={"comments": False}, cookies=cookies)
+    assert r.status_code == 200
+    assert r.json()["prefs"]["comments"] is False
+
+    r = client.get("/api/notifications/prefs", cookies=cookies)
+    assert r.json()["prefs"]["comments"] is False
+
+
+def test_v490_user_search(client):
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    with get_conn() as conn:
+        uids = _v490_users(conn, 2, "v490d")
+    session = SessionManager.create_session({"id": uids[0], "login": "v490d0", "is_admin": 0})
+    r = client.get("/api/notifications/users/search?q=v490d", cookies={"flowdeck_session": session})
+    assert r.status_code == 200
+    assert len(r.json()["users"]) >= 2
+
+
+def test_v490_resolve_and_delete_comment(client):
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    with get_conn() as conn:
+        uids = _v490_users(conn, 1, "v490e")
+        pid = _v490_page(conn, uids[0])
+    session = SessionManager.create_session({"id": uids[0], "login": "v490e0", "is_admin": 0})
+    cookies = {"flowdeck_session": session}
+    r = client.post(f"/api/pages/{pid}/comments", json={"body": "a comment"}, cookies=cookies)
+    cid = r.json()["id"]
+
+    r = client.put(f"/api/comments/{cid}", json={"resolved": True}, cookies=cookies)
+    assert r.status_code == 200
+    r = client.delete(f"/api/comments/{cid}", cookies=cookies)
+    assert r.status_code == 200
+    r = client.get(f"/api/pages/{pid}/comments", cookies=cookies)
+    assert r.json()["comments"] == []
+
+
+def test_v490_page_mentions_endpoint(client):
+    from app.db import get_conn
+    from app.auth.session import SessionManager
+    with get_conn() as conn:
+        uids = _v490_users(conn, 2, "v490f")
+        pid = _v490_page(conn, uids[0])
+    tid = uids[1]
+    session = SessionManager.create_session({"id": uids[0], "login": "v490f0", "is_admin": 0})
+    r = client.post(f"/api/pages/{pid}/mentions", json={"text": "hey @v490f1"}, cookies={"flowdeck_session": session})
+    assert r.status_code == 200
+    assert r.json()["mentioned"] == [tid]
+    with get_conn() as conn:
+        assert conn.execute("SELECT COUNT(*) c FROM notifications WHERE user_id=? AND ntype='mention'", (tid,)).fetchone()["c"] >= 1
+
+
+def test_v490_notifications_require_auth(client):
+    r = client.get("/api/notifications")
+    assert r.status_code == 401
+