Spark
8ad3a86e60
CI / test (push) Failing after 1m27s
Le serveur MCP n'exposait en écriture que create_issue et add_comment : un déplacement approuvé côté service d'agents (kanbante.move_card, vague H3) aurait échoué faute d'outil. - move_issue (owner, repo, number, column) réutilise le service de déplacement de l'interface (_do_move_card, extrait de POST /api/move sans changement de comportement) : board local, journal source « mcp », miroir d'étiquette kanban:* remplacée, autres étiquettes préservées. - Colonne acceptée par clé, par étiquette de forge ou par libellé français affiché (définitions de l'espace). - Garde doctrine : la cible done (« Terminé ») est refusée avant toute écriture — les agents s'arrêtent en Révision. - Erreurs propres : colonne inconnue (liste des valides), dépôt/issue inconnu, pull request refusée. - 296 tests verts (5 nouveaux + épingles de la liste d'outils mises à jour).
330 lines
13 KiB
Python
330 lines
13 KiB
Python
"""Écritures MCP minimales : create_issue + add_comment (0.26.0),
|
|
move_issue (0.36.0).
|
|
|
|
Le service d'agents doit pouvoir proposer des cartes, commenter
|
|
et déplacer des cartes via Kanbante, sans passer par l'interface
|
|
web. Les trois outils réutilisent les services existants
|
|
(création via _do_capture → client de forge, commentaire via
|
|
client.create_comment, déplacement via _do_move_card) et sont
|
|
gardés par la même authentification que les lectures /mcp :
|
|
session, Basic ou jeton d'API kbt_ — jamais l'interface web.
|
|
"""
|
|
|
|
import json
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app import accounts, models
|
|
|
|
|
|
def _call(client, name, arguments, headers=None):
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "tools/call",
|
|
"params": {"name": name, "arguments": arguments},
|
|
}
|
|
return client.post("/mcp", json=payload, headers=headers or {})
|
|
|
|
|
|
def _result_text(response):
|
|
result = response.json()["result"]
|
|
return result, result["content"][0]["text"]
|
|
|
|
|
|
# ── Création de carte ──────────────────────────────────────
|
|
|
|
def test_mcp_create_issue_places_card(client, fake_gitea):
|
|
r = _call(client, "create_issue", {
|
|
"owner": "Projets", "repo": "demo",
|
|
"title": "Carte créée par MCP", "body": "Corps de la carte",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
created = json.loads(text)
|
|
assert created["title"] == "Carte créée par MCP"
|
|
assert created["repo_full_name"] == "Projets/demo"
|
|
assert created["column"] == "backlog"
|
|
assert created["number"] > 0
|
|
# L'écriture est passée par le client de forge existant.
|
|
assert fake_gitea.created == [("Projets", "demo", "Carte créée par MCP")]
|
|
# La carte est placée sur le board (lecture MCP relue).
|
|
r = _call(client, "get_board", {"owner": "Projets", "repo": "demo"})
|
|
board = json.loads(_result_text(r)[1])
|
|
assert any(
|
|
c["title"] == "Carte créée par MCP" for c in board["backlog"]
|
|
)
|
|
# Journal : création tracée avec la source « mcp ».
|
|
events = models.get_events(repo_full_name="Projets/demo", limit=10)
|
|
created_events = [
|
|
e for e in events
|
|
if e["event_type"] == "created"
|
|
and e["issue_number"] == created["number"]
|
|
]
|
|
assert created_events and created_events[0]["source"] == "mcp"
|
|
|
|
|
|
def test_mcp_create_issue_column_choice(client, fake_gitea):
|
|
r = _call(client, "create_issue", {
|
|
"owner": "Projets", "repo": "demo",
|
|
"title": "Directement à faire", "column": "todo",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
assert json.loads(text)["column"] == "todo"
|
|
|
|
|
|
def test_mcp_create_issue_validation_errors(client, fake_gitea):
|
|
# Titre manquant : « Missing argument » (patron d'erreur du serveur).
|
|
r = _call(client, "create_issue", {"owner": "Projets", "repo": "demo"})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "Missing argument" in text
|
|
# Titre vide et colonne inconnue : erreurs métier en isError.
|
|
for arguments in (
|
|
{"owner": "Projets", "repo": "demo", "title": " "},
|
|
{"owner": "Projets", "repo": "demo", "title": "X",
|
|
"column": "nimporte"},
|
|
):
|
|
result, _text = _result_text(_call(client, "create_issue", arguments))
|
|
assert result["isError"] is True
|
|
assert fake_gitea.created == []
|
|
|
|
|
|
# ── Commentaire ────────────────────────────────────────────
|
|
|
|
def test_mcp_add_comment_posts_on_forge(client, fake_gitea):
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"body": "Commentaire du service d'agents",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
payload = json.loads(text)
|
|
assert payload["number"] == 1
|
|
assert payload["repo_full_name"] == "Projets/demo"
|
|
assert payload["body"] == "Commentaire du service d'agents"
|
|
assert fake_gitea.posted_comments == [
|
|
("Projets", "demo", 1, "Commentaire du service d'agents")
|
|
]
|
|
events = models.get_events(repo_full_name="Projets/demo", limit=10)
|
|
commented = [e for e in events if e["event_type"] == "commented"]
|
|
assert commented and commented[0]["source"] == "mcp"
|
|
|
|
|
|
def test_mcp_add_comment_validation_errors(client, fake_gitea):
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1, "body": " ",
|
|
})
|
|
assert _result_text(r)[0]["isError"] is True
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "body": "sans numéro",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "Missing argument" in text
|
|
assert fake_gitea.posted_comments == []
|
|
|
|
|
|
def test_mcp_write_tools_are_listed(client):
|
|
r = client.post("/mcp", json={
|
|
"jsonrpc": "2.0", "id": 1, "method": "tools/list",
|
|
})
|
|
tools = {t["name"]: t for t in r.json()["result"]["tools"]}
|
|
assert tools["create_issue"]["inputSchema"]["required"] == [
|
|
"owner", "repo", "title",
|
|
]
|
|
assert tools["add_comment"]["inputSchema"]["required"] == [
|
|
"owner", "repo", "number", "body",
|
|
]
|
|
# Aucun outil destructeur n'apparaît.
|
|
assert not any(
|
|
name.startswith(("delete", "remove", "close"))
|
|
for name in tools
|
|
)
|
|
|
|
|
|
# ── Déplacement de carte (0.36.0) ──────────────────────────
|
|
|
|
def _seed_carte_todo(fake_gitea, number=1):
|
|
"""Place la carte dans « À faire », étiquette kanban:todo
|
|
miroir posée (la carte garde son étiquette « bug »)."""
|
|
etiquette = {"id": 950, "name": "kanban:todo", "color": "3b82f6"}
|
|
fake_gitea.repo_labels["Projets/demo"]["kanban:todo"] = etiquette
|
|
issue = next(
|
|
i for i in fake_gitea.issues["Projets/demo"]
|
|
if i["number"] == number
|
|
)
|
|
issue["labels"].append(etiquette)
|
|
models.move_issue("Projets/demo", number, "todo")
|
|
return issue
|
|
|
|
|
|
def test_mcp_move_issue_moves_card_and_replaces_label(client, fake_gitea):
|
|
issue = _seed_carte_todo(fake_gitea)
|
|
r = _call(client, "move_issue", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"column": "review",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
payload = json.loads(text)
|
|
assert payload["from_column"] == "todo"
|
|
assert payload["column"] == "review"
|
|
assert payload["label_sync"] == "ok"
|
|
# Étiquette kanban remplacée, les autres préservées.
|
|
noms = [l["name"] for l in issue["labels"]]
|
|
assert "kanban:review" in noms
|
|
assert "kanban:todo" not in noms
|
|
assert "bug" in noms
|
|
assert ("remove", 1, "kanban:todo") in fake_gitea.label_ops
|
|
assert ("add", 1, "kanban:review") in fake_gitea.label_ops
|
|
# Le board suit le déplacement (lecture MCP relue).
|
|
board = json.loads(_result_text(_call(
|
|
client, "get_board", {"owner": "Projets", "repo": "demo"}
|
|
))[1])
|
|
assert any(c["number"] == 1 for c in board["review"])
|
|
assert not any(c["number"] == 1 for c in board["todo"])
|
|
# Journal : déplacement tracé avec la source « mcp ».
|
|
events = models.get_events(repo_full_name="Projets/demo", limit=10)
|
|
moved = [e for e in events if e["event_type"] == "moved"]
|
|
assert moved and moved[0]["source"] == "mcp"
|
|
assert moved[0]["from_column"] == "todo"
|
|
assert moved[0]["to_column"] == "review"
|
|
|
|
|
|
def test_mcp_move_issue_accepts_french_labels(client, fake_gitea):
|
|
_seed_carte_todo(fake_gitea)
|
|
for libelle, cle in (
|
|
("En cours", "in_progress"),
|
|
("Révision", "review"),
|
|
("kanban:qa", "qa"),
|
|
("À faire", "todo"),
|
|
):
|
|
r = _call(client, "move_issue", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"column": libelle,
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False, (libelle, text)
|
|
assert json.loads(text)["column"] == cle
|
|
|
|
|
|
def test_mcp_move_issue_refuses_done_before_any_write(client, fake_gitea):
|
|
_seed_carte_todo(fake_gitea)
|
|
for libelle in ("done", "Terminé"):
|
|
r = _call(client, "move_issue", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"column": libelle,
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "Terminé" in text
|
|
# Aucune écriture : ni étiquette, ni état, ni journal.
|
|
assert fake_gitea.label_ops == []
|
|
assert fake_gitea.state_changes == []
|
|
events = models.get_events(repo_full_name="Projets/demo", limit=10)
|
|
assert not [e for e in events if e["event_type"] == "moved"]
|
|
|
|
|
|
def test_mcp_move_issue_errors(client, fake_gitea):
|
|
# Colonne inconnue : erreur propre, avec les colonnes valides.
|
|
r = _call(client, "move_issue", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"column": "nimporte",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "Colonnes valides" in text and "review" in text
|
|
# Issue inconnue, puis dépôt inconnu.
|
|
for arguments in (
|
|
{"owner": "Projets", "repo": "demo", "number": 999,
|
|
"column": "review"},
|
|
{"owner": "Projets", "repo": "inexistant", "number": 1,
|
|
"column": "review"},
|
|
):
|
|
result, text = _result_text(_call(client, "move_issue", arguments))
|
|
assert result["isError"] is True
|
|
assert "inconnu" in text
|
|
# Une pull request n'est pas une carte.
|
|
r = _call(client, "move_issue", {
|
|
"owner": "Projets", "repo": "demo", "number": 2,
|
|
"column": "review",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "pull request" in text
|
|
assert fake_gitea.label_ops == []
|
|
|
|
|
|
def test_mcp_move_issue_is_listed(client):
|
|
r = client.post("/mcp", json={
|
|
"jsonrpc": "2.0", "id": 1, "method": "tools/list",
|
|
})
|
|
tools = {t["name"]: t for t in r.json()["result"]["tools"]}
|
|
assert tools["move_issue"]["inputSchema"]["required"] == [
|
|
"owner", "repo", "number", "column",
|
|
]
|
|
info = client.get("/mcp").json()
|
|
assert "move_issue" in info["write_tools"]
|
|
|
|
|
|
# ── Garde par jeton kbt_ (auth active) ─────────────────────
|
|
|
|
@pytest.fixture()
|
|
def utilisateur():
|
|
# Base de test partagée : nom unique, nettoyage complet au
|
|
# teardown (le pont SSO a une sémantique « un seul compte »).
|
|
uid = models.create_user(
|
|
f"mcp-{uuid.uuid4().hex[:8]}",
|
|
accounts.hash_password("motdepasse"), role="member")
|
|
user = models.get_user(uid)
|
|
yield user
|
|
with models.get_db() as conn:
|
|
conn.execute("DELETE FROM api_tokens WHERE user_id = ?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id = ?", (uid,))
|
|
|
|
|
|
def test_mcp_writes_guarded_by_api_token(monkeypatch, fake_gitea, utilisateur):
|
|
monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno")
|
|
monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse")
|
|
_row, clair = models.create_api_token(utilisateur["id"], "service-agents")
|
|
jeton = {"Authorization": f"Bearer {clair}"}
|
|
create_args = {
|
|
"owner": "Projets", "repo": "demo", "title": "Carte par jeton",
|
|
}
|
|
with TestClient(main.app) as client:
|
|
# Sans jeton : refusé (401), aucune écriture sur la forge.
|
|
assert _call(client, "create_issue", create_args).status_code == 401
|
|
# Jeton inconnu : refusé aussi.
|
|
faux = {"Authorization": "Bearer kbt_inconnu"}
|
|
assert _call(
|
|
client, "create_issue", create_args, headers=faux
|
|
).status_code == 401
|
|
assert fake_gitea.created == []
|
|
# Jeton valide : création et commentaire passent.
|
|
r = _call(client, "create_issue", create_args, headers=jeton)
|
|
assert r.status_code == 200
|
|
assert _result_text(r)[0]["isError"] is False
|
|
assert fake_gitea.created == [("Projets", "demo", "Carte par jeton")]
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"body": "Vu par le service d'agents",
|
|
}, headers=jeton)
|
|
assert _result_text(r)[0]["isError"] is False
|
|
# Le même jeton n'ouvre jamais l'interface web.
|
|
assert client.get(
|
|
"/", headers={**jeton, "Accept": "application/json"}
|
|
).status_code == 401
|
|
# Révocation : effet immédiat sur les écritures.
|
|
assert models.revoke_api_token(utilisateur["id"], _row["id"]) is True
|
|
assert _call(
|
|
client, "create_issue", create_args, headers=jeton
|
|
).status_code == 401
|
|
assert _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1, "body": "x",
|
|
}, headers=jeton).status_code == 401
|