"""FlowDeck — v7.3.0 Teamspaces + Verified pages + collab polish.

Covers migration 29, teamspace CRUD + private 404 + members/roles, verified
badge with expiry + notification sweep, wiki home, follows, comment
reactions, guest shares (expiry/revocation) and page views, plus the
mermaid/equation/progress block renderers.
"""
from __future__ import annotations

import datetime
import secrets

from conftest import anon

from app.db import get_conn

# ── helpers ────────────────────────────────────────────────────────────────

def _user(login=None, is_admin=0):
    login = login or f"v73_{secrets.token_hex(4)}"
    with get_conn() as conn:
        conn.execute("INSERT INTO users (login, full_name, email, is_admin) VALUES (?,?,?,?)",
                     (login, login, f"{login}@test.com", is_admin))
        uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
        conn.commit()
    from app.auth.session import SessionManager
    return uid, login, {"flowdeck_session": SessionManager.create_session(
        {"id": uid, "login": login})}


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


def _member(ws_id, uid, role="editor"):
    with get_conn() as conn:
        conn.execute("INSERT OR IGNORE INTO workspace_members (workspace_id, user_id, role)"
                     " VALUES (?,?,?)", (ws_id, uid, role))
        conn.commit()


def _page(ws_id, title="Doc", teamspace_id=None, owner_id=None, content="hello"):
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO pages (workspace, title, workspace_id, content,"
                           " content_format, teamspace_id) VALUES ('',?,?,?,'blocks',?)",
                           (title, ws_id, content, teamspace_id))
        conn.commit()
        return cur.lastrowid


def _collection(ws_id, name="Tasks"):
    """``ws_id=None`` creates a throwaway workspace (blocks tests need only the FK)."""
    if not ws_id:
        ws_id = _workspace(_user()[0], f"blkw{secrets.token_hex(3)}")
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO collections (name, workspace_id) VALUES (?,?)",
                           (name, ws_id))
        conn.commit()
        return cur.lastrowid


def _add_rows(cid, values_by_pid):
    import json
    with get_conn() as conn:
        for vals in values_by_pid:
            conn.execute("INSERT INTO collection_pages (collection_id, title, position,"
                         " property_values_json) VALUES (?,?,0,?)",
                         (cid, f"r{secrets.token_hex(2)}", json.dumps(vals)))
        conn.commit()


# ── migration 29 ───────────────────────────────────────────────────────────

def test_migration_29_tables_exist(client):
    with get_conn() as conn:
        names = {r["name"] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    for t in ("teamspaces", "teamspace_members", "page_verifications",
              "comment_reactions", "page_follows", "guest_shares", "page_views"):
        assert t in names, f"missing {t}"


def test_migration_29_teamspace_columns(client):
    with get_conn() as conn:
        for table in ("pages", "collections"):
            cols = {r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()}
            assert "teamspace_id" in cols, table


# ── teamspaces ─────────────────────────────────────────────────────────────

def test_teamspace_create_and_list(client):
    uid, login, c = _user()
    wid = _workspace(uid)
    r = client.post("/api/v2/wiki/teamspaces", cookies=c,
                    json={"workspace_id": wid, "name": "Engineering"})
    assert r.status_code == 201, r.text
    assert r.json()["name"] == "Engineering"
    lst = client.get(f"/api/v2/wiki/teamspaces?workspace_id={wid}", cookies=c)
    assert lst.status_code == 200
    spaces = lst.json()["teamspaces"]
    assert len(spaces) == 1
    assert spaces[0]["role"] == "owner"


def test_teamspace_creator_is_owner(client):
    from app.services.wiki import get_teamspace_role
    uid, _, _ = _user()
    wid = _workspace(uid)
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by)"
                           " VALUES (?,?,?)", (wid, "Docs", uid))
        conn.commit()
    assert get_teamspace_role(uid, cur.lastrowid) == "owner"


def test_teamspace_duplicate_name_conflict(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    assert client.post("/api/v2/wiki/teamspaces", cookies=c,
                       json={"workspace_id": wid, "name": "Dup"}).status_code == 201
    assert client.post("/api/v2/wiki/teamspaces", cookies=c,
                       json={"workspace_id": wid, "name": "Dup"}).status_code == 409


def test_teamspace_requires_name(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    assert client.post("/api/v2/wiki/teamspaces", cookies=c,
                       json={"workspace_id": wid, "name": "  "}).status_code == 400


def test_teamspace_requires_auth(client):
    anon(client)
    assert client.get("/api/v2/wiki/teamspaces?workspace_id=1").status_code == 401
    assert client.post("/api/v2/wiki/teamspaces", json={}).status_code == 401


def test_teamspace_private_is_404_for_outsider(client):
    owner, _, oc = _user()
    other, _, _ = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "Secret",
                             "private": True}).json()["id"]
    # owner sees it
    assert client.get(f"/api/v2/wiki/teamspaces/{tsid}", cookies=oc).status_code == 200
    # non-member of a private teamspace gets 404 (not 403)
    with get_conn() as conn:
        conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role)"
                     " VALUES (?,?,'viewer')", (wid, other))
        conn.commit()
    _, _, otherc = _user()
    r = client.get(f"/api/v2/wiki/teamspaces/{tsid}", cookies=otherc)
    assert r.status_code == 404


def test_teamspace_public_visible_to_workspace_members(client):
    owner, _, oc = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "Open"}).json()["id"]
    _member(wid, 999999 - 999999 + 0 or 0) if False else None
    other_uid, _, otherc = _user()
    _member(wid, other_uid, "viewer")
    assert client.get(f"/api/v2/wiki/teamspaces/{tsid}", cookies=otherc).status_code == 200
    lst = client.get(f"/api/v2/wiki/teamspaces?workspace_id={wid}", cookies=otherc)
    assert [t["name"] for t in lst.json()["teamspaces"]] == ["Open"]


def test_teamspace_list_empty_for_non_member(client):
    owner, _, _ = _user()
    wid = _workspace(owner)
    _, _, c = _user()
    assert client.get(f"/api/v2/wiki/teamspaces?workspace_id={wid}", cookies=c).json()[
        "teamspaces"] == []


def test_teamspace_members_add_update_remove(client):
    owner, _, oc = _user()
    member, _, _ = _user()
    wid = _workspace(owner)
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "Team"}).json()["id"]
    r = client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}", cookies=oc,
                   json={"role": "viewer"})
    assert r.status_code == 200, r.text
    assert r.json()["role"] == "viewer"
    lst = client.get(f"/api/v2/wiki/teamspaces/{tsid}/members", cookies=oc)
    assert {m["user_id"] for m in lst.json()["members"]} == {owner, member}
    # update
    assert client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}", cookies=oc,
                      json={"role": "editor"}).json()["role"] == "editor"
    # remove
    assert client.delete(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}",
                         cookies=oc).status_code == 200
    assert client.delete(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}",
                         cookies=oc).status_code == 404


def test_teamspace_member_bad_role(client):
    owner, _, oc = _user()
    member, _, _ = _user()
    wid = _workspace(owner)
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "T"}).json()["id"]
    assert client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/{member}", cookies=oc,
                      json={"role": "godmode"}).status_code == 400


def test_teamspace_member_unknown_user(client):
    owner, _, oc = _user()
    wid = _workspace(owner)
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "T"}).json()["id"]
    assert client.put(f"/api/v2/wiki/teamspaces/{tsid}/members/999999", cookies=oc,
                      json={"role": "editor"}).status_code == 404


def test_teamspace_role_helpers(client):
    from app.services import wiki
    owner, _, _ = _user()
    member, _, _ = _user()
    wid = _workspace(owner)
    _member(wid, member, "viewer")
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO teamspaces (workspace_id, name, private, created_by)"
                           " VALUES (?,?,0,?)", (wid, "Pub", owner))
        conn.commit()
        tsid = cur.lastrowid
    assert wiki.can_read_teamspace(member, tsid) is True
    assert wiki.can_write_teamspace(member, tsid) is False
    assert wiki.can_write_teamspace(owner, tsid) is True
    assert wiki.get_teamspace_role(999999, tsid) is None
    assert wiki.get_teamspace_role(None, tsid) is None


# ── verified pages ─────────────────────────────────────────────────────────

def test_verify_page_badge(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    r = client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=c, json={"note": "reviewed"})
    assert r.status_code == 200, r.text
    v = r.json()["verification"]
    assert v["page_id"] == pid
    assert v["active"] is True
    assert v["note"] == "reviewed"
    got = client.get(f"/api/v2/wiki/pages/{pid}/verification", cookies=c)
    assert got.json()["verification"]["verified_by"] == uid


def test_verify_page_requires_editor(client):
    owner, _, _ = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    pid = _page(wid)
    viewer, _, vc = _user()
    _member(wid, viewer, "viewer")
    assert client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=vc).status_code == 403


def test_verify_page_admin_override(client):
    admin, _, ac = _user(is_admin=1)
    owner, _, _ = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    pid = _page(wid)
    assert client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=ac).status_code == 200


def test_verify_page_in_private_teamspace(client):
    owner, _, oc = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "Priv",
                             "private": True}).json()["id"]
    pid = _page(wid, "Secret", teamspace_id=tsid)
    # owner is teamspace owner → can verify
    assert client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=oc).status_code == 200


def test_verify_page_unknown(client):
    uid, _, c = _user()
    assert client.post("/api/v2/wiki/pages/999999/verify", cookies=c).status_code == 404


def test_unverify_page(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=c)
    assert client.delete(f"/api/v2/wiki/pages/{pid}/verify", cookies=c).status_code == 200
    assert client.delete(f"/api/v2/wiki/pages/{pid}/verify", cookies=c).status_code == 404
    assert client.get(f"/api/v2/wiki/pages/{pid}/verification",
                      cookies=c).json()["verification"] is None


def test_verified_list_excludes_expired(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    fresh = _page(wid, "Fresh")
    stale = _page(wid, "Stale")
    client.post(f"/api/v2/wiki/pages/{fresh}/verify", cookies=c, json={"days": 30})
    client.post(f"/api/v2/wiki/pages/{stale}/verify", cookies=c, json={"days": 30})
    past = (datetime.datetime.now(datetime.UTC)
            - datetime.timedelta(days=1)).replace(microsecond=0).isoformat()
    with get_conn() as conn:
        conn.execute("UPDATE page_verifications SET expires_at=? WHERE page_id=?", (past, stale))
        conn.commit()
    r = client.get(f"/api/v2/wiki/verified?workspace_id={wid}", cookies=c)
    titles = [p["title"] for p in r.json()["pages"]]
    assert "Fresh" in titles
    assert "Stale" not in titles


def test_verify_days_clamped(client):
    from app.services.wiki import verify_page
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    # 0 and 9999 are both clamped into 1..365
    assert verify_page(pid, uid, days=9999)["expires_at"] is not None
    assert verify_page(pid, uid, days=0)["expires_at"] is not None
    with get_conn() as conn:
        assert conn.execute("SELECT COUNT(*) AS n FROM page_verifications"
                            " WHERE page_id=?", (pid,)).fetchone()["n"] == 1


def test_verify_replaces_previous_verification(client):
    from app.services.wiki import verification, verify_page
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    first = verify_page(pid, uid, days=10, note="first")
    second = verify_page(pid, uid, days=20, note="second")
    assert first["note"] == "first"
    assert verification(pid)["note"] == "second"
    assert second["expires_at"] != first["expires_at"]


def test_verified_list_hides_private_teamspace(client):
    owner, _, oc = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    tsid = client.post("/api/v2/wiki/teamspaces", cookies=oc,
                       json={"workspace_id": wid, "name": "P",
                             "private": True}).json()["id"]
    pid = _page(wid, "Hidden", teamspace_id=tsid)
    client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=oc)
    other, _, oc2 = _user()
    _member(wid, other, "viewer")
    r = client.get(f"/api/v2/wiki/verified?workspace_id={wid}", cookies=oc2)
    assert "Hidden" not in [p["title"] for p in r.json()["pages"]]


def test_expiry_sweep_notifies(client):
    admin, _, ac = _user(is_admin=1)
    owner, _, _ = _user()
    wid = _workspace(owner)
    _member(wid, owner, "admin")
    pid = _page(wid, "About to expire")
    client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=ac, json={"days": 1})
    soon = (datetime.datetime.now(datetime.UTC)
            + datetime.timedelta(days=3)).replace(microsecond=0).isoformat()
    with get_conn() as conn:
        conn.execute("UPDATE page_verifications SET expires_at=? WHERE page_id=?", (soon, pid))
        conn.commit()
    r = client.post("/api/v2/wiki/verify-expiry-sweep", cookies=ac)
    assert r.status_code == 200, r.text
    assert r.json()["notified"] == 1
    with get_conn() as conn:
        row = conn.execute("SELECT ntype FROM notifications WHERE user_id=? AND"
                           " resource_id=?", (admin, pid)).fetchone()
    assert row and row["ntype"] == "page.verification_expiring"


def test_expiry_sweep_requires_admin(client):
    uid, _, c = _user()
    assert client.post("/api/v2/wiki/verify-expiry-sweep", cookies=c).status_code == 403


# ── follows ────────────────────────────────────────────────────────────────

def test_toggle_follow(client):
    from app.services.wiki import is_following
    uid, _, c = _user()
    wid = _workspace(uid)
    pid = _page(wid)
    r = client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=c)
    assert r.json()["following"] is True
    assert is_following(pid, uid) is True
    r = client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=c)
    assert r.json()["following"] is False
    assert is_following(pid, uid) is False


def test_list_followers(client):
    a, _, ac = _user()
    b, _, bc = _user()
    wid = _workspace(a)
    pid = _page(wid)
    client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=ac)
    client.post(f"/api/v2/wiki/pages/{pid}/follow", cookies=bc)
    r = client.get(f"/api/v2/wiki/pages/{pid}/followers", cookies=ac)
    assert {f["id"] for f in r.json()["followers"]} == {a, b}


def test_follow_unknown_page(client):
    _, _, c = _user()
    assert client.post("/api/v2/wiki/pages/999999/follow", cookies=c).status_code == 404


# ── comment reactions ──────────────────────────────────────────────────────

def _comment(page_id, uid, body="hi"):
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO comments (page_id, user_id, body) VALUES (?,?,?)",
                           (page_id, uid, body))
        conn.commit()
        return cur.lastrowid


def test_comment_reaction_toggle(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    pid = _page(wid)
    cid = _comment(pid, uid)
    r = client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=c, json={"emoji": "👍"})
    assert r.status_code == 200, r.text
    assert r.json()["reactions"]["👍"]["count"] == 1
    # same emoji again removes it
    r = client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=c, json={"emoji": "👍"})
    assert r.json()["reactions"] == {}


def test_comment_reaction_multiple_users(client):
    a, _, ac = _user()
    b, _, bc = _user()
    wid = _workspace(a)
    pid = _page(wid)
    cid = _comment(pid, a)
    client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=ac, json={"emoji": "🎉"})
    client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=bc, json={"emoji": "🎉"})
    got = client.get(f"/api/v2/wiki/comments/{cid}/reactions", cookies=ac)
    data = got.json()["reactions"]
    assert data["🎉"]["count"] == 2
    assert set(data["🎉"]["users"]) == {a, b}


def test_comment_reaction_requires_emoji(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    pid = _page(wid)
    cid = _comment(pid, uid)
    assert client.post(f"/api/v2/wiki/comments/{cid}/reactions", cookies=c,
                       json={}).status_code == 400


def test_comment_reaction_unknown_comment(client):
    uid, _, c = _user()
    assert client.post("/api/v2/wiki/comments/999999/reactions", cookies=c,
                       json={"emoji": "👍"}).status_code == 404


# ── guest shares ───────────────────────────────────────────────────────────

def test_guest_share_flow(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid, "Shared doc", content="the body")
    r = client.post(f"/api/v2/wiki/pages/{pid}/guests", cookies=c,
                    json={"email": "ext@partner.example", "role": "viewer"})
    assert r.status_code == 201, r.text
    body = r.json()
    assert body["url"] == f"/g/{body['token']}"
    # the guest page is reachable with no session at all
    guest = client.get(f"/g/{body['token']}")
    assert guest.status_code == 200
    assert "Shared doc" in guest.text
    assert "the body" in guest.text
    # a view was recorded
    with get_conn() as conn:
        assert conn.execute("SELECT views FROM page_views WHERE page_id=?", (pid,)).fetchone()[
            "views"] == 1


def test_guest_share_unknown_token(client):
    assert client.get("/g/g_totally_bogus").status_code == 404


def test_guest_share_expired(client):
    from app.services.wiki import create_guest_share, resolve_guest_share
    uid, _, c = _user()
    wid = _workspace(uid)
    pid = _page(wid)
    share = create_guest_share(pid, "a@b.example", "viewer", uid, days=30)
    assert resolve_guest_share(share["token"]) is not None
    past = (datetime.datetime.now(datetime.UTC)
            - datetime.timedelta(days=1)).replace(microsecond=0).isoformat()
    with get_conn() as conn:
        conn.execute("UPDATE guest_shares SET expires_at=? WHERE id=?",
                     (past, share["id"]))
        conn.commit()
    assert resolve_guest_share(share["token"]) is None
    assert client.get(f"/g/{share['token']}").status_code == 404


def test_guest_share_revoked(client):
    from app.services.wiki import resolve_guest_share
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    created = client.post(f"/api/v2/wiki/pages/{pid}/guests", cookies=c,
                          json={"email": "x@y.example"}).json()
    share_id = created["id"]
    assert resolve_guest_share(created["token"]) is not None
    assert client.delete(f"/api/v2/wiki/guests/{share_id}", cookies=c).status_code == 200
    # revocation is idempotent, and the token stops working
    assert client.delete(f"/api/v2/wiki/guests/{share_id}", cookies=c).status_code == 200
    assert resolve_guest_share(created["token"]) is None
    assert client.get(f"/g/{created['token']}").status_code == 404
    lst = client.get(f"/api/v2/wiki/pages/{pid}/guests", cookies=c)
    assert lst.json()["guests"][0]["revoked"] == 1
    # unknown id → 404
    assert client.delete("/api/v2/wiki/guests/999999", cookies=c).status_code == 404


def test_guest_share_bad_role(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    assert client.post(f"/api/v2/wiki/pages/{pid}/guests", cookies=c,
                       json={"email": "x@y.example", "role": "admin"}).status_code == 400


def test_guest_share_requires_session(client):
    anon(client)
    assert client.post("/api/v2/wiki/pages/1/guests", json={}).status_code == 401


# ── page views ─────────────────────────────────────────────────────────────

def test_page_view_recording_and_stats(client):
    from app.services.wiki import record_view, view_stats
    uid, _, _ = _user()
    wid = _workspace(uid)
    pid = _page(wid)
    today = datetime.datetime.now(datetime.UTC)
    d0 = today.strftime("%Y-%m-%d")
    d1 = (today - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
    d2 = (today - datetime.timedelta(days=2)).strftime("%Y-%m-%d")
    assert record_view(pid, day=d0) == 1
    assert record_view(pid, day=d0) == 2
    assert record_view(pid, day=d1) == 1
    stats = view_stats(pid, days=5)
    assert stats["total"] == 3
    assert len(stats["series"]) == 5
    day_map = {d["day"]: d["views"] for d in stats["series"]}
    assert day_map[d0] == 2
    assert day_map[d1] == 1
    assert day_map[d2] == 0  # gap filled
    # outside the window
    assert view_stats(pid, days=1)["total"] == 2


def test_page_view_stats_outside_window(client):
    from app.services.wiki import record_view, view_stats
    uid, _, _ = _user()
    wid = _workspace(uid)
    pid = _page(wid)
    record_view(pid, day="2020-01-01")
    assert view_stats(pid, days=30)["total"] == 0
    assert view_stats(pid, days=30)["total"] == 0


def test_page_views_endpoint(client):
    from app.services.wiki import record_view
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    pid = _page(wid)
    record_view(pid)
    r = client.get(f"/api/v2/wiki/pages/{pid}/views?days=7", cookies=c)
    assert r.status_code == 200, r.text
    assert r.json()["total"] == 1


# ── wiki home ──────────────────────────────────────────────────────────────

def test_wiki_home(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    _member(wid, uid, "admin")
    client.post("/api/v2/wiki/teamspaces", cookies=c,
                json={"workspace_id": wid, "name": "Eng"})
    pid = _page(wid, "Runbook")
    client.post(f"/api/v2/wiki/pages/{pid}/verify", cookies=c)
    r = client.get(f"/api/v2/wiki/home?workspace_id={wid}", cookies=c)
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["workspace_id"] == wid
    assert [t["name"] for t in body["teamspaces"]] == ["Eng"]
    assert body["verified"][0]["page_id"] == pid
    assert any(p["title"] == "Runbook" for p in body["recents"])


def test_wiki_home_requires_workspace(client):
    _, _, c = _user()
    assert client.get("/api/v2/wiki/home", cookies=c).status_code == 400
    assert client.get("/api/v2/wiki/home?workspace_id=abc", cookies=c).status_code == 400
    assert client.get("/api/v2/wiki/home?workspace_id=999999", cookies=c).status_code == 404


# ── blocks: mermaid / equation / progress ──────────────────────────────────

def test_mermaid_fallback_without_mmdc(monkeypatch):
    from app.services import wiki_blocks
    monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: False)
    out = wiki_blocks.render_mermaid("graph TD; A-->B;")
    assert 'class="mermaid"' in out
    assert "graph TD" in out


def test_mermaid_uses_svg_when_available(monkeypatch):
    from app.services import wiki_blocks
    monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: True)
    monkeypatch.setattr(wiki_blocks, "mermaid_to_svg", lambda src, timeout=20: "<svg><g/></svg>")
    out = wiki_blocks.render_mermaid("graph TD; A-->B;")
    assert out.startswith("<figure")
    assert "<svg>" in out


def test_mermaid_escapes_source(monkeypatch):
    from app.services import wiki_blocks
    monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: False)
    out = wiki_blocks.render_mermaid("<script>alert(1)</script>")
    assert "<script>" not in out
    assert "&lt;script&gt;" in out


def test_equation_render():
    from app.services.wiki_blocks import render_equation
    out = render_equation("E = mc^2")
    assert "fd-equation" in out
    assert "mc^2" in out
    assert "\\(" in out and "\\)" in out


def test_equation_sanitizes_delimiter_breakout():
    """`<`, `>` and `\\` must be stripped so the source cannot close the
    KaTeX delimiter early or inject markup into the exported HTML."""
    from app.services.wiki_blocks import render_equation, sanitize_equation
    out = render_equation("</span><img src=x onerror=alert(1)>")
    assert "<img" not in out
    assert sanitize_equation("</span>") == "/span"  # tags neutralised, no `<`/`>`
    assert "<" not in sanitize_equation("a<b>c")
    assert ">" not in sanitize_equation("a<b>c")
    assert sanitize_equation("a\\)</span>b") == "a)/spanb"  # backslash + tags stripped


def test_equation_empty_returns_blank():
    from app.services.wiki_blocks import render_equation
    assert render_equation("   ") == ""


def test_progress_block_checkbox(client):
    from app.services.wiki_blocks import progress_value
    cid = _collection(None, "Tasks")
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO collection_properties (collection_id, name,"
                           " prop_type) VALUES (?,?,?)", (cid, "Done", "checkbox"))
        conn.commit()
        pid = cur.lastrowid
    _add_rows(cid, [{str(pid): True}, {str(pid): True}, {str(pid): False}, {str(pid): False}])
    stats = progress_value({"collection_id": cid, "done_property_id": pid})
    assert stats["done"] == 2
    assert stats["total"] == 4
    assert stats["percent"] == 50.0


def test_progress_block_number_sum(client):
    from app.services.wiki_blocks import progress_value
    cid = _collection(None, "Budget")
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO collection_properties (collection_id, name,"
                           " prop_type) VALUES (?,?,?)", (cid, "Amount", "number"))
        conn.commit()
        pid = cur.lastrowid
    _add_rows(cid, [{str(pid): 50}, {str(pid): 50}])
    stats = progress_value({"collection_id": cid, "property_id": pid, "func": "sum"})
    assert stats["percent"] == 100.0


def test_progress_block_empty_collection(client):
    from app.services.wiki_blocks import progress_value
    cid = _collection(None, "Empty")
    stats = progress_value({"collection_id": cid, "property_id": 1})
    assert stats["percent"] is None


def test_progress_block_no_ref():
    from app.services.wiki_blocks import progress_value, render_progress
    assert progress_value(None)["percent"] is None
    assert progress_value({})["percent"] is None
    out = render_progress({"type": "progress"})
    assert "fd-progress" in out


def test_progress_block_render(client):
    from app.services.wiki_blocks import render_progress
    cid = _collection(None, "Tasks2")
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO collection_properties (collection_id, name,"
                           " prop_type) VALUES (?,?,?)", (cid, "Done", "checkbox"))
        conn.commit()
        pid = cur.lastrowid
    _add_rows(cid, [{str(pid): True}, {str(pid): False}])
    out = render_progress({"type": "progress", "label": "Sprint",
                           "rollup_ref": {"collection_id": cid, "done_property_id": pid}})
    assert "50.0%" in out
    assert "width:50" in out
    assert "Sprint" in out


def test_blocks_preview_endpoint(client):
    uid, _, c = _user()
    r = client.post("/api/v2/wiki/blocks/preview", cookies=c, json={"blocks": [
        {"type": "mermaid", "content": "graph TD; A-->B;"},
        {"type": "equation_inline", "content": "a^2+b^2=c^2"},
        {"type": "paragraph", "content": "ignored"},
    ]})
    assert r.status_code == 200, r.text
    body = r.json()
    types = [b["type"] for b in body["rendered"]]
    assert types == ["mermaid", "equation_inline"]  # unknown types skipped


def test_blocks_preview_validation(client):
    anon(client)
    _, _, c = _user()
    assert client.post("/api/v2/wiki/blocks/preview", cookies=c, json={}).status_code == 400
    assert client.post("/api/v2/wiki/blocks/preview", cookies=c,
                       json={"blocks": [{"type": "mermaid"}] * 201}).status_code == 400
    assert client.post("/api/v2/wiki/blocks/preview", json={}).status_code == 401


def test_export_includes_new_blocks(client, monkeypatch):
    """blocks_to_html must render the v7.3 block types (HTML/PDF export)."""
    from app.services import export, wiki_blocks
    monkeypatch.setattr(wiki_blocks, "mmdc_available", lambda: False)
    html = export.blocks_to_html([
        {"type": "mermaid", "content": "graph TD; A-->B;"},
        {"type": "equation_inline", "content": "x^2"},
        {"type": "progress", "label": "Done",
         "rollup_ref": {"collection_id": 1}},
    ])
    assert 'class="mermaid"' in html
    assert "fd-equation" in html
    assert "fd-progress" in html


# ── v7.3.0 follow-up: sidebar teamspaces ───────────────────────────────────

def test_teamspaces_list_cross_workspace(client):
    """Without workspace_id the sidebar list spans every teamspace the user
    can see (workspace_name disambiguates the sections)."""
    a, _, _ = _user()
    b, _, _ = _user()
    w1 = _workspace(a, "Alpha")
    w2 = _workspace(b, "Beta")
    with get_conn() as conn:
        conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by) VALUES (?,?,?)",
                     (w1, "A1", a))
        conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by) VALUES (?,?,?)",
                     (w2, "B1", b))
        conn.commit()
    uid, _, c = _user()
    _member(w1, uid, "viewer")
    _member(w2, uid, "viewer")
    r = client.get("/api/v2/wiki/teamspaces", cookies=c)
    assert r.status_code == 200, r.text
    by_name = {s["name"]: s for s in r.json()["teamspaces"]}
    assert set(by_name) == {"A1", "B1"}
    assert by_name["A1"]["workspace_name"] == "Alpha"
    assert by_name["B1"]["workspace_name"] == "Beta"


def test_teamspace_page_html(client):
    uid, _, c = _user()
    wid = _workspace(uid)
    pid = _page(wid, "Doc")
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO teamspaces (workspace_id, name, created_by) VALUES (?,?,?)",
                           (wid, "Eng", uid))
        conn.commit()
        tid = cur.lastrowid
        conn.execute("UPDATE pages SET teamspace_id=? WHERE id=?", (tid, pid))
        conn.commit()
    r = client.get(f"/wiki/teamspaces/{tid}", cookies=c)
    assert r.status_code == 200, r.text
    assert "Eng" in r.text
    assert f"/pages/{pid}" in r.text
    assert "Teamspace" in r.text


def test_teamspace_page_private_404(client):
    from fastapi.testclient import TestClient

    from app.main import app as _app

    owner, _, _ = _user()
    wid = _workspace(owner)
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO teamspaces (workspace_id, name, created_by, private) VALUES (?,?,?,1)",
            (wid, "Secret", owner))
        conn.commit()
        tid = cur.lastrowid
    outsider, _, oc = _user()
    # HTML pages that 404 redirect to /workspaces; the API answers a real 404.
    c = TestClient(_app, follow_redirects=False)
    r = c.get(f"/wiki/teamspaces/{tid}", cookies=oc)
    assert r.status_code in (302, 403, 404)
    r2 = client.get(f"/api/v2/wiki/teamspaces/{tid}", cookies=oc)
    assert r2.status_code == 404


# ── v7.3.0 follow-up: page.updated notifications ───────────────────────────

def test_comment_auto_follows_author(client):
    """Commenting implies following (default ON), so the author receives
    page.updated notifications from then on."""
    from app.services.wiki import is_following
    a, _, ac = _user()
    wid = _workspace(a)
    pid = _page(wid, "Page")
    assert is_following(pid, a) is False
    r = client.post(f"/api/pages/{pid}/comments", cookies=ac, json={"body": "lgtm"})
    assert r.status_code == 200, r.text
    assert is_following(pid, a) is True


def test_page_updated_notifies_followers_throttled(client):
    """Followers get an in-app page.updated notification, throttled to one per
    10-minute window; the actor never notifies themselves."""
    import asyncio

    from app.services import automations
    a, _, _ = _user()
    b, _, _ = _user()
    wid = _workspace(a)
    pid = _page(wid, "Titre")
    with get_conn() as conn:
        conn.execute("INSERT INTO page_follows (page_id, user_id) VALUES (?,?)", (pid, b))
        conn.commit()
    payload = {"page_id": pid, "title": "Titre", "content_format": "blocks", "actor_id": a}
    asyncio.run(automations.fire_event("page.updated", payload))
    with get_conn() as conn:
        n = conn.execute(
            "SELECT COUNT(*) AS c FROM notifications WHERE user_id=? AND"
            " resource_type='page' AND resource_id=? AND ntype='page.updated'",
            (b, pid)).fetchone()["c"]
    assert n == 1
    asyncio.run(automations.fire_event("page.updated", payload))
    with get_conn() as conn:
        n2 = conn.execute(
            "SELECT COUNT(*) AS c FROM notifications WHERE user_id=? AND"
            " resource_type='page' AND resource_id=? AND ntype='page.updated'",
            (b, pid)).fetchone()["c"]
    assert n2 == 1  # autosave spamming → still just one
    with get_conn() as conn:
        na = conn.execute(
            "SELECT COUNT(*) AS c FROM notifications WHERE user_id=? AND"
            " resource_type='page' AND resource_id=? AND ntype='page.updated'",
            (a, pid)).fetchone()["c"]
    assert na == 0  # actor excluded


def test_ensure_follow_idempotent(client):
    from app.services.wiki import ensure_follow, is_following
    a, _, _ = _user()
    wid = _workspace(a)
    pid = _page(wid)
    assert ensure_follow(pid, a) is True
    assert ensure_follow(pid, a) is False
    assert is_following(pid, a) is True


# ── v7.3.0 follow-up: gitea:/github: unfurl ────────────────────────────────

def test_unfurl_repo_ref_regex():
    from app.routers.board import _REPO_REF_RE
    assert _REPO_REF_RE.match("gitea:flowdeck/flowdeck")
    assert _REPO_REF_RE.match("github:torvalds/linux")
    assert not _REPO_REF_RE.match("https://github.com/foo/bar")
    assert not _REPO_REF_RE.match("gitea:owner")
    assert not _REPO_REF_RE.match("evil:owner/repo")


def test_unfurl_repo_gitea(monkeypatch):
    import asyncio

    from app.routers.board import _unfurl_repo
    from app.services import gitea_client

    async def fake(self, owner, repo):
        return {"full_name": "flowdeck/flowdeck", "description": "A great deck",
                "default_branch": "main",
                "html_url": "https://gitea.local/flowdeck/flowdeck",
                "language": "Python"}

    monkeypatch.setattr(gitea_client.GiteaClient, "get_repo_info", fake)
    data = asyncio.run(_unfurl_repo("gitea", "flowdeck", "flowdeck"))
    assert data["title"] == "flowdeck/flowdeck"
    assert data["description"] == "A great deck"
    assert data["site_name"] == "Gitea"


def test_unfurl_repo_github_unauthenticated(monkeypatch):
    """Without settings.github_token the forge API is called unauthenticated
    (own http client), which is the production default."""
    import asyncio

    import httpx

    from app.routers.board import _unfurl_repo

    class _Resp:
        def raise_for_status(self):
            pass

        def json(self):
            return {"full_name": "torvalds/linux", "default_branch": "master",
                    "html_url": "https://github.com/torvalds/linux", "language": "C"}

    class _Client:
        def __init__(self, *a, **k):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def get(self, url, headers=None):
            return _Resp()

    monkeypatch.setattr(httpx, "AsyncClient", _Client)
    data = asyncio.run(_unfurl_repo("github", "torvalds", "linux"))
    assert data["site_name"] == "GitHub"
    assert data["language"] == "C"


def test_og_metadata_gitea_ref(client, monkeypatch):
    import secrets

    from app.services import gitea_client

    async def fake(self, owner, repo):
        return {"full_name": "flowdeck/flowdeck", "description": "A great deck",
                "default_branch": "main"}

    monkeypatch.setattr(gitea_client.GiteaClient, "get_repo_info", fake)
    uid, _, c = _user()
    csrf = secrets.token_hex(32)
    r = client.post(
        "/board/api/og/metadata",
        cookies={**c, "csrf_token": csrf},
        headers={"X-CSRF-Token": csrf},
        json={"url": "gitea:flowdeck/flowdeck"},
    )
    assert r.status_code == 200, r.text
    assert r.json()["ok"] is True
    assert r.json()["title"] == "flowdeck/flowdeck"


# ── v7.3.0 follow-up: number KPI charts + multi-DB dashboards ──────────────

def test_chart_values_preserves_zero():
    """A 0 must stay 0 (it used to be coerced to 1 for the bar count)."""
    from app.routers.collections import _chart_values
    pages = [
        {"property_values_json": '{"n": 0}'},
        {"property_values_json": '{"n": 5}'},
        {"property_values_json": '{"n": ""}'},
    ]
    assert _chart_values(pages, "n") == [0.0, 5.0]


def test_chart_number_kpi():
    import json

    from app.routers.collections import _chart_aggregate, _fmt_number, _render_chart
    pages = [
        {"title": "a", "property_values_json": json.dumps({"prio": 3})},
        {"title": "b", "property_values_json": json.dumps({"prio": 5})},
        {"title": "c", "property_values_json": json.dumps({"prio": 0})},
    ]
    assert _chart_aggregate(pages, "prio", "count") == 3.0
    assert _chart_aggregate(pages, "prio", "sum") == 8.0
    assert _chart_aggregate(pages, "prio", "avg") == 8.0 / 3.0
    assert _chart_aggregate(pages, "prio", "min") == 0.0
    assert _chart_aggregate(pages, "prio", "max") == 5.0
    assert _fmt_number(350) == "350"
    assert _fmt_number(1234) == "1.2K"
    html = _render_chart("chart", {"name": "B", "icon": "X"}, pages,
                         {"chart_type": "number", "chart_property": "prio",
                          "aggregate": "sum"})
    assert "kpi-value" in html
    assert ">8<" in html


def test_dashboard_render_multi_db(client):
    import json as _json

    a, _, ac = _user()
    w1 = _workspace(a, "W1")
    w2 = _workspace(a, "W2")
    c1 = _collection(w1, "Sales")
    c2 = _collection(w2, "Issues")
    _add_rows(c1, [{"amount": 100}, {"amount": 250}])
    _add_rows(c2, [{"prio": 1}, {"prio": 3}])
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO collection_dashboards (collection_id, name, layout_json)"
            " VALUES (?,?,?)",
            (c1, "Board", _json.dumps({
                "columns": 2,
                "widgets": [
                    {"collection_id": c1, "width": 1, "chart_type": "number",
                     "chart_property": "amount", "aggregate": "sum",
                     "title": "Revenue"},
                    {"collection_id": c2, "width": 1, "chart_type": "bar",
                     "title": "Issues"},
                ],
            })))
        conn.commit()
        did = cur.lastrowid
    r = client.get(f"/db/{c1}/dashboards/{did}", cookies=ac)
    assert r.status_code == 200, r.text
    assert "Board" in r.text
    assert "Revenue" in r.text
    assert ">350<" in r.text  # widget sums across its own database
    assert "Issues" in r.text


def test_dashboard_render_404(client):
    from fastapi.testclient import TestClient

    from app.main import app as _app

    a, _, ac = _user()
    wid = _workspace(a)
    cid = _collection(wid, "X")
    # Missing dashboard on an HTML path → 404 raised, app HTTP-404 handler
    # redirects non-API paths to /workspaces.
    c = TestClient(_app, follow_redirects=False)
    r = c.get(f"/db/{cid}/dashboards/999999", cookies=ac)
    assert r.status_code in (302, 404)