flowdeck/tests/test_v57_db_advanced.py

"""FlowDeck — v5.7.0 Database Avancée (Pt. 2).

Covers: person + auto properties (created/last-edited time & by), property
groups, per-user saved views, swimlanes/WIP/card/calendar view config, row
covers and the multi-view editor rendering.
"""
from __future__ import annotations

import json
import os
import pathlib
import secrets
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-tests"
    os.environ["RATE_LIMIT_ENABLED"] = "false"
    os.environ["PUBLIC_API_INSECURE_OK"] = "true"

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

    from app.db import init_db
    from app.main import app
    init_db()

    yield login_test_client(TestClient(app))

    os.unlink(db_path)


def _add_collection(client, name="DB"):
    return client.post("/db/api", json={"name": name}).json()["id"]


def _add_prop(client, cid, name, ptype, **extra):
    body = {"name": name, "prop_type": ptype}
    body.update(extra)
    r = client.post(f"/db/{cid}/properties/api", json=body)
    assert r.status_code == 200, r.text
    return r.json()["id"]


def _props(client, cid):
    return client.get(f"/db/{cid}/properties/api").json()["properties"]


def _login(client):
    from app.auth.session import SessionManager
    from app.db import get_conn

    with get_conn() as conn:
        login = f"v57admin_{secrets.token_hex(4)}"
        conn.execute(
            "INSERT INTO users (login, full_name, email, password_hash, is_admin) "
            "VALUES (?, 'V57 Admin', ?, ?, 1)",
            (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, "is_admin": 1})
    csrf = client.get("/api/csrf-token", cookies={"flowdeck_session": session}).json()["csrf_token"]
    return session, csrf, uid


# ── Schema / migration ──

def test_migration_v57_columns(client):
    from app.db import get_conn
    with get_conn() as conn:
        pcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_properties)").fetchall()}
        vcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_views)").fetchall()}
        cpcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_pages)").fetchall()}
        version = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
    assert "group_name" in pcols
    assert "created_by" in vcols
    assert "cover_url" in cpcols
    assert version >= 10


# ── Person property ──

def test_person_property_value_roundtrip(client):
    cid = _add_collection(client)
    pid = _add_prop(client, cid, "Assignee", "person")
    r = client.post(f"/db/{cid}/pages/api", json={
        "title": "Row", "properties": {str(pid): [{"id": 7, "login": "alice"}]}
    })
    assert r.status_code == 200
    page = client.get(f"/db/pages/{r.json()['id']}/api").json()
    pv = json.loads(page["property_values_json"])
    assert pv[str(pid)] == [{"id": 7, "login": "alice"}]


def test_person_property_rejects_non_list(client):
    cid = _add_collection(client)
    pid = _add_prop(client, cid, "Owner", "person")
    r = client.post(f"/db/{cid}/pages/api", json={
        "title": "Row", "properties": {str(pid): "alice"}
    })
    assert r.status_code == 400
    assert "person" in r.json()["detail"].lower()


def test_collection_members_endpoint(client):
    from app.db import get_conn
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO users (login, full_name, email, avatar_color) VALUES ('bob', 'Bob', '[email protected]', '#123456')"
        )
        conn.commit()
    cid = _add_collection(client)
    r = client.get(f"/db/{cid}/members/api")
    assert r.status_code == 200
    members = r.json()["members"]
    assert any(m["login"] == "bob" for m in members)


# ── Auto properties ──

def test_auto_properties_filled_on_create_and_update(client):
    cid = _add_collection(client)
    ct = _add_prop(client, cid, "Created", "created_time")
    cb = _add_prop(client, cid, "Created by", "created_by")
    lt = _add_prop(client, cid, "Edited", "last_edited_time")
    lb = _add_prop(client, cid, "Edited by", "last_edited_by")

    r = client.post(f"/db/{cid}/pages/api", json={"title": "Task"})
    assert r.status_code == 200
    page_id = r.json()["id"]
    pv = json.loads(client.get(f"/db/pages/{page_id}/api").json()["property_values_json"])
    assert pv[str(ct)]
    assert pv[str(lt)]
    assert pv[str(cb)]["login"] == "tester"
    assert pv[str(lb)]["login"] == "tester"

    created = pv[str(ct)]
    # Partial update keeps created_time and refreshes last_edited_time.
    r = client.put(f"/db/pages/{page_id}/api", json={"properties": {}})
    assert r.status_code == 200
    pv2 = json.loads(client.get(f"/db/pages/{page_id}/api").json()["property_values_json"])
    assert pv2[str(ct)] == created
    assert pv2[str(lt)] >= created


def test_partial_update_merges_properties(client):
    cid = _add_collection(client)
    a = _add_prop(client, cid, "A", "text")
    b = _add_prop(client, cid, "B", "text")
    r = client.post(f"/db/{cid}/pages/api", json={
        "title": "Row", "properties": {str(a): "one", str(b): "two"}
    })
    pid = r.json()["id"]
    client.put(f"/db/pages/{pid}/api", json={"properties": {str(a): "changed"}})
    pv = json.loads(client.get(f"/db/pages/{pid}/api").json()["property_values_json"])
    assert pv[str(a)] == "changed"
    assert pv[str(b)] == "two"


# ── Property groups ──

def test_property_group_assignment(client):
    cid = _add_collection(client)
    p1 = _add_prop(client, cid, "Name2", "text", group_name="Basics")
    p2 = _add_prop(client, cid, "Status2", "status")
    props = {p["id"]: p for p in _props(client, cid)}
    assert props[p1]["group_name"] == "Basics"

    r = client.post(f"/db/{cid}/property-groups/api", json={
        "groups": [{"name": "Workflow", "property_ids": [p1, p2]}]
    })
    assert r.status_code == 200
    props = {p["id"]: p for p in _props(client, cid)}
    assert props[p1]["group_name"] == "Workflow"
    assert props[p2]["group_name"] == "Workflow"

    # Clearing groups
    client.post(f"/db/{cid}/property-groups/api", json={"groups": []})
    props = {p["id"]: p for p in _props(client, cid)}
    assert props[p1]["group_name"] == ""


# ── Per-user saved views ──

def test_saved_views_are_per_user(client):
    cookie, csrf, uid = _login(client)
    cid = _add_collection(client)
    r = client.post(f"/db/{cid}/views/save-as",
                    json={"name": "My Board", "view_type": "board", "config": {"group_by": 1}},
                    cookies={"flowdeck_session": cookie}, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200
    view = r.json()
    assert view["created_by"] == uid
    assert view["view_type"] == "board"

    # The owner sees it.
    views = client.get(f"/db/{cid}/views/api", cookies={"flowdeck_session": cookie}).json()["views"]
    assert any(v["id"] == view["id"] for v in views)

    # Duplicate + delete.
    dup = client.post(f"/db/views/{view['id']}/duplicate", json={},
                      cookies={"flowdeck_session": cookie}, headers={"X-CSRF-Token": csrf}).json()
    assert dup["id"] != view["id"]
    d = client.delete(f"/db/views/{view['id']}/api", cookies={"flowdeck_session": cookie},
                      headers={"X-CSRF-Token": csrf})
    assert d.status_code == 200
    views = client.get(f"/db/{cid}/views/api", cookies={"flowdeck_session": cookie}).json()["views"]
    assert not any(v["id"] == view["id"] for v in views)


def test_view_config_extended_keys(client):
    cid = _add_collection(client)
    views = client.get(f"/db/{cid}/views/api").json()["views"]
    vid = views[0]["id"]
    payload = {
        "sub_group_by": 3, "wip_limits": {"In progress": 2}, "card_size": "small",
        "cover_mode": "color", "card_properties": [1, 2], "date_property": 4,
    }
    r = client.put(f"/db/views/{vid}/config", json=payload)
    assert r.status_code == 200
    cfg = r.json()["config"]
    for k, v in payload.items():
        assert cfg[k] == v


# ── Row cover ──

def test_row_cover_url(client):
    cid = _add_collection(client)
    r = client.post(f"/db/{cid}/pages/api", json={"title": "Row", "cover_url": "https://x.test/a.png"})
    pid = r.json()["id"]
    page = client.get(f"/db/pages/{pid}/api").json()
    assert page["cover_url"] == "https://x.test/a.png"

    client.put(f"/db/pages/{pid}/api", json={"cover_url": "https://x.test/b.png"})
    page = client.get(f"/db/pages/{pid}/api").json()
    assert page["cover_url"] == "https://x.test/b.png"


# ── Table-data payload includes views ──

def test_table_data_includes_views(client):
    cid = _add_collection(client)
    data = client.get(f"/api/collections/{cid}/table-data").json()
    assert "views" in data
    assert isinstance(data["views"], list)
    assert len(data["views"]) >= 1


# ── Full-page editor renders the multi-view component ──

def test_collection_page_renders_multi_view(client):
    cookie, csrf, uid = _login(client)
    page = client.post(
        "/board/api/pages?title=My%20DB&section=Private&parent_id=0",
        headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": cookie},
    ).json()["id"]
    r = client.post(f"/api/pages/{page}/convert-to-database", json={"name": "My DB"},
                    headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": cookie})
    assert r.status_code == 200
    resp = client.get(f"/pages/{page}", cookies={"flowdeck_session": cookie})
    assert resp.status_code == 200
    assert "db-view-bar" in resp.text
    # New view types + settings are available client-side.
    # (A27 : le JS de la DB est extrait dans static/js/database_table.js.)
    src = pathlib.Path("static/js/database_table.js").read_text(encoding="utf-8")
    assert "FlowDeckDB" in src
    assert "db-board" in src
    assert "db-cal-grid" in src
    assert "db-gallery" in src