flowdeck/tests/test_share_groups.py

"""FlowDeck — Partage de page par groupe (page_shares.shared_with_group_id).

Couvre : création d'un partage groupe, upsert anti-doublon,
listage (kind=group, group_name), mise à jour de permission,
suppression, et erreurs 404 (page / groupe inconnu).
"""
from app.auth.session import SessionManager
from app.db import get_conn


def _token(user_id, login="user", is_admin=1):
    return SessionManager.create_session(
        {"id": user_id, "login": login, "full_name": login.title(), "is_admin": is_admin}
    )


def _ensure_owner():
    with get_conn() as conn:
        row = conn.execute("SELECT id FROM users WHERE id=1").fetchone()
        if not row:
            conn.execute(
                "INSERT INTO users (id, login, full_name, email, is_admin) "
                "VALUES (1, 'owner', 'Owner', '[email protected]', 1)"
            )
            conn.commit()


def _as(client, user_id=1, login="owner"):
    _ensure_owner()
    client.cookies.set("flowdeck_session", _token(user_id, login))


def _insert_page(title="Group Share Page"):
    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


def _insert_group(name="Team"):
    _ensure_owner()
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO user_groups (workspace_id, name, description, created_by) "
            "VALUES (NULL, ?, '', 1)",
            (name,),
        )
        conn.commit()
        return cur.lastrowid


def test_share_page_with_group(client):
    _as(client)
    pid = _insert_page()
    gid = _insert_group()
    r = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "edit"})
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["status"] == "shared"
    assert d["shared_with_group_id"] == gid


def test_share_page_group_upsert_no_duplicate(client):
    _as(client)
    pid = _insert_page()
    gid = _insert_group()
    r1 = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "view"})
    r2 = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "comment"})
    assert r1.status_code == 200 and r2.status_code == 200
    assert r1.json()["id"] == r2.json()["id"]
    with get_conn() as conn:
        n = conn.execute(
            "SELECT COUNT(*) AS c FROM page_shares WHERE page_id=? AND shared_with_group_id=?",
            (pid, gid),
        ).fetchone()["c"]
    assert n == 1


def test_list_shares_includes_group(client):
    _as(client)
    pid = _insert_page()
    gid = _insert_group(name="Editors")
    client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "view"})
    r = client.get(f"/api/pages/{pid}/shares")
    assert r.status_code == 200, r.text
    shares = r.json()["shares"]
    g = [s for s in shares if s.get("shared_with_group_id") == gid]
    assert len(g) == 1
    assert g[0]["kind"] == "group"
    assert g[0]["group_name"] == "Editors"


def test_update_group_share_permission(client):
    _as(client)
    pid = _insert_page()
    gid = _insert_group()
    sid = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "view"}).json()["id"]
    r = client.put(f"/api/pages/{pid}/share/{sid}", json={"permission": "edit"})
    assert r.status_code == 200, r.text
    assert r.json()["permission"] == "edit"


def test_remove_group_share(client):
    _as(client)
    pid = _insert_page()
    gid = _insert_group()
    sid = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "view"}).json()["id"]
    r = client.delete(f"/api/pages/{pid}/share/{sid}")
    assert r.status_code == 200, r.text
    with get_conn() as conn:
        row = conn.execute("SELECT id FROM page_shares WHERE id=?", (sid,)).fetchone()
    assert row is None


def test_share_unknown_group_404(client):
    _as(client)
    pid = _insert_page()
    r = client.post(f"/api/pages/{pid}/share", json={"group_id": 999999, "permission": "view"})
    assert r.status_code == 404


def test_share_needs_target_400(client):
    _as(client)
    pid = _insert_page()
    r = client.post(f"/api/pages/{pid}/share", json={"permission": "view"})
    assert r.status_code == 400


def test_group_share_grants_effective_access(client):
    """A member of an invited group sees the page via page_permissions mirror."""
    from app.services.permission_manager import PermissionManager
    _as(client)
    pid = _insert_page()
    ws = _insert_workspace(1)
    _bind_page_workspace(pid, ws)
    gid = _insert_group(name="Readers")
    member = _insert_user("reader", "[email protected]", "Reader")
    with get_conn() as conn:
        conn.execute("INSERT INTO group_members (group_id, user_id) VALUES (?, ?)", (gid, member))
        conn.commit()
    r = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "edit"})
    assert r.status_code == 200, r.text
    assert PermissionManager(member).can_edit_page(pid)
    assert PermissionManager(member).can_view_page(pid)


def test_group_share_update_syncs_grant(client):
    from app.services.permission_manager import PermissionManager
    _as(client)
    pid = _insert_page()
    ws = _insert_workspace(1)
    _bind_page_workspace(pid, ws)
    gid = _insert_group(name="Writers")
    member = _insert_user("writer", "[email protected]", "Writer")
    with get_conn() as conn:
        conn.execute("INSERT INTO group_members (group_id, user_id) VALUES (?, ?)", (gid, member))
        conn.commit()
    sid = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "view"}).json()["id"]
    assert PermissionManager(member).can_view_page(pid)
    assert not PermissionManager(member).can_edit_page(pid)
    client.put(f"/api/pages/{pid}/share/{sid}", json={"permission": "edit"})
    assert PermissionManager(member).can_edit_page(pid)


def test_group_share_remove_revokes_grant(client):
    from app.services.permission_manager import PermissionManager
    _as(client)
    pid = _insert_page()
    ws = _insert_workspace(1)
    _bind_page_workspace(pid, ws)
    gid = _insert_group(name="Temp")
    member = _insert_user("temp", "[email protected]", "Temp")
    with get_conn() as conn:
        conn.execute("INSERT INTO group_members (group_id, user_id) VALUES (?, ?)", (gid, member))
        conn.commit()
    sid = client.post(f"/api/pages/{pid}/share", json={"group_id": gid, "permission": "edit"}).json()["id"]
    assert PermissionManager(member).can_edit_page(pid)
    client.delete(f"/api/pages/{pid}/share/{sid}")
    assert not PermissionManager(member).can_view_page(pid)


def _insert_workspace(owner_id, name="WS"):
    _ensure_owner()
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO workspaces (name, owner_id) VALUES (?, ?)", (name, owner_id))
        conn.commit()
        return cur.lastrowid


def _bind_page_workspace(page_id, workspace_id):
    with get_conn() as conn:
        conn.execute(
            "UPDATE pages SET workspace_id=?, permission_type='restricted' WHERE id=?",
            (workspace_id, page_id),
        )
        conn.commit()


def _insert_user(login, email, full_name):
    _ensure_owner()
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO users (login, email, full_name, is_active) VALUES (?,?,?,1)",
            (login, email, full_name),
        )
        conn.commit()
        return cur.lastrowid