feat(0.26.0): écritures MCP minimales — create_issue + add_comment gardés par jeton kbt_

tests/test_mcp_writes.py

@@ -0,0 +1,202 @@
+"""Écritures MCP minimales (0.26.0) : create_issue + add_comment.
+
+Le service d'agents doit pouvoir proposer des cartes et commenter
+via Kanbante, sans passer par l'interface web. Les deux outils
+réutilisent les services existants (création via _do_capture →
+client de forge, commentaire via client.create_comment) 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
+    )
+
+
+# ── 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

tests/test_palier3.py

@@ -121,6 +121,7 @@ def test_mcp_initialize_and_tools_list(client):
     assert names == [
         "list_projects", "get_board", "get_issue", "get_focus",
         "search_issues", "get_activity", "get_metrics",
+        "create_issue", "add_comment",
     ]
 
 
@@ -153,7 +154,7 @@ def test_mcp_unknown_method_and_tool(client):
     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
+    assert result["isError"] is True  # aucun outil de ce nom (pas de suppression)
 
 
 def test_mcp_notification_gets_202(client):
@@ -164,7 +165,10 @@ def test_mcp_notification_gets_202(client):
 def test_mcp_info_endpoint(client):
     r = client.get("/mcp")
     assert r.status_code == 200
-    assert r.json()["read_only"] is True
+    # 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 ────────────────────────