flowdeck/tests/test_web_clipper.py

"""FlowDeck — v6.2.0 Web Clipper tests."""
from __future__ import annotations

import json

from conftest import anon

from app.auth.session import SessionManager


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


def _as(client, user_id, login="user", is_admin=0):
    client.cookies.set("flowdeck_session", _token(user_id, login, is_admin))


def _insert_user(login="clipper_user", email="[email protected]"):
    from app.db import get_conn

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


def _insert_workspace(owner_id, name="ClipWS"):
    from app.db import get_conn

    with get_conn() as conn:
        cur = conn.execute("INSERT INTO workspaces (name, owner_id) VALUES (?,?)", (name, owner_id))
        conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?, 'admin')", (cur.lastrowid, owner_id))
        conn.commit()
        return cur.lastrowid


# ── Service unit tests ──

def test_sanitize_removes_script(client):
    from app.services.web_clipper import sanitize_html

    html = '<p>Hello</p><script>alert(1)</script><div onclick="evil()">x</div><a href="javascript:alert(1)">click</a>'
    clean = sanitize_html(html)
    assert "<script" not in clean
    assert "onclick" not in clean
    assert "javascript:" not in clean
    assert "Hello" in clean


def test_html_to_blocks_basic(client):
    from app.services.web_clipper import html_to_blocks

    html = "<h1>Title</h1><p>Paragraph text</p><ul><li>Item A</li><li>Item B</li></ul><blockquote>Quote</blockquote>"
    blocks = html_to_blocks(html)
    types = [b["type"] for b in blocks]
    assert "heading_1" in types
    assert "paragraph" in types
    assert "bulleted_list" in types
    assert "quote" in types


def test_html_to_blocks_image(client):
    from app.services.web_clipper import html_to_blocks

    html = '<p><img src="https://example.com/img.png" alt="alt"></p><p>Text</p>'
    blocks = html_to_blocks(html)
    assert any(b["type"] == "image" and b["src"] == "https://example.com/img.png" for b in blocks)


def test_extract_article(client):
    from app.services.web_clipper import extract_article

    html = "<html><head><title> My Article </title></head><body><article><h1>Heading</h1><p>Body text</p></article></body></html>"
    res = extract_article(html, url="https://example.com/a")
    assert res["title"] == "My Article"
    assert any(b["type"] == "heading_1" for b in res["blocks"])


# ── API tests ──

def test_clip_requires_auth(client):
    anon(client)
    r = client.post("/api/v2/web-clipper/clip", json={"url": "https://example.com", "title": "T"})
    assert r.status_code == 401


def test_clip_article_creates_page(client):
    uid = _insert_user("clip_a")
    _insert_workspace(uid)
    _as(client, uid, "clip_a")
    html = "<html><head><title>Example Article</title></head><body><h1>Hello</h1><p>World paragraph</p><p><img src='https://example.com/x.png' alt='x'></p></body></html>"
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={"url": "https://example.com/article", "title": "Example Article", "content": html, "content_type": "article", "device_id": "dev1"},
    )
    assert r.status_code == 200, r.text
    data = r.json()
    assert data["status"] == "ok"
    assert data["page_id"]
    # Verify page in DB
    from app.db import get_conn

    with get_conn() as conn:
        row = conn.execute("SELECT title, content, workspace_id FROM pages WHERE id=?", (data["page_id"],)).fetchone()
        assert row is not None
        assert row["title"] == "Example Article"
        blocks = json.loads(row["content"])
        assert any(b["type"] == "paragraph" for b in blocks)
        # Clip logged
        cnt = conn.execute("SELECT COUNT(*) FROM extension_clips WHERE user_id=?", (uid,)).fetchone()[0]
        assert cnt == 1


def test_clip_bookmark(client):
    uid = _insert_user("clip_b")
    _insert_workspace(uid)
    _as(client, uid, "clip_b")
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={"url": "https://example.com", "title": "My Bookmark", "content_type": "bookmark", "device_id": "dev-b"},
    )
    assert r.status_code == 200
    data = r.json()
    from app.db import get_conn

    with get_conn() as conn:
        row = conn.execute("SELECT content FROM pages WHERE id=?", (data["page_id"],)).fetchone()
        blocks = json.loads(row["content"])
        assert any(b["type"] == "bookmark" and b["url"] == "https://example.com" for b in blocks)


def test_clip_selection(client):
    uid = _insert_user("clip_sel")
    _insert_workspace(uid)
    _as(client, uid, "clip_sel")
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={
            "url": "https://example.com/p",
            "title": "Sel Test",
            "content_type": "selection",
            "selection_html": "<p>Selected <b>text</b> here</p>",
            "device_id": "dev-sel",
        },
    )
    assert r.status_code == 200
    from app.db import get_conn

    with get_conn() as conn:
        row = conn.execute("SELECT content FROM pages WHERE id=?", (r.json()["page_id"],)).fetchone()
        blocks = json.loads(row["content"])
        assert any("Selected" in b.get("content", "") for b in blocks)


def test_clip_screenshot(client):
    uid = _insert_user("clip_shot")
    _insert_workspace(uid)
    _as(client, uid, "clip_shot")
    b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={"url": "https://example.com", "title": "Shot", "content_type": "screenshot", "image_base64": b64, "device_id": "dev-shot"},
    )
    assert r.status_code == 200
    from app.db import get_conn

    with get_conn() as conn:
        row = conn.execute("SELECT content FROM pages WHERE id=?", (r.json()["page_id"],)).fetchone()
        blocks = json.loads(row["content"])
        assert any(b["type"] == "image" for b in blocks)


def test_clip_sanitization_script_stripped(client):
    uid = _insert_user("clip_sani")
    _insert_workspace(uid)
    _as(client, uid, "clip_sani")
    html = '<p>ok</p><script>alert(1)</script>'
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={"url": "https://example.com", "content": html, "content_type": "article", "device_id": "dev-sani"},
    )
    assert r.status_code == 200
    from app.db import get_conn

    with get_conn() as conn:
        row = conn.execute("SELECT content FROM pages WHERE id=?", (r.json()["page_id"],)).fetchone()
        assert "alert" not in row["content"]
        assert "ok" in row["content"]


def test_clip_rate_limit(client):
    uid = _insert_user("clip_rate")
    _insert_workspace(uid)
    _as(client, uid, "clip_rate")
    # Reset rate store bucket for isolation (uses in-memory dict keyed by user:device)
    from app.services.web_clipper import _rate_store

    _rate_store.clear()
    for i in range(50):
        r = client.post(
            "/api/v2/web-clipper/clip",
            json={"url": f"https://example.com/{i}", "title": f"T{i}", "content": "<p>hi</p>", "device_id": "rate-dev"},
        )
        assert r.status_code == 200, r.text
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={"url": "https://example.com/overflow", "title": "overflow", "content": "<p>hi</p>", "device_id": "rate-dev"},
    )
    assert r.status_code == 429
    _rate_store.clear()


def test_clip_bearer_token_auth(client):
    import hashlib

    from app.db import get_conn

    uid = _insert_user("clip_bearer")
    _insert_workspace(uid)
    # Create api_token
    token = "fd_test_clipper_bearer_123"
    th = hashlib.sha256(token.encode()).hexdigest()
    with get_conn() as conn:
        conn.execute("INSERT INTO api_tokens (user_id, name, token_hash, token_prefix) VALUES (?,?,?,?)", (uid, "test", th, token[:12]))
        conn.commit()
    # No session cookie
    anon(client)
    r = client.post(
        "/api/v2/web-clipper/clip",
        json={"url": "https://example.com/bearer", "title": "BearerClip", "content": "<p>via bearer</p>", "device_id": "bearer-dev"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r.status_code == 200, r.text
    with get_conn() as conn:
        row = conn.execute("SELECT title FROM pages WHERE id=?", (r.json()["page_id"],)).fetchone()
        assert row["title"] == "BearerClip"


def test_status_and_devices_flow(client):
    uid = _insert_user("clip_status")
    _insert_workspace(uid)
    # unauth status
    anon(client)
    r = client.get("/api/v2/web-clipper/status")
    assert r.status_code == 200
    assert r.json()["authenticated"] is False
    # auth status
    _as(client, uid, "clip_status")
    r = client.get("/api/v2/web-clipper/status")
    assert r.json()["authenticated"] is True
    # register device via verify
    r = client.post("/api/v2/web-clipper/auth/verify", json={"device_id": "dev-xyz-123", "device_name": "Chrome — Test", "extension_name": "chrome"})
    assert r.status_code == 200
    data = r.json()
    assert data["device_id"] == "dev-xyz-123"
    # second verify same device reports existing
    r2 = client.post("/api/v2/web-clipper/auth/verify", json={"device_id": "dev-xyz-123", "device_name": "Chrome — Test"})
    assert r2.json().get("existing") is True
    # list devices
    r = client.get("/api/v2/web-clipper/devices")
    assert r.status_code == 200
    devs = r.json()["devices"]
    assert any(d["device_id"] == "dev-xyz-123" for d in devs)
    did = [d for d in devs if d["device_id"] == "dev-xyz-123"][0]["id"]
    # revoke
    r = client.delete(f"/api/v2/web-clipper/devices/{did}")
    assert r.status_code == 200
    r = client.get("/api/v2/web-clipper/devices")
    remaining = [d for d in r.json()["devices"] if d["id"] == did]
    assert remaining[0]["revoked"] == 1


def test_extensions_page(client):
    r = client.get("/extensions")
    assert r.status_code == 200
    assert "Web Clipper" in r.text
    assert "Chrome" in r.text


def test_clip_invalid_url_rejected(client):
    uid = _insert_user("clip_badurl")
    _insert_workspace(uid)
    _as(client, uid, "clip_badurl")
    r = client.post("/api/v2/web-clipper/clip", json={"url": "javascript:alert(1)", "title": "x", "device_id": "d"})
    assert r.status_code == 400


def test_clip_payload_too_large(client):
    uid = _insert_user("clip_big")
    _insert_workspace(uid)
    _as(client, uid, "clip_big")
    big = "x" * (11 * 1024 * 1024)  # 11 MB string
    r = client.post("/api/v2/web-clipper/clip", json={"url": "https://example.com", "title": "big", "content": big, "device_id": "big-dev"})
    # Expect 413 or 400 due to body size / content-length
    assert r.status_code in (413, 400, 422)