flowdeck/tests/test_sharing.py

"""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
    # A27 : la valeur n'est plus interpolée dans le JS inline mais exposée
    # dans le JSON #page-data (le JS lit PD.is_shared)
    import json as _json
    import re as _re

    m = _re.search(r'id="page-data"[^>]*>(.*?)</script>', r.text, _re.S)
    assert m, "bloc #page-data absent de la page"
    assert _json.loads(m.group(1))["is_shared"] is True


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()