239 lines
8.3 KiB
Python
239 lines
8.3 KiB
Python
import json
|
|
import uuid
|
|
import pytest
|
|
from sqlalchemy import delete
|
|
import app.api.mcp_server as mcp_server_module
|
|
from app.core.config import settings
|
|
from app.api.mcp_server import (
|
|
get_todos, create_todo, update_todo, postpone_todo, delete_todo,
|
|
)
|
|
from app.api.mcp_server import (
|
|
search_notes, get_note, create_note, update_note, delete_note,
|
|
)
|
|
from app.api.mcp_server import (
|
|
get_shopping_lists, get_active_shopping_list, search_products,
|
|
create_shopping_list, add_shopping_item, check_shopping_item,
|
|
)
|
|
from app.models.todos import TodoItem
|
|
from app.models.notes import NoteItem
|
|
from app.models.shopping import ShoppingList, ListItem
|
|
|
|
pytestmark = pytest.mark.usefixtures("mcp_nullpool_session")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def cleanup_mcp_todos():
|
|
yield
|
|
# Utilise AsyncSessionLocal depuis le module mcp_server (patché par mcp_nullpool_session)
|
|
async with mcp_server_module.AsyncSessionLocal() as session:
|
|
await session.execute(delete(TodoItem).where(TodoItem.title.like("TEST_MCP_%")))
|
|
await session.execute(delete(NoteItem).where(NoteItem.title.like("TEST_MCP_%")))
|
|
await session.execute(delete(ShoppingList).where(ShoppingList.name.like("TEST_MCP_%")))
|
|
await session.commit()
|
|
|
|
|
|
async def test_get_todos_retourne_liste_json():
|
|
result = await get_todos(status="pending")
|
|
data = json.loads(result)
|
|
assert isinstance(data, list)
|
|
|
|
|
|
async def test_create_todo_outil_cree_une_tache():
|
|
result = await create_todo(title="TEST_MCP_todo_create")
|
|
data = json.loads(result)
|
|
assert data["title"] == "TEST_MCP_todo_create"
|
|
assert data["status"] == "pending"
|
|
assert data["priority"] == "medium"
|
|
# Cleanup
|
|
await delete_todo(id=str(data["id"]))
|
|
|
|
|
|
async def test_update_todo_outil():
|
|
created = json.loads(await create_todo(title="TEST_MCP_todo_update"))
|
|
result = await update_todo(id=str(created["id"]), status="done")
|
|
data = json.loads(result)
|
|
assert data["status"] == "done"
|
|
await delete_todo(id=str(created["id"]))
|
|
|
|
|
|
async def test_postpone_todo_outil():
|
|
created = json.loads(await create_todo(title="TEST_MCP_todo_postpone"))
|
|
result = await postpone_todo(id=str(created["id"]), days=3)
|
|
data = json.loads(result)
|
|
assert data["postponed_count"] == 1
|
|
await delete_todo(id=str(created["id"]))
|
|
|
|
|
|
async def test_delete_todo_outil():
|
|
created = json.loads(await create_todo(title="TEST_MCP_todo_delete"))
|
|
result = await delete_todo(id=str(created["id"]))
|
|
data = json.loads(result)
|
|
assert "deleted" in data
|
|
|
|
|
|
async def test_update_todo_id_invalide():
|
|
result = await update_todo(id="pas-un-uuid", title="x")
|
|
data = json.loads(result)
|
|
assert "error" in data
|
|
|
|
|
|
async def test_delete_todo_introuvable():
|
|
result = await delete_todo(id="00000000-0000-0000-0000-000000000000")
|
|
data = json.loads(result)
|
|
assert "error" in data
|
|
|
|
|
|
# ── NOTES ──────────────────────────────────────────────────────────────────────
|
|
|
|
async def test_search_notes_retourne_liste():
|
|
result = await search_notes(query="inexistant_xyz_abc_999")
|
|
data = json.loads(result)
|
|
assert isinstance(data, list)
|
|
assert data == []
|
|
|
|
|
|
async def test_search_notes_fts_trouve_par_mot_cle():
|
|
await create_note(title="TEST_MCP_note_fts", content="recette de cuisine française traditionnelle")
|
|
result = await search_notes(query="cuisine")
|
|
data = json.loads(result)
|
|
assert any(n["title"] == "TEST_MCP_note_fts" for n in data)
|
|
|
|
|
|
async def test_create_note_outil():
|
|
result = await create_note(title="TEST_MCP_note_create", content="Contenu de test MCP")
|
|
data = json.loads(result)
|
|
assert data["title"] == "TEST_MCP_note_create"
|
|
assert "id" in data
|
|
await delete_note(id=str(data["id"]))
|
|
|
|
|
|
async def test_get_note_outil():
|
|
created = json.loads(await create_note(title="TEST_MCP_note_get", content="Contenu get"))
|
|
result = await get_note(id=str(created["id"]))
|
|
data = json.loads(result)
|
|
assert data["title"] == "TEST_MCP_note_get"
|
|
assert data["content"] == "Contenu get"
|
|
assert "attachments" in data
|
|
await delete_note(id=str(created["id"]))
|
|
|
|
|
|
async def test_update_note_outil():
|
|
created = json.loads(await create_note(title="TEST_MCP_note_update", content="avant"))
|
|
result = await update_note(id=str(created["id"]), content="après")
|
|
data = json.loads(result)
|
|
assert "id" in data
|
|
await delete_note(id=str(created["id"]))
|
|
|
|
|
|
async def test_delete_note_outil():
|
|
created = json.loads(await create_note(title="TEST_MCP_note_delete", content="x"))
|
|
result = await delete_note(id=str(created["id"]))
|
|
data = json.loads(result)
|
|
assert "deleted" in data
|
|
|
|
|
|
async def test_get_note_introuvable():
|
|
result = await get_note(id="00000000-0000-0000-0000-000000000000")
|
|
data = json.loads(result)
|
|
assert "error" in data
|
|
|
|
|
|
# ── SHOPPING ──────────────────────────────────────────────────────────────────
|
|
|
|
async def test_get_shopping_lists_retourne_liste():
|
|
result = await get_shopping_lists()
|
|
data = json.loads(result)
|
|
assert isinstance(data, list)
|
|
|
|
|
|
async def test_create_shopping_list_outil():
|
|
result = await create_shopping_list(name="TEST_MCP_liste")
|
|
data = json.loads(result)
|
|
assert data["name"] == "TEST_MCP_liste"
|
|
assert data["status"] == "draft"
|
|
assert "id" in data
|
|
|
|
|
|
async def test_create_shopping_list_nom_auto():
|
|
result = await create_shopping_list()
|
|
data = json.loads(result)
|
|
# Le nom auto est au format "S{semaine} {année}" (ex: "S22 2026")
|
|
assert data["name"].startswith("S")
|
|
# Cleanup manuel — la liste auto n'a pas le préfixe TEST_MCP_
|
|
async with mcp_server_module.AsyncSessionLocal() as session:
|
|
await session.execute(
|
|
delete(ShoppingList).where(ShoppingList.id == uuid.UUID(data["id"]))
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def test_add_shopping_item_outil():
|
|
liste = json.loads(await create_shopping_list(name="TEST_MCP_liste_item"))
|
|
result = await add_shopping_item(
|
|
list_id=str(liste["id"]),
|
|
name="TEST_MCP_article",
|
|
quantity=2.0,
|
|
unit="kg",
|
|
)
|
|
data = json.loads(result)
|
|
assert data["name"] == "TEST_MCP_article"
|
|
assert data["is_checked"] is False
|
|
assert float(data["quantity"]) == 2.0
|
|
|
|
|
|
async def test_check_shopping_item_outil():
|
|
liste = json.loads(await create_shopping_list(name="TEST_MCP_liste_check"))
|
|
article = json.loads(await add_shopping_item(
|
|
list_id=str(liste["id"]),
|
|
name="TEST_MCP_article_check",
|
|
))
|
|
result = await check_shopping_item(
|
|
list_id=str(liste["id"]),
|
|
item_id=str(article["id"]),
|
|
)
|
|
data = json.loads(result)
|
|
assert data["is_checked"] is True
|
|
|
|
|
|
async def test_search_products_retourne_liste():
|
|
result = await search_products(q="inexistant_xyz_abc_999")
|
|
data = json.loads(result)
|
|
assert isinstance(data, list)
|
|
assert data == []
|
|
|
|
|
|
async def test_get_active_shopping_list_structure():
|
|
result = await get_active_shopping_list()
|
|
data = json.loads(result)
|
|
assert isinstance(data, dict)
|
|
|
|
|
|
async def test_add_item_liste_invalide():
|
|
result = await add_shopping_item(list_id="pas-un-uuid", name="article")
|
|
data = json.loads(result)
|
|
assert "error" in data
|
|
|
|
|
|
# ── AUTH ──────────────────────────────────────────────────────────────────────
|
|
|
|
async def test_mcp_auth_rejet_sans_token(client):
|
|
"""Le middleware renvoie 401 si aucun header Authorization."""
|
|
resp = await client.get("/mcp")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_mcp_auth_rejet_mauvais_token(client):
|
|
"""Le middleware renvoie 401 si le token est incorrect."""
|
|
resp = await client.get("/mcp", headers={"Authorization": "Bearer mauvais-token"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_mcp_auth_accepte_bon_token(client, monkeypatch):
|
|
"""Le middleware laisse passer avec le token correct."""
|
|
monkeypatch.setattr(settings, "mcp_api_key", "test-mcp-key-xyz")
|
|
resp = await client.get(
|
|
"/mcp",
|
|
headers={"Authorization": "Bearer test-mcp-key-xyz"},
|
|
)
|
|
assert resp.status_code != 401
|