test(palier-3): 19 tests — recherche, mes issues, briefing, MCP, étiquettes (84 verts)
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.
This commit is contained in:
+33
-2
@@ -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")
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user