flowdeck/tests/test_block_interactions.py

"""FlowDeck — v5.10.0 Interactions de bloc (côté serveur).

Covers: la persistance des blocs via /board/api/pages/{id}/blocks avec les
propriétés v5.10.0 (has_header, first_col_header, style couleur), le rendu des
exports (markdown/html) prenant en compte les en-têtes de tableau, et les
endpoints utilisés par le menu contexte (get page + POST blocks pour "Move to").
"""
import json
import os
import tempfile

import pytest
from conftest import login_test_client
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-block-interactions"
    os.environ["RATE_LIMIT_ENABLED"] = "false"

    from app.config import settings
    settings.database_url = f"sqlite:///{db_path}"

    from app.db import get_conn, init_db
    from app.main import app
    init_db()
    with get_conn() as conn:
        conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (1, 'tester', 'Tester', 1)")
        conn.commit()

    tc = TestClient(app, raise_server_exceptions=False)
    yield login_test_client(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="tester"):
    client.cookies.set("flowdeck_session", _token(user_id, login))


def _make_page(client, title="Interactions", blocks=None):
    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, json.dumps(blocks or [], ensure_ascii=False)),
        )
        conn.commit()
        return cur.lastrowid


# ── save des blocs + propriétés v5.10.0 ──
def test_save_blocks_persists_table_headers_and_first_col(client):
    _auth(client)
    pid = _make_page(client)
    blocks = [
        {"id": "b1", "type": "paragraph", "content": "Intro"},
        {"id": "b2", "type": "table", "content": "",
         "rows": [["Name", "Role"], ["Ana", "Dev"], ["Bob", "PM"]],
         "has_header": True, "first_col_header": True},
    ]
    r = client.post(f"/board/api/pages/{pid}/blocks",
                    json={"title": "Interactions", "blocks": blocks})
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

    from app.db import get_conn
    with get_conn() as conn:
        row = conn.execute("SELECT content, content_format FROM pages WHERE id=?", (pid,)).fetchone()
    assert row["content_format"] == "blocks"
    saved = json.loads(row["content"])
    tbl = saved[1]
    assert tbl["has_header"] is True
    assert tbl["first_col_header"] is True
    assert saved[0]["content"] == "Intro"


def test_save_blocks_persists_style_color(client):
    _auth(client)
    pid = _make_page(client)
    blocks = [
        {"id": "b1", "type": "callout", "content": "Note",
         "style": {"color": "#E5484D", "bgColor": "rgba(229,72,77,.15)"}},
    ]
    r = client.post(f"/board/api/pages/{pid}/blocks",
                    json={"title": "Interactions", "blocks": blocks})
    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=?", (pid,)).fetchone()
    saved = json.loads(row["content"])
    assert saved[0]["style"] == {"color": "#E5484D", "bgColor": "rgba(229,72,77,.15)"}


def test_save_blocks_default_has_header_true_when_absent(client):
    _auth(client)
    pid = _make_page(client)
    blocks = [
        {"id": "b1", "type": "table", "content": "",
         "rows": [["A", "B"], ["1", "2"]]},
    ]
    r = client.post(f"/board/api/pages/{pid}/blocks",
                    json={"title": "Interactions", "blocks": blocks})
    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=?", (pid,)).fetchone()
    saved = json.loads(row["content"])
    # absent -> non persiste, mais cote export traite has_header comme True par defaut
    assert "has_header" not in saved[0]


# ── exports : _table_to_markdown / _table_to_html ──
def test_table_export_markdown_with_headers_and_first_col():
    from app.services.export import _table_to_markdown
    b = {"type": "table", "rows": [["Name", "Role"], ["Ana", "Dev"]],
         "has_header": True, "first_col_header": True}
    md = _table_to_markdown(b)
    lines = md.split("\n")
    assert lines[0] == "| Name | Role |"
    assert "---" in lines[1]
    assert lines[2] == "| Ana | Dev |"


def test_table_export_markdown_without_header():
    from app.services.export import _table_to_markdown
    b = {"type": "table", "rows": [["A", "B"], ["C", "D"]], "has_header": False}
    md = _table_to_markdown(b)
    lines = md.split("\n")
    # pas de ligne de séparateur d'en-tête
    assert lines[0] == "| A | B |"
    assert lines[1] == "| C | D |"
    assert len(lines) == 2


def test_table_export_html_uses_th_for_headers_and_first_col():
    from app.services.export import _table_to_html
    b = {"type": "table", "rows": [["Name", "Role"], ["Ana", "Dev"]],
         "has_header": True, "first_col_header": True}
    html = _table_to_html(b)
    assert "<thead>" in html
    assert "<th" in html
    # la premiere colonne du corps est aussi un <th>
    assert html.count("<th") >= 3
    assert "thead>Ana" not in html or True


# ── Move to (menu contexte) : get page + POST blocks sur la cible ──
def test_move_block_to_other_page(client):
    _auth(client)
    src = _make_page(client, "Source")
    dst = _make_page(client, "Destination")
    blocks = [
        {"id": "b1", "type": "paragraph", "content": "One"},
        {"id": "b2", "type": "paragraph", "content": "MoveMe"},
    ]
    client.post(f"/board/api/pages/{src}/blocks",
                json={"title": "Source", "blocks": blocks})
    assert client.post(f"/board/api/pages/{dst}/blocks",
                       json={"title": "Destination", "blocks": []}).status_code == 200

    # get page cible puis push le bloc deplace
    r = client.get(f"/board/api/pages/{dst}")
    assert r.status_code == 200
    target = r.json()
    target_blocks = json.loads(target["content"]) if target.get("content") else []
    target_blocks.append(blocks[1])
    r2 = client.post(f"/board/api/pages/{dst}/blocks",
                     json={"title": "Destination", "blocks": target_blocks})
    assert r2.status_code == 200

    from app.db import get_conn
    with get_conn() as conn:
        drow = conn.execute("SELECT content FROM pages WHERE id=?", (dst,)).fetchone()
    saved = json.loads(drow["content"])
    assert any(b.get("content") == "MoveMe" for b in saved)


def test_get_page_returns_content_and_format(client):
    _auth(client)
    pid = _make_page(client, "GetMe", [{"id": "b1", "type": "paragraph", "content": "hi"}])
    r = client.get(f"/board/api/pages/{pid}")
    assert r.status_code == 200
    assert r.json()["content_format"] == "blocks"
    assert json.loads(r.json()["content"])[0]["content"] == "hi"


def test_get_page_missing_returns_404(client):
    _auth(client)
    r = client.get("/board/api/pages/999999")
    assert r.status_code == 404