245 lines
8.5 KiB
Python
245 lines
8.5 KiB
Python
"""Palier 3 (0.5.0): search + my issues, briefing/ntfy, MCP, labels UI API."""
|
|
|
|
import json
|
|
|
|
from app import models
|
|
from tests.conftest import csrf_from
|
|
|
|
|
|
# ── Point 7 : recherche globale + mes issues ───────────────
|
|
|
|
def test_search_page_without_query(client):
|
|
r = client.get("/search")
|
|
assert r.status_code == 200
|
|
assert "search-hero" in r.text
|
|
|
|
|
|
def test_search_finds_issues_across_repos(client):
|
|
r = client.get("/search?q=focus")
|
|
assert r.status_code == 200
|
|
assert "Tâche focus" in r.text
|
|
assert "Tâche un" not in r.text
|
|
assert "Projets/demo" in r.text
|
|
|
|
|
|
def test_search_includes_closed_with_done_chip(client):
|
|
r = client.get("/search?q=ancienne")
|
|
assert "Ancienne tâche" in r.text
|
|
assert "Terminé" in r.text # column chip for closed issues
|
|
|
|
|
|
def test_my_issues_lists_only_assigned_open(client):
|
|
r = client.get("/my")
|
|
assert r.status_code == 200
|
|
assert "bruno" in r.text
|
|
assert "Tâche un" in r.text # assigned to bruno, open
|
|
assert "Tâche focus" not in r.text # not assigned
|
|
|
|
|
|
# ── Point 8 : briefing ─────────────────────────────────────
|
|
|
|
def test_kv_store_roundtrip():
|
|
models.set_kv("k1", "v1")
|
|
assert models.get_kv("k1") == "v1"
|
|
models.set_kv("k1", "v2")
|
|
assert models.get_kv("k1") == "v2"
|
|
assert models.get_kv("absent") is None
|
|
|
|
|
|
def test_briefing_page_and_api(client):
|
|
models.move_issue("Projets/demo", 5, "in_progress")
|
|
r = client.get("/briefing")
|
|
assert r.status_code == 200
|
|
assert "briefing-text" in r.text
|
|
r = client.get("/api/briefing")
|
|
data = r.json()
|
|
assert data["ok"] is True
|
|
assert data["briefing"]["focus_total"] >= 1
|
|
assert "Kanbante" in data["text"]
|
|
|
|
|
|
def test_briefing_send_requires_ntfy_config(client):
|
|
token = csrf_from(client)
|
|
r = client.post("/api/briefing/send", data={"csrf_token": token})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_briefing_send_posts_to_ntfy(client, monkeypatch):
|
|
monkeypatch.setenv("KANBANTE_NTFY_URL", "https://ntfy.example")
|
|
monkeypatch.setenv("KANBANTE_NTFY_TOPIC", "kanbante-bruno")
|
|
captured = {}
|
|
|
|
async def fake_send(settings, text):
|
|
captured["topic"] = settings.ntfy_topic
|
|
captured["text"] = text
|
|
return True
|
|
|
|
monkeypatch.setattr("app.main.send_ntfy", fake_send)
|
|
token = csrf_from(client)
|
|
r = client.post("/api/briefing/send", data={"csrf_token": token})
|
|
assert r.status_code == 200
|
|
assert r.json()["sent"] is True
|
|
assert captured["topic"] == "kanbante-bruno"
|
|
assert "briefing" in captured["text"]
|
|
assert models.get_kv("briefing_last_sent") is not None
|
|
|
|
|
|
def test_briefing_format_is_plain_text():
|
|
from app.briefing import format_briefing_text
|
|
|
|
text = format_briefing_text({
|
|
"date": "11/10/2026",
|
|
"activity_24h": {"closed": 2, "created": 1, "moved": 5, "avg_cycle_days": 3.5},
|
|
"focus_total": 1,
|
|
"focus_groups": [{"full_name": "a/b", "count": 1, "titles": ["X"]}],
|
|
"aging": [{"repo": "a/b", "number": 4, "title": "Vieille",
|
|
"days": 12, "column_label": "À faire"}],
|
|
"milestones_due": [{"repo": "a/b", "title": "S1",
|
|
"due_on": "2026-10-15", "open": 3}],
|
|
})
|
|
assert "2 fermée(s)" in text
|
|
assert "Vieille" in text and "12 j" in text
|
|
assert "S1" in text
|
|
|
|
|
|
# ── Point 9 : MCP ──────────────────────────────────────────
|
|
|
|
def _rpc(client, method, params=None, rpc_id=1):
|
|
payload = {"jsonrpc": "2.0", "id": rpc_id, "method": method}
|
|
if params is not None:
|
|
payload["params"] = params
|
|
return client.post("/mcp", json=payload)
|
|
|
|
|
|
def test_mcp_initialize_and_tools_list(client):
|
|
r = _rpc(client, "initialize", {"protocolVersion": "2024-11-05"})
|
|
result = r.json()["result"]
|
|
assert result["serverInfo"]["name"] == "kanbante"
|
|
assert result["protocolVersion"] == "2024-11-05"
|
|
r = _rpc(client, "tools/list")
|
|
names = [t["name"] for t in r.json()["result"]["tools"]]
|
|
assert names == [
|
|
"list_projects", "get_board", "get_issue", "get_focus",
|
|
"search_issues", "get_activity", "get_metrics",
|
|
"create_issue", "add_comment",
|
|
]
|
|
|
|
|
|
def test_mcp_call_get_board_and_search(client):
|
|
r = _rpc(client, "tools/call",
|
|
{"name": "get_board", "arguments": {"owner": "Projets", "repo": "demo"}})
|
|
result = r.json()["result"]
|
|
assert result["isError"] is False
|
|
board = json.loads(result["content"][0]["text"])
|
|
assert any(c["title"] == "Tâche un" for c in board["backlog"])
|
|
|
|
r = _rpc(client, "tools/call",
|
|
{"name": "search_issues", "arguments": {"query": "focus"}})
|
|
found = json.loads(r.json()["result"]["content"][0]["text"])
|
|
assert [f["title"] for f in found] == ["Tâche focus"]
|
|
|
|
|
|
def test_mcp_call_metrics_and_activity(client):
|
|
models.log_event("Projets/demo", 1, "moved", to_column="todo")
|
|
r = _rpc(client, "tools/call", {"name": "get_activity", "arguments": {}})
|
|
events = json.loads(r.json()["result"]["content"][0]["text"])
|
|
assert events[0]["event_type"] == "moved"
|
|
r = _rpc(client, "tools/call", {"name": "get_metrics", "arguments": {}})
|
|
metrics = json.loads(r.json()["result"]["content"][0]["text"])
|
|
assert metrics["moved"] >= 1
|
|
|
|
|
|
def test_mcp_unknown_method_and_tool(client):
|
|
r = _rpc(client, "resources/list")
|
|
assert r.json()["error"]["code"] == -32601
|
|
r = _rpc(client, "tools/call", {"name": "delete_everything", "arguments": {}})
|
|
result = r.json()["result"]
|
|
assert result["isError"] is True # aucun outil de ce nom (pas de suppression)
|
|
|
|
|
|
def test_mcp_notification_gets_202(client):
|
|
r = client.post("/mcp", json={"jsonrpc": "2.0", "method": "notifications/initialized"})
|
|
assert r.status_code == 202
|
|
|
|
|
|
def test_mcp_info_endpoint(client):
|
|
r = client.get("/mcp")
|
|
assert r.status_code == 200
|
|
# 0.26.0 : écritures minimales ajoutées — le serveur n'est
|
|
# plus en lecture seule, et il le déclare honnêtement.
|
|
assert r.json()["read_only"] is False
|
|
assert r.json()["write_tools"] == ["create_issue", "add_comment"]
|
|
|
|
|
|
# ── Board : gestion des étiquettes ────────────────────────
|
|
|
|
def test_labels_api_create_list_delete(client, fake_gitea):
|
|
token = csrf_from(client)
|
|
r = client.post(
|
|
"/api/repos/Projets/demo/labels",
|
|
data={"name": "urgent", "color": "e11d48", "csrf_token": token},
|
|
)
|
|
assert r.status_code == 201
|
|
label = r.json()["label"]
|
|
assert label["name"] == "urgent"
|
|
|
|
r = client.get("/api/repos/Projets/demo/labels")
|
|
assert [l["name"] for l in r.json()["labels"]] == ["urgent"]
|
|
|
|
r = client.request(
|
|
"DELETE", f"/api/repos/Projets/demo/labels/{label['id']}",
|
|
headers={"X-CSRF-Token": token},
|
|
)
|
|
assert r.json()["ok"] is True
|
|
r = client.get("/api/repos/Projets/demo/labels")
|
|
assert r.json()["labels"] == []
|
|
|
|
|
|
def test_labels_api_validation_and_csrf(client):
|
|
r = client.post(
|
|
"/api/repos/Projets/demo/labels",
|
|
data={"name": "x", "color": "ffffff"},
|
|
)
|
|
assert r.status_code == 403 # no CSRF token
|
|
token = csrf_from(client)
|
|
r = client.post(
|
|
"/api/repos/Projets/demo/labels",
|
|
data={"name": "x", "color": "pashex", "csrf_token": token},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_issue_label_toggle(client, fake_gitea):
|
|
token = csrf_from(client)
|
|
r = client.post(
|
|
"/api/repos/Projets/demo/labels",
|
|
data={"name": "revue", "color": "0ea5e9", "csrf_token": token},
|
|
)
|
|
label_id = r.json()["label"]["id"]
|
|
r = client.post(
|
|
"/api/issues/Projets/demo/5/labels",
|
|
data={"label_id": label_id, "action": "add", "csrf_token": token},
|
|
)
|
|
assert r.status_code == 200
|
|
assert [l["name"] for l in r.json()["labels"]] == ["revue"]
|
|
r = client.post(
|
|
"/api/issues/Projets/demo/5/labels",
|
|
data={"label_id": label_id, "action": "remove", "csrf_token": token},
|
|
)
|
|
assert r.json()["labels"] == []
|
|
r = client.post(
|
|
"/api/issues/Projets/demo/5/labels",
|
|
data={"label_id": label_id, "action": "explode", "csrf_token": token},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_board_has_label_manager_and_new_notes_panel(client):
|
|
r = client.get("/board/Projets/demo")
|
|
assert 'id="labels-btn"' in r.text
|
|
assert 'id="label-modal"' in r.text
|
|
assert 'id="notes-panel"' in r.text
|
|
assert 'id="notes-resize"' in r.text
|
|
assert 'id="notes-toggle"' in r.text
|
|
assert 'id="notes-status"' in r.text
|