kanbante/tests/test_palier5.py

"""Palier 5 (0.8.0): palette de commandes (#10) + digest hebdo (#13)."""

import httpx
import pytest

from app import digest as digest_service
from app import models
from tests.conftest import csrf_from


# ── Palette (#10) ─────────────────────────────────────────────────


def test_palette_endpoint_projects_and_issues(client, fake_gitea):
    r = client.get("/api/palette", params={"q": "tâche"})
    assert r.status_code == 200
    data = r.json()
    assert any(p["full_name"] == "Projets/demo" for p in data["projects"])
    titles = [i["title"] for i in data["issues"]]
    assert "Tâche un" in titles
    found = next(i for i in data["issues"] if i["title"] == "Tâche un")
    assert found["repo"] == "Projets/demo"
    assert found["column"] == "backlog"


def test_palette_short_query_skips_issue_search(client):
    r = client.get("/api/palette", params={"q": "x"})
    assert r.status_code == 200
    assert r.json()["issues"] == []
    assert r.json()["projects"]  # projects are always there


def test_palette_markup_on_pages(client):
    for path in ("/", "/focus", "/capture"):
        r = client.get(path)
        assert r.status_code == 200
        assert 'id="palette-overlay"' in r.text, path
        assert "/static/js/palette.js" in r.text, path
        assert "palette-trigger" in r.text, path


# ── Digest : composition (#13) ────────────────────────────────────


def test_digest_markdown_format():
    md = digest_service.format_digest_markdown(
        "2026-10-05 au 2026-10-11",
        [{"repo": "Projets/demo", "number": 3, "title": "Ancienne tâche",
          "html_url": "http://gitea.test/x"}],
        created_count=2,
        moved_count=5,
        focus_groups=[{
            "full_name": "Projets/demo",
            "cards": [{"issue": {"number": 5, "title": "Tâche focus"},
                       "column": "🚧 En cours"}],
        }],
    )
    assert "# 🧠 Kanbante — semaine du 2026-10-05" in md
    assert "## ✅ Livré (1)" in md
    assert "Projets/demo#3" in md
    assert "Tâche focus" in md
    assert "Déplacements de cartes : 5" in md


def test_render_mcp_args_escapes_content():
    args = digest_service.render_mcp_args(
        '{"title": "{title}", "content": "{content}", "folder": "Kanbante"}',
        'Titre "cité"',
        "Ligne 1\nLigne 2 avec \\ barre",
    )
    assert args["title"] == 'Titre "cité"'
    assert args["content"] == "Ligne 1\nLigne 2 avec \\ barre"
    assert args["folder"] == "Kanbante"
    with pytest.raises(ValueError):
        digest_service.render_mcp_args("{pas json", "t", "c")


def test_digest_page_renders(client):
    r = client.get("/digest")
    assert r.status_code == 200
    assert "Digest hebdo" in r.text
    assert "approbation" in r.text


# ── Digest : livraison ────────────────────────────────────────────


def _mcp_handler_factory(calls: list):
    def handler(request: httpx.Request) -> httpx.Response:
        import json as _json
        payload = _json.loads(request.content.decode() or "{}")
        calls.append(payload.get("method"))
        method = payload.get("method")
        if method == "initialize":
            return httpx.Response(200, json={"jsonrpc": "2.0", "id": 1,
                "result": {"protocolVersion": "2024-11-05", "capabilities": {}}})
        if method == "notifications/initialized":
            return httpx.Response(202, json={})
        if method == "tools/list":
            return httpx.Response(200, json={"jsonrpc": "2.0", "id": 2,
                "result": {"tools": [{"name": "create_note"}, {"name": "search"}]}})
        if method == "tools/call":
            calls.append(payload["params"])
            return httpx.Response(200, json={"jsonrpc": "2.0", "id": 3,
                "result": {"content": [{"type": "text", "text": "note créée"}]}})
        return httpx.Response(200, json={})
    return handler


@pytest.fixture
def mcp_transport(monkeypatch):
    calls: list = []
    transport = httpx.MockTransport(_mcp_handler_factory(calls))
    real_async_client = httpx.AsyncClient

    def factory(*args, **kwargs):
        kwargs["transport"] = transport
        return real_async_client(*args, **kwargs)

    monkeypatch.setattr(httpx, "AsyncClient", factory)
    return calls


async def test_mcp_list_tools(mcp_transport):
    ok, tools = await digest_service.mcp_list_tools("http://obsigate.test/mcp", "")
    assert ok is True
    assert tools == ["create_note", "search"]


async def test_mcp_call_tool(mcp_transport):
    ok, message = await digest_service.mcp_call_tool(
        "http://obsigate.test/mcp", "", "create_note",
        {"title": "T", "content": "C"},
    )
    assert ok is True
    assert "note créée" in message
    call = next(c for c in mcp_transport if isinstance(c, dict))
    assert call["name"] == "create_note"
    assert call["arguments"] == {"title": "T", "content": "C"}


def test_digest_send_to_obsigate(client, fake_gitea, monkeypatch, mcp_transport):
    monkeypatch.setenv("KANBANTE_OBSIGATE_MCP_URL", "http://obsigate.test/mcp")
    monkeypatch.setenv("KANBANTE_OBSIGATE_MCP_TOOL", "create_note")
    token = csrf_from(client)
    r = client.post("/digest/send", data={"target": "obsigate", "csrf_token": token})
    assert r.status_code == 200
    assert "✅ Digest envoyé (obsigate)" in r.text
    assert models.get_kv("digest_last_result")


def test_digest_send_requires_configuration(client, fake_gitea):
    token = csrf_from(client)
    r = client.post("/digest/send", data={"target": "obsigate", "csrf_token": token})
    assert r.status_code == 200
    assert "❌ ObsiGate incomplet" in r.text


def test_digest_probe_lists_tools(client, fake_gitea, monkeypatch, mcp_transport):
    monkeypatch.setenv("KANBANTE_OBSIGATE_MCP_URL", "http://obsigate.test/mcp")
    token = csrf_from(client)
    r = client.post("/digest/probe", data={"csrf_token": token})
    assert r.status_code == 200
    assert "create_note" in r.text