flowdeck/tests/test_v68_sites_forms.py

"""FlowDeck — v6.8.0 Sites & public Forms.

Covers migration 24, site CRUD + pages + stats, public rendering (/s/),
password/expiry gating, sitemap, form config + anonymous submission
(validation, rate limit, honeypot, embed) and auth guards.
"""
from __future__ import annotations

import json
import secrets

from conftest import anon

from app.db import get_conn


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


def _cookies(session):
    return {"flowdeck_session": session}


def _make_page(title="Hello", content="world", fmt="markdown"):
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format) VALUES (?, ?, ?, ?)",
            ("test", title, content, fmt),
        )
        pid = cur.lastrowid
        conn.commit()
    return pid


def _make_collection(name="Contacts"):
    with get_conn() as conn:
        cur = conn.execute("INSERT INTO collections (name) VALUES (?)", (name,))
        cid = cur.lastrowid
        conn.execute(
            "INSERT INTO collection_properties (collection_id, name, prop_type, position)"
            " VALUES (?, 'Name', 'text', 0)",
            (cid,),
        )
        conn.execute(
            "INSERT INTO collection_properties (collection_id, name, prop_type, position)"
            " VALUES (?, 'Email', 'email', 1)",
            (cid,),
        )
        conn.commit()
    return cid


def _create_site(client, session, pid, **kw):
    body = {"root_page_id": pid}
    body.update(kw)
    r = client.post("/api/v2/sites", json=body, cookies=_cookies(session))
    assert r.status_code == 201, r.text
    return r.json()


# ── migration ──────────────────────────────────────────────────────────────

def test_migration_24_tables(client):
    with get_conn() as conn:
        tables = {r[0] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    for t in ("sites", "site_pages", "site_views", "form_responses"):
        assert t in tables
    with get_conn() as conn:
        cols = {r[1] for r in conn.execute("PRAGMA table_info(collections)").fetchall()}
    assert "form_config_json" in cols
    with get_conn() as conn:
        v = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
    assert v >= 24


# ── sites CRUD ─────────────────────────────────────────────────────────────

def test_site_crud(client):
    session, _uid = _login(client)
    pid = _make_page("My Site Root")
    site = _create_site(client, session, pid, slug="my-site")
    assert site["slug"] == "my-site"
    sid = site["id"]

    r = client.get("/api/v2/sites", cookies=_cookies(session))
    assert r.status_code == 200
    assert any(s["id"] == sid for s in r.json())
    assert "X-Total-Count" in r.headers

    r = client.get(f"/api/v2/sites/{sid}", cookies=_cookies(session))
    assert r.status_code == 200
    assert r.json()["pages"][0]["id"] == pid

    r = client.patch(f"/api/v2/sites/{sid}", json={"title": "New title", "theme": "light"},
                     cookies=_cookies(session))
    assert r.status_code == 200
    assert r.json()["title"] == "New title"

    r = client.delete(f"/api/v2/sites/{sid}", cookies=_cookies(session))
    assert r.status_code == 200
    r = client.get(f"/api/v2/sites/{sid}", cookies=_cookies(session))
    assert r.status_code == 404


def test_site_slug_validation_and_conflict(client):
    session, _ = _login(client)
    pid = _make_page("Root")
    r = client.post("/api/v2/sites", json={"root_page_id": pid, "slug": "BAD SLUG!!"},
                    cookies=_cookies(session))
    assert r.status_code == 400
    _create_site(client, session, pid, slug="taken-slug")
    pid2 = _make_page("Root 2")
    r = client.post("/api/v2/sites", json={"root_page_id": pid2, "slug": "taken-slug"},
                    cookies=_cookies(session))
    assert r.status_code == 409


def test_site_requires_auth(client):
    anon(client)
    pid = _make_page("Root")
    r = client.post("/api/v2/sites", json={"root_page_id": pid})
    assert r.status_code == 401


def test_site_isolation_between_users(client):
    s1, _ = _login(client)
    s2, _ = _login(client)
    pid = _make_page("Root")
    site = _create_site(client, s1, pid, slug="private-site")
    r = client.get(f"/api/v2/sites/{site['id']}", cookies=_cookies(s2))
    assert r.status_code == 404


def test_site_pages_add_remove(client):
    session, _ = _login(client)
    pid = _make_page("Root")
    site = _create_site(client, session, pid, slug="nav-site")
    sid = site["id"]
    pid2 = _make_page("Second page")
    r = client.post(f"/api/v2/sites/{sid}/pages", json={"page_id": pid2},
                    cookies=_cookies(session))
    assert r.status_code == 200
    assert len(r.json()["pages"]) == 2
    r = client.delete(f"/api/v2/sites/{sid}/pages/{pid2}", cookies=_cookies(session))
    assert r.status_code == 200
    r = client.delete(f"/api/v2/sites/{sid}/pages/{pid}", cookies=_cookies(session))
    assert r.status_code == 400  # root protected


# ── public rendering ───────────────────────────────────────────────────────

def test_public_site_home_and_subpage(client):
    session, _ = _login(client)
    pid = _make_page("Welcome Home", "hello public")
    site = _create_site(client, session, pid, slug="public-home")
    r = client.get("/s/public-home")
    assert r.status_code == 200
    assert "Welcome Home" in r.text
    assert "hello public" in r.text
    # sub-page by slug
    pid2 = _make_page("Second Page", "second body")
    client.post(f"/api/v2/sites/{site['id']}/pages", json={"page_id": pid2},
                cookies=_cookies(session))
    r = client.get("/s/public-home/second-page")
    assert r.status_code == 200
    assert "second body" in r.text
    # unknown page
    r = client.get("/s/public-home/nope")
    assert r.status_code == 404


def test_public_site_blocks_render(client):
    session, _ = _login(client)
    content = json.dumps([{"type": "heading_1", "content": "Big Title"},
                          {"type": "paragraph", "content": "para body"}])
    pid = _make_page("Blocks", content, fmt="blocks")
    _create_site(client, session, pid, slug="blocks-site")
    r = client.get("/s/blocks-site")
    assert r.status_code == 200
    assert "Big Title" in r.text


def test_public_site_404(client):
    r = client.get("/s/does-not-exist")
    assert r.status_code == 404


def test_site_views_counted(client):
    from app.routers.sites import _reset_form_rate  # noqa - ensure router loaded
    _ = _reset_form_rate
    session, _ = _login(client)
    pid = _make_page("Root")
    site = _create_site(client, session, pid, slug="stats-site")
    client.get("/s/stats-site")
    client.get("/s/stats-site")
    r = client.get(f"/api/v2/sites/{site['id']}/stats", cookies=_cookies(session))
    assert r.status_code == 200
    assert r.json()["total_views"] >= 2


def test_site_password_gate(client):
    session, _ = _login(client)
    pid = _make_page("Secret", "top secret body")
    site = _create_site(client, session, pid, slug="secret-site")
    client.patch(f"/api/v2/sites/{site['id']}", json={"password": "s3cr3t"},
                 cookies=_cookies(session))
    # anonymous client without cookies
    from fastapi.testclient import TestClient

    from app.main import app
    anon = TestClient(app)
    r = anon.get("/s/secret-site")
    assert r.status_code == 401
    r = anon.post("/s/secret-site/auth", json={"password": "wrong"})
    assert r.status_code == 401
    r = anon.post("/s/secret-site/auth", json={"password": "s3cr3t"})
    assert r.status_code == 200
    r = anon.get("/s/secret-site")
    assert r.status_code == 200
    assert "top secret body" in r.text


def test_site_expiry(client):
    session, _ = _login(client)
    pid = _make_page("Root")
    _create_site(client, session, pid, slug="old-site",
                   expires_at="2000-01-01T00:00:00Z")
    r = client.get("/s/old-site")
    assert r.status_code == 410


def test_site_sitemap(client):
    session, _ = _login(client)
    pid = _make_page("Root")
    _create_site(client, session, pid, slug="map-site")
    r = client.get("/s/map-site/sitemap.xml")
    assert r.status_code == 200
    assert "/s/map-site" in r.text


def test_site_noindex_meta(client):
    session, _ = _login(client)
    pid = _make_page("Root")
    _create_site(client, session, pid, slug="noindex-site", noindex=True)
    r = client.get("/s/noindex-site")
    assert "noindex" in r.text


# ── forms ──────────────────────────────────────────────────────────────────

def _enable_form(client, session, cid, **kw):
    cfg = {"enabled": True, "fields": ["Name", "Email"], "required": ["Name"]}
    cfg.update(kw)
    r = client.put(f"/api/v2/collections/{cid}/form", json=cfg, cookies=_cookies(session))
    assert r.status_code == 200, r.text
    return r.json()["form"]


def test_form_config_crud(client):
    session, _ = _login(client)
    cid = _make_collection()
    r = client.get(f"/api/v2/collections/{cid}/form", cookies=_cookies(session))
    assert r.status_code == 200
    form = _enable_form(client, session, cid)
    assert form["enabled"] is True
    assert form["public_token"].startswith("f_")


def test_public_form_get_and_submit_json(client):
    from app.routers.sites import _reset_form_rate
    _reset_form_rate()
    session, _ = _login(client)
    cid = _make_collection()
    form = _enable_form(client, session, cid)
    token = form["public_token"]
    r = client.get(f"/f/{token}")
    assert r.status_code == 200
    assert "Name" in r.text
    r = client.post(f"/f/{token}", json={"Name": "Alice", "Email": "[email protected]"})
    assert r.status_code == 200, r.text
    with get_conn() as conn:
        row = conn.execute(
            "SELECT title FROM collection_pages WHERE collection_id=? ORDER BY id DESC LIMIT 1",
            (cid,)).fetchone()
        assert row is not None


def test_public_form_required_and_404(client):
    from app.routers.sites import _reset_form_rate
    _reset_form_rate()
    session, _ = _login(client)
    cid = _make_collection()
    form = _enable_form(client, session, cid)
    r = client.post(f"/f/{form['public_token']}", json={"Email": "[email protected]"})
    assert r.status_code == 400
    r = client.get("/f/f_doesnotexist123")
    assert r.status_code == 404
    r = client.post("/f/f_doesnotexist123", json={"Name": "x"})
    assert r.status_code == 404


def test_public_form_honeypot(client):
    from app.routers.sites import _reset_form_rate
    _reset_form_rate()
    session, _ = _login(client)
    cid = _make_collection()
    form = _enable_form(client, session, cid)
    r = client.post(f"/f/{form['public_token']}",
                    json={"Name": "Spammer", "__hp": "bot"})
    assert r.status_code == 400


def test_public_form_rate_limit(client):
    from app.routers.sites import _reset_form_rate
    _reset_form_rate()
    session, _ = _login(client)
    cid = _make_collection()
    form = _enable_form(client, session, cid)
    token = form["public_token"]
    last = None
    for i in range(21):
        last = client.post(f"/f/{token}", json={"Name": f"U{i}"})
    assert last.status_code == 429


def test_public_form_embed_mode(client):
    from app.routers.sites import _reset_form_rate
    _reset_form_rate()
    session, _ = _login(client)
    cid = _make_collection()
    form = _enable_form(client, session, cid, title="Contact Us")
    r = client.get(f"/f/{form['public_token']}?embed=1")
    assert r.status_code == 200
    assert "<h1>" not in r.text  # chrome stripped in embed