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