From b99e87ccca1d31d6d4404ee8d39e85780b27605e Mon Sep 17 00:00:00 2001 From: Spark Date: Sun, 11 Oct 2026 11:00:53 -0400 Subject: [PATCH] =?UTF-8?q?test(palier-3):=2019=20tests=20=E2=80=94=20rech?= =?UTF-8?q?erche,=20mes=20issues,=20briefing,=20MCP,=20=C3=A9tiquettes=20(?= =?UTF-8?q?84=20verts)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FakeGitea: get_user, search_issues (filtres q/état/assigné, PR exclues, repository joint), delete_label, forget_label_cache; issue 1 assignée à bruno pour /my. --- tests/conftest.py | 35 +++++- tests/test_palier3.py | 240 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 273 insertions(+), 2 deletions(-) create mode 100644 tests/test_palier3.py diff --git a/tests/conftest.py b/tests/conftest.py index 270f93c..4abbf2c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -48,8 +48,8 @@ def _issue(number: int, title: str, state: str = "open", pr: bool = False) -> di "body": f"Corps de l'issue {number} en **markdown**.", "state": state, "labels": [{"id": 900, "name": "bug", "color": "ff0000"}] if number == 1 else [], - "assignee": None, - "assignees": [], + "assignee": {"login": "bruno", "avatar_url": ""} if number == 1 else None, + "assignees": [{"login": "bruno", "avatar_url": ""}] if number == 1 else [], "milestone": {"id": 1, "title": "Sprint 1"} if number in (1, 3) else None, "due_date": None, "created_at": "2026-09-01T09:00:00Z", @@ -109,6 +109,37 @@ class FakeGitea: async def ping(self) -> bool: return self.ping_result + async def get_user(self): + return {"id": 1, "login": "bruno"} + + async def search_issues(self, q="", state="open", assigned_by=None): + results = [] + for full_name, issues in self.issues.items(): + for issue in issues: + if issue.get("pull_request"): + continue + if state != "all" and issue["state"] != state: + continue + if q and q.lower() not in issue["title"].lower(): + continue + if assigned_by: + logins = [a["login"] for a in issue.get("assignees") or []] + if issue.get("assignee"): + logins.append(issue["assignee"]["login"]) + if assigned_by not in logins: + continue + results.append({**issue, "repository": {"full_name": full_name}}) + return results + + async def delete_label(self, owner, repo, label_id): + labels = self.repo_labels.setdefault(f"{owner}/{repo}", {}) + for name in [n for n, l in labels.items() if l["id"] == label_id]: + del labels[name] + self.label_ops.append(("delete", f"{owner}/{repo}", label_id)) + + def forget_label_cache(self, owner, repo): + pass + async def search_repos(self): if self.raise_on_search: raise httpx.HTTPError("simulated Gitea outage") diff --git a/tests/test_palier3.py b/tests/test_palier3.py new file mode 100644 index 0000000..7185301 --- /dev/null +++ b/tests/test_palier3.py @@ -0,0 +1,240 @@ +"""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", + ] + + +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 # read-only server: no such tool + + +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 + assert r.json()["read_only"] is True + + +# ── 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