bruno
9ea0cddf3c
Deux causes indépendantes faisaient échouer le job test de la CI (et seulement en CI : ni .env ni les mêmes ordres de chargement en local) : - test_agent_web_tools.mock_http patche http_client.shared_client pendant sa fenêtre d'exécution. Le premier import d'importers/url_fetch dans cette fenêtre fige la factory moquée dans le namespace du module → tous les appels fetch_url suivants du processus passaient par le handler mocké de l'autre test (« assert '…/post' == '…/page' » dans test_v56_import). url_fetch résout désormais le client à l'appel, et le mock_http restaure la vraie factory (référence figée au chargement du module) y compris sur url_fetch ; - test_agent.py posait RATE_LIMIT_ENABLED=false en env var, sans effet sur le singleton Settings déjà instancié : sous pytest -n auto, le worker dépassait le quota de 60 req/min et 8 tests recevaient des 429 (KeyError 'id' au passage). Le fixture désactive maintenant le limiter sur le singleton, comme les autres fichiers de tests.
1203 lines
49 KiB
Python
1203 lines
49 KiB
Python
"""FlowDeck — Agent tests (v4.10.0).
|
|
|
|
Covers the DB schema, LLM client (offline mock), permission manager, tool
|
|
registry (writes + undo snapshots + rollback), the ReAct engine (SSE events)
|
|
and the /api/agent/* router (agents, conversations, run, audit, skills).
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from conftest import login_test_client
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
db_path = db_file.name
|
|
db_file.close()
|
|
|
|
os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
|
|
os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
|
|
os.environ["RATE_LIMIT_ENABLED"] = "false"
|
|
os.environ["LLM_PROVIDER"] = "offline" # deterministic, no network
|
|
|
|
from app.config import settings
|
|
from app.db import get_conn, init_db
|
|
from app.main import app
|
|
from app.password_utils import hash_password
|
|
|
|
# Ensure settings singleton uses OUR temp DB (not one set by another test file).
|
|
settings.database_url = f"sqlite:///{db_path}"
|
|
settings.llm_provider = "offline"
|
|
settings.agent_max_iterations = 12
|
|
settings.agent_max_tokens_budget = 500_000
|
|
settings.agent_run_timeout_seconds = 30
|
|
# Le rate limiter est DÉJÀ instancié (singleton importé avant ce fixture) :
|
|
# `os.environ["RATE_LIMIT_ENABLED"]="false"]` ci-dessus ne le touche pas.
|
|
# Sans cette mutation, les tests suivants d'un même worker dépassent le
|
|
# quota (60 req/min) et reçoivent des 429 — visible sous `pytest -n auto`
|
|
# en CI, invisible en local (.env de dev désactivé le limiter).
|
|
settings.rate_limit_enabled = False
|
|
|
|
init_db()
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO users (login, full_name, email, password_hash, is_admin) "
|
|
"VALUES ('admin', 'Admin', '', ?, 1)",
|
|
(hash_password("test"),),
|
|
)
|
|
conn.commit()
|
|
|
|
yield login_test_client(TestClient(app))
|
|
|
|
try:
|
|
os.unlink(db_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _admin_id() -> int:
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT id FROM users WHERE login='admin'").fetchone()
|
|
return row["id"] if row else 1
|
|
|
|
|
|
def _collect(events):
|
|
return [e for e in events]
|
|
|
|
|
|
async def _run_engine(engine, *args, **kwargs):
|
|
return [e async for e in engine.run(*args, **kwargs)]
|
|
|
|
|
|
# ── LLM client (offline mock) ──
|
|
|
|
def test_llm_client_offline_mock(client):
|
|
from app.services.llm_client import LLMClient
|
|
llm = LLMClient(provider="offline")
|
|
resp = asyncio.run(llm.complete(
|
|
[{"role": "user", "content": "create collection Projets"}], tools=[],
|
|
))
|
|
assert resp.tool_calls, "offline mock should detect a create-collection intent"
|
|
assert resp.tool_calls[0]["name"] == "create_collection"
|
|
assert resp.tool_calls[0]["arguments"]["name"] == "Projets"
|
|
|
|
|
|
def test_llm_client_providers_listed(client):
|
|
from app.services.llm_client import PROVIDERS
|
|
for name in ("openai", "anthropic", "google", "deepseek", "qwencloud",
|
|
"nvidia", "openrouter", "ollama", "offline"):
|
|
assert name in PROVIDERS
|
|
|
|
|
|
def test_mistral_defaults_are_tier_safe(client):
|
|
"""La clé Mistral plan basique → 403 « tier_not_allowed » sur mistral-large/pixtral-large.
|
|
Le modèle par défaut et le premier de la liste du test de connexion doivent
|
|
être des modèles servis par tous les plans."""
|
|
from app.services.llm_client import PROVIDER_MODELS, PROVIDERS
|
|
tier_blocked = {"mistral-large-latest", "pixtral-large-latest"}
|
|
assert PROVIDERS["mistral"][1] not in tier_blocked
|
|
assert PROVIDER_MODELS["mistral"][0] not in tier_blocked
|
|
# le fallback du modèle retiré (PROVIDERS default) est présent dans la liste
|
|
assert PROVIDERS["mistral"][1] in PROVIDER_MODELS["mistral"]
|
|
# et la liste des modèles Mistral est validée chat à chaque fetch (403 exclus)
|
|
from app.services.llm_config import _CHAT_VALIDATED_PROVIDERS
|
|
assert "mistral" in _CHAT_VALIDATED_PROVIDERS
|
|
|
|
|
|
def test_llm_client_openai_compatible_bases(client):
|
|
"""Providers that need an OpenAI-compatible surface must point at it.
|
|
|
|
Anthropic/Google/Cohere native APIs are not chat-completions endpoints:
|
|
the client always speaks `/chat/completions` with a Bearer token, so the
|
|
base URLs must target the compatibility surfaces (validated live).
|
|
"""
|
|
from app.services.llm_client import PROVIDERS
|
|
assert PROVIDERS["cohere"][0] == "https://api.cohere.ai/compatibility/v1"
|
|
assert PROVIDERS["google"][0] == (
|
|
"https://generativelanguage.googleapis.com/v1beta/openai")
|
|
# Perplexity serves /chat/completions at the host root (no /v1 prefix).
|
|
assert PROVIDERS["perplexity"][0] == "https://api.perplexity.ai"
|
|
# Chutes' OpenAI-compatible inference host.
|
|
assert PROVIDERS["chutes"][0] == "https://llm.chutes.ai/v1"
|
|
assert PROVIDERS["sensenova"][0].endswith("/compatible-mode/v1")
|
|
|
|
|
|
def test_user_llm_key_api_base_normalization(client):
|
|
"""A stored api_base equal to the provider default is not persisted as an
|
|
override, so a corrected provider URL always takes effect."""
|
|
from app.services.llm_config import get_user_llm_key, upsert_user_llm_key
|
|
|
|
uid = _admin_id()
|
|
# default base → stored as empty (no override)
|
|
upsert_user_llm_key(uid, "mistral", api_key="sk-test",
|
|
api_base="https://api.mistral.ai/v1")
|
|
assert get_user_llm_key(uid, "mistral")["api_base"] == ""
|
|
# custom base → kept
|
|
upsert_user_llm_key(uid, "mistral", api_base="https://proxy.example.com/v1")
|
|
assert get_user_llm_key(uid, "mistral")["api_base"] == "https://proxy.example.com/v1"
|
|
# api_base="" resets the override, api_base=None keeps it
|
|
upsert_user_llm_key(uid, "mistral", api_base="")
|
|
assert get_user_llm_key(uid, "mistral")["api_base"] == ""
|
|
upsert_user_llm_key(uid, "mistral", api_base="https://proxy.example.com/v1")
|
|
upsert_user_llm_key(uid, "mistral", default_model="mistral-large-latest")
|
|
assert get_user_llm_key(uid, "mistral")["api_base"] == "https://proxy.example.com/v1"
|
|
|
|
|
|
def test_migration_clears_stale_llm_api_bases(client):
|
|
"""Migration 14 clears legacy/incorrect api_base values so the corrected
|
|
provider default applies (e.g. Mistral `/v2` → `/v1`)."""
|
|
from app.db import get_conn
|
|
from app.migrations import _migration_fix_llm_api_bases
|
|
from app.services.llm_config import get_user_llm_key, upsert_user_llm_key
|
|
|
|
uid = _admin_id()
|
|
upsert_user_llm_key(uid, "mistral", api_key="sk-test")
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"UPDATE user_llm_keys SET api_base='https://api.mistral.ai/v2' "
|
|
"WHERE user_id=? AND provider='mistral'",
|
|
(uid,),
|
|
)
|
|
conn.commit()
|
|
assert get_user_llm_key(uid, "mistral")["api_base"] == "https://api.mistral.ai/v2"
|
|
with get_conn() as conn:
|
|
_migration_fix_llm_api_bases(conn)
|
|
conn.commit()
|
|
assert get_user_llm_key(uid, "mistral")["api_base"] == ""
|
|
|
|
|
|
def test_llm_client_offline_no_credentials(client):
|
|
from app.services.llm_client import LLMClient
|
|
llm = LLMClient(provider="openai", api_key="")
|
|
resp = asyncio.run(llm.complete(
|
|
[{"role": "user", "content": "hello"}], tools=[],
|
|
))
|
|
# No key → falls back to mock (a final text answer, no crash)
|
|
assert resp.text != "" or resp.tool_calls
|
|
|
|
|
|
def test_llm_client_offline_mock_creates_document(client):
|
|
"""A document creation request must resolve to the create_document tool."""
|
|
from app.services.llm_client import LLMClient
|
|
llm = LLMClient(provider="offline")
|
|
resp = asyncio.run(llm.complete(
|
|
[{"role": "user", "content": 'Faire la création d\'un document "Projet" '
|
|
'dans le workspace "TEST WS 1"'}],
|
|
tools=[],
|
|
))
|
|
assert resp.tool_calls
|
|
assert resp.tool_calls[0]["name"] == "create_document"
|
|
args = resp.tool_calls[0]["arguments"]
|
|
assert args["title"] == "Projet"
|
|
assert args["workspace_name"] == "TEST WS 1"
|
|
|
|
|
|
# ── DB schema ──
|
|
|
|
def test_agent_tables_exist(client):
|
|
tables = {"agents", "agent_conversations", "agent_messages",
|
|
"agent_actions", "agent_skills", "agent_triggers",
|
|
"agent_feedback"}
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()
|
|
names = {r["name"] for r in rows}
|
|
assert tables.issubset(names)
|
|
|
|
|
|
# ── Tools ──
|
|
|
|
def test_tool_create_collection_and_search(client):
|
|
from app.services.tool_registry import ToolRegistry
|
|
reg = ToolRegistry()
|
|
res = asyncio.run(reg.execute("create_collection", {"name": "Projets"}))
|
|
assert res.status == "success"
|
|
cid = res.data["collection_id"]
|
|
assert res.undo == {"action": "delete", "table": "collections", "id": cid}
|
|
|
|
search = asyncio.run(reg.execute("search_workspace", {"query": "Projets"}))
|
|
assert search.status == "success"
|
|
assert any(c["id"] == cid for c in search.data["collections"])
|
|
|
|
|
|
def test_tool_create_page_with_properties(client):
|
|
from app.services.tool_registry import ToolRegistry
|
|
reg = ToolRegistry()
|
|
coll = asyncio.run(reg.execute("create_collection", {"name": "Tasks"}))
|
|
cid = coll.data["collection_id"]
|
|
asyncio.run(reg.execute("add_property", {"collection_id": cid, "name": "Status", "prop_type": "select", "options": ["Todo", "Done"]}))
|
|
page = asyncio.run(reg.execute("create_page", {"collection_id": cid, "title": "Task 1", "properties": {"Status": "Todo"}}))
|
|
assert page.status == "success"
|
|
assert page.target_type == "page"
|
|
assert page.data["title"] == "Task 1"
|
|
|
|
|
|
def test_tool_unknown_returns_error(client):
|
|
from app.services.tool_registry import ToolRegistry
|
|
res = asyncio.run(ToolRegistry().execute("nope", {}))
|
|
assert res.status == "error"
|
|
|
|
|
|
def test_tool_workspace_document_crud_and_search(client):
|
|
from app.db import get_conn
|
|
from app.services.tool_registry import ToolRegistry
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('TEST WS 1', ?)",
|
|
(_admin_id(),))
|
|
conn.commit()
|
|
|
|
reg = ToolRegistry()
|
|
|
|
ws = asyncio.run(reg.execute("read_workspaces", {}))
|
|
assert ws.status == "success"
|
|
assert any(w["name"] == "TEST WS 1" for w in ws.data["workspaces"])
|
|
|
|
res = asyncio.run(reg.execute("create_document",
|
|
{"title": "Projet", "workspace_name": "TEST WS 1"}))
|
|
assert res.status == "success", res.message
|
|
pid = res.data["document_id"]
|
|
assert res.target_type == "document"
|
|
assert res.undo == {"action": "delete", "table": "pages", "id": pid}
|
|
|
|
rd = asyncio.run(reg.execute("read_document", {"page_id": pid}))
|
|
assert rd.status == "success"
|
|
assert rd.data["document"]["title"] == "Projet"
|
|
|
|
ws2 = asyncio.run(reg.execute("read_workspaces", {}))
|
|
row = next(w for w in ws2.data["workspaces"] if w["name"] == "TEST WS 1")
|
|
assert row["document_count"] == 1
|
|
|
|
search = asyncio.run(reg.execute("search_workspace", {"query": "Projet"}))
|
|
assert any(d["id"] == pid for d in search.data["documents"])
|
|
|
|
# write_blocks can also rename the document
|
|
upd = asyncio.run(reg.execute("write_blocks", {"page_id": pid, "title": "Projet v2"}))
|
|
assert upd.status == "success"
|
|
rd2 = asyncio.run(reg.execute("read_document", {"page_id": pid}))
|
|
assert rd2.data["document"]["title"] == "Projet v2"
|
|
|
|
# unknown workspace name → clear error, no document created
|
|
bad = asyncio.run(reg.execute("create_document",
|
|
{"title": "X", "workspace_name": "NOPE"}))
|
|
assert bad.status == "error"
|
|
assert "introuvable" in bad.message
|
|
|
|
|
|
def test_write_blocks_normalizes_llm_blocks(client):
|
|
"""LLM blocks are normalized: `text`→`content`, `heading`→`heading_1`,
|
|
missing ids are generated, unknown types fall back to paragraph."""
|
|
from app.services.tool_registry import ToolRegistry
|
|
|
|
reg = ToolRegistry()
|
|
doc = asyncio.run(reg.execute("create_document", {"title": "Script"}))
|
|
pid = doc.data["document_id"]
|
|
|
|
bad_blocks = [
|
|
{"type": "heading", "text": "test.ps1 — Script de test PowerShell"},
|
|
{"type": "paragraph", "text": "# Test script\nWrite-Host 'Hello, World!'"},
|
|
{"type": "code", "text": "Write-Host 'hi'", "language": "powershell"},
|
|
{"type": "made_up_type", "content": "inconnu"},
|
|
]
|
|
res = asyncio.run(reg.execute("write_blocks", {"page_id": pid, "blocks": bad_blocks}))
|
|
assert res.status == "success", res.message
|
|
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT content, content_format FROM pages WHERE id=?", (pid,)).fetchone()
|
|
assert row["content_format"] == "blocks"
|
|
stored = json.loads(row["content"])
|
|
|
|
assert stored[0]["type"] == "heading_1", stored[0]
|
|
assert stored[0]["content"] == "test.ps1 — Script de test PowerShell"
|
|
assert stored[0]["id"], "id must be generated"
|
|
|
|
assert stored[1]["type"] == "paragraph"
|
|
assert stored[1]["content"] == "# Test script\nWrite-Host 'Hello, World!'"
|
|
|
|
assert stored[2]["type"] == "code"
|
|
assert stored[2]["content"] == "Write-Host 'hi'"
|
|
assert stored[2]["language"] == "powershell"
|
|
|
|
assert stored[3]["type"] == "paragraph" # unknown → paragraph
|
|
assert stored[3]["content"] == "inconnu"
|
|
|
|
|
|
def test_engine_titles_conversation_on_each_request(client):
|
|
"""The conversation title is refreshed after EVERY AI request (not only
|
|
while the title is a default placeholder)."""
|
|
from app.services.agent_engine import AgentEngine
|
|
|
|
engine = AgentEngine(_admin_id(), workspace_id=None)
|
|
conv = client.post("/api/agent/conversations",
|
|
json={"title": "Nouvelle conversation"}).json()["id"]
|
|
first = "crée un document « Rapport Q3 »"
|
|
asyncio.run(_run_engine(engine, conv, first))
|
|
detail = client.get(f"/api/agent/conversations/{conv}").json()
|
|
title1 = detail["conversation"]["title"]
|
|
assert "Rapport Q3" in title1
|
|
|
|
second = "crée une collection « Launches »"
|
|
asyncio.run(_run_engine(engine, conv, second))
|
|
detail = client.get(f"/api/agent/conversations/{conv}").json()
|
|
title2 = detail["conversation"]["title"]
|
|
assert "Launches" in title2, title2
|
|
assert title2 != title1, "title must be updated on the second request too"
|
|
|
|
|
|
def test_suggest_title_strips_composer_prefixes(client):
|
|
from app.services.agent_engine import AgentEngine
|
|
title = AgentEngine._suggest_title(
|
|
"Contexte « test.ps1 »\n\nÉcrire un script de test en powershell dans la page courante",
|
|
None,
|
|
)
|
|
assert title != "Contexte", title
|
|
assert "Contexte" not in title
|
|
assert title
|
|
|
|
|
|
def test_tool_delete_document_and_undo(client):
|
|
from app.db import get_conn
|
|
from app.services.agent_engine import undo_action
|
|
from app.services.tool_registry import ToolRegistry
|
|
|
|
reg = ToolRegistry()
|
|
res = asyncio.run(reg.execute("create_document", {"title": "À supprimer"}))
|
|
pid = res.data["document_id"]
|
|
|
|
dele = asyncio.run(reg.execute("delete_document", {"page_id": pid}))
|
|
assert dele.status == "success"
|
|
assert dele.undo == {"action": "softdelete", "table": "pages", "id": pid}
|
|
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT deleted_at FROM pages WHERE id=?", (pid,)).fetchone()
|
|
assert row is not None and row["deleted_at"] is not None
|
|
|
|
# store an audit row so undo_action can restore the document
|
|
conv_id = client.post("/api/agent/conversations", json={"title": "t"}).json()["id"]
|
|
with get_conn() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO agent_actions (conversation_id, tool_name, target_type,
|
|
target_id, payload_json, result_json, status, undo_snapshot_json, executed_by)
|
|
VALUES (?, 'delete_document', 'document', ?, '{}', '{}', 'success', ?, ?)""",
|
|
(conv_id, str(pid), json.dumps(dele.undo), _admin_id()),
|
|
)
|
|
conn.commit()
|
|
aid = cur.lastrowid
|
|
|
|
assert undo_action(aid) is True
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT deleted_at FROM pages WHERE id=?", (pid,)).fetchone()
|
|
assert row["deleted_at"] is None
|
|
|
|
|
|
# ── Permissions ──
|
|
|
|
def test_permission_manager_roles(client):
|
|
from app.db import get_conn
|
|
from app.services.permission_manager import PermissionManager
|
|
with get_conn() as conn:
|
|
owner = conn.execute("SELECT id FROM users WHERE login='admin'").fetchone()
|
|
conn.execute("INSERT INTO users (login, is_admin) VALUES ('viewer_user', 0)")
|
|
viewer = conn.execute("SELECT id FROM users WHERE login='viewer_user'").fetchone()
|
|
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('ws', ?)", (owner["id"],))
|
|
ws = conn.execute("SELECT id FROM workspaces WHERE name='ws'").fetchone()
|
|
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,?)", (ws["id"], viewer["id"], "viewer"))
|
|
conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,?)", (ws["id"], owner["id"], "owner"))
|
|
conn.commit()
|
|
|
|
pm_owner = PermissionManager(owner["id"])
|
|
pm_viewer = PermissionManager(viewer["id"])
|
|
assert pm_owner.can_write(ws["id"]) is True
|
|
assert pm_viewer.can_write(ws["id"]) is False
|
|
|
|
from fastapi import HTTPException
|
|
# viewer cannot write
|
|
with pytest.raises(HTTPException):
|
|
pm_viewer.assert_can("create_page", {}, ws["id"])
|
|
# owner can write
|
|
pm_owner.assert_can("create_page", {}, ws["id"])
|
|
# destructive requires confirm mode
|
|
with pytest.raises(HTTPException):
|
|
pm_owner.assert_can("delete_page", {}, ws["id"], approval_mode="auto")
|
|
pm_owner.assert_can("delete_page", {}, ws["id"], approval_mode="confirm")
|
|
|
|
|
|
# ── Engine (ReAct + audit + rollback) ──
|
|
|
|
def test_engine_run_creates_collection(client):
|
|
from app.db import get_conn
|
|
from app.services.agent_engine import AgentEngine
|
|
|
|
engine = AgentEngine(_admin_id(), workspace_id=None)
|
|
events = asyncio.run(_run_engine(engine, _make_conversation(client), "crée une collection Stats"))
|
|
|
|
types = [e["type"] for e in events]
|
|
assert "reasoning" in types
|
|
assert "action" in types
|
|
assert "final" in types
|
|
|
|
actions = [e for e in events if e["type"] == "action"]
|
|
assert any(a["tool"] == "create_collection" and a["status"] == "success" for a in actions)
|
|
|
|
# Collection actually persisted
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT id FROM collections WHERE name='Stats'").fetchone()
|
|
assert row is not None
|
|
|
|
|
|
def test_engine_audit_and_undo(client):
|
|
from app.db import get_conn
|
|
from app.services.agent_engine import AgentEngine, undo_action
|
|
|
|
conv = _make_conversation(client)
|
|
engine = AgentEngine(_admin_id())
|
|
asyncio.run(_run_engine(engine, conv, "créer une collection CRM"))
|
|
|
|
with get_conn() as conn:
|
|
coll = conn.execute("SELECT id FROM collections WHERE name='CRM'").fetchone()
|
|
assert coll is not None
|
|
action = conn.execute(
|
|
"SELECT id FROM agent_actions WHERE tool_name='create_collection' AND target_id=?",
|
|
(str(coll["id"]),),
|
|
).fetchone()
|
|
assert action is not None
|
|
|
|
# undo removes the collection and marks the action reverted
|
|
undo_action(action["id"])
|
|
from app.db import get_conn as gc
|
|
with gc() as conn:
|
|
assert conn.execute("SELECT id FROM collections WHERE name='CRM'").fetchone() is None
|
|
a = conn.execute("SELECT status FROM agent_actions WHERE id=?", (action["id"],)).fetchone()
|
|
assert a["status"] == "reverted"
|
|
|
|
|
|
def test_engine_run_creates_document_in_workspace_and_titles_conversation(client):
|
|
from app.db import get_conn
|
|
from app.services.agent_engine import AgentEngine
|
|
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('TEST WS 1', ?)",
|
|
(_admin_id(),))
|
|
conn.commit()
|
|
|
|
engine = AgentEngine(_admin_id(), workspace_id=None)
|
|
conv = client.post("/api/agent/conversations",
|
|
json={"title": "Nouvelle conversation"}).json()["id"]
|
|
objective = 'Faire la création d\'un document « Projet » dans le workspace « TEST WS 1 »'
|
|
events = asyncio.run(_run_engine(engine, conv, objective))
|
|
|
|
actions = [e for e in events if e["type"] == "action"]
|
|
assert any(a["tool"] == "create_document" and a["status"] == "success" for a in actions), actions
|
|
|
|
with get_conn() as conn:
|
|
doc = conn.execute("SELECT * FROM pages WHERE title='Projet'").fetchone()
|
|
ws = conn.execute("SELECT id FROM workspaces WHERE name='TEST WS 1'").fetchone()
|
|
assert doc is not None
|
|
assert doc["workspace_id"] == ws["id"]
|
|
|
|
# the conversation is renamed with a descriptive title
|
|
detail = client.get(f"/api/agent/conversations/{conv}").json()
|
|
title = detail["conversation"]["title"]
|
|
assert title.lower() != "nouvelle conversation"
|
|
assert "Projet" in title
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, text="", tool_calls=None):
|
|
self.text = text
|
|
self.tool_calls = tool_calls or []
|
|
self.usage = {}
|
|
|
|
|
|
def test_engine_tool_protocol_messages(client):
|
|
"""Le message assistant qui annonce un outil porte ses `tool_calls`, et le
|
|
résultat d'outil est renvoyé avec le `tool_call_id` correspondant — sans quoi
|
|
le fournisseur refuse la passe suivante (et l'agent retombait sur le mock)."""
|
|
from app.services.agent_engine import AgentEngine
|
|
|
|
class FakeLLM:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def complete(self, messages, *, model=None, tools=None, stream=False):
|
|
self.calls.append([dict(m) for m in messages])
|
|
if len(self.calls) == 1:
|
|
return _Resp("Plan de création.",
|
|
[{"id": "call_abc", "name": "create_collection",
|
|
"arguments": {"name": "ProtoCol"}}])
|
|
return _Resp("Collection ProtoCol créée.", [])
|
|
|
|
fake = FakeLLM()
|
|
engine = AgentEngine(_admin_id(), llm=fake)
|
|
conv = client.post("/api/agent/conversations",
|
|
json={"title": "protocol"}).json()["id"]
|
|
events = asyncio.run(_run_engine(engine, conv, "crée la collection ProtoCol"))
|
|
|
|
finals = [e for e in events if e["type"] == "final"]
|
|
assert any("ProtoCol" in e.get("content", "") for e in finals), events
|
|
|
|
# la 2e passe reçoit un bloc assistant `tool_calls` + un résultat d'outil lié
|
|
call2 = fake.calls[1]
|
|
asm = next((m for m in call2 if m["role"] == "assistant" and m.get("tool_calls")), None)
|
|
tmsg = next((m for m in call2 if m["role"] == "tool"), None)
|
|
assert asm is not None and tmsg is not None
|
|
assert asm["tool_calls"][0]["id"] == "call_abc"
|
|
assert tmsg["tool_call_id"] == asm["tool_calls"][0]["id"]
|
|
payload = json.loads(tmsg["content"])
|
|
assert payload["status"] == "ok"
|
|
|
|
|
|
def test_engine_multiple_skills_applied(client):
|
|
"""Plusieurs skills (skill_ids) sont injectés ensemble dans les instructions."""
|
|
from app.db import get_conn
|
|
from app.services.agent_engine import AgentEngine
|
|
|
|
with get_conn() as conn:
|
|
c1 = conn.execute(
|
|
"INSERT INTO agent_skills (workspace_id, name, description, prompt_template, allowed_tools_json, created_by) "
|
|
"VALUES (NULL, 'Skill A', '', 'Template A', '[]', ?)", (_admin_id(),))
|
|
c2 = conn.execute(
|
|
"INSERT INTO agent_skills (workspace_id, name, description, prompt_template, allowed_tools_json, created_by) "
|
|
"VALUES (NULL, 'Skill B', '', 'Template B', '[]', ?)", (_admin_id(),))
|
|
conn.commit()
|
|
id1, id2 = c1.lastrowid, c2.lastrowid
|
|
|
|
class FakeLLM:
|
|
def __init__(self):
|
|
self.system = ""
|
|
|
|
async def complete(self, messages, *, model=None, tools=None, stream=False):
|
|
self.system = next((m["content"] for m in messages if m["role"] == "system"), "")
|
|
return _Resp("ok", [])
|
|
|
|
fake = FakeLLM()
|
|
engine = AgentEngine(_admin_id(), llm=fake)
|
|
conv = client.post("/api/agent/conversations", json={"title": "multi-skill"}).json()["id"]
|
|
asyncio.run(_run_engine(engine, conv, "bonjour", skill_ids=[id1, id2]))
|
|
|
|
assert "Skill A" in fake.system and "Template A" in fake.system
|
|
assert "Skill B" in fake.system and "Template B" in fake.system
|
|
|
|
|
|
def _make_conversation(client) -> int:
|
|
resp = client.post("/api/agent/conversations", json={"title": "test"})
|
|
return resp.json()["id"]
|
|
# ── Router: agents, conversations, run, skills ──
|
|
|
|
def test_router_list_agents(client):
|
|
resp = client.get("/api/agent")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["agents"]
|
|
assert any(a["agent_type"] == "personal" for a in resp.json()["agents"])
|
|
|
|
|
|
def test_router_create_custom_agent(client):
|
|
resp = client.post("/api/agent", json={"name": "Reporting", "agent_type": "custom",
|
|
"system_instructions": "Produce weekly report"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["id"]
|
|
# list now includes it
|
|
agents = client.get("/api/agent").json()["agents"]
|
|
assert any(a["name"] == "Reporting" for a in agents)
|
|
|
|
|
|
def test_router_conversation_crud(client):
|
|
c = client.post("/api/agent/conversations", json={"title": "Hello"}).json()
|
|
cid = c["id"]
|
|
detail = client.get(f"/api/agent/conversations/{cid}")
|
|
assert detail.status_code == 200
|
|
assert detail.json()["conversation"]["id"] == cid
|
|
assert client.delete(f"/api/agent/conversations/{cid}").status_code == 200
|
|
assert client.get(f"/api/agent/conversations/{cid}").status_code == 404
|
|
|
|
|
|
def test_router_run_sse_stream(client):
|
|
cid = _make_conversation(client)
|
|
resp = client.post(f"/api/agent/conversations/{cid}/run",
|
|
json={"message": "create collection Roadmap"})
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"].startswith("text/event-stream")
|
|
body = resp.text
|
|
assert "data:" in body
|
|
events = [_parse_sse(line) for line in body.strip().split("\n\n") if line.strip()]
|
|
types = [e.get("type") for e in events if e]
|
|
assert "final" in types
|
|
|
|
|
|
def test_router_run_malformed(client):
|
|
cid = _make_conversation(client)
|
|
resp = client.post(f"/api/agent/conversations/{cid}/run", json={})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_router_actions_and_undo(client):
|
|
cid = _make_conversation(client)
|
|
client.post(f"/api/agent/conversations/{cid}/run", json={"message": "créer une collection Docs"})
|
|
actions = client.get(f"/api/agent/conversations/{cid}/actions").json()["actions"]
|
|
assert actions, "run should have logged actions"
|
|
target = actions[0]
|
|
assert target["tool_name"] == "create_collection"
|
|
# undo via endpoint
|
|
r = client.post(f"/api/agent/actions/{target['id']}/undo")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "reverted"
|
|
|
|
|
|
def test_router_skills(client):
|
|
s = client.post("/api/agent/skills", json={"name": "Préparation de sprint",
|
|
"prompt_template": "Analyse les issues et crée un sprint",
|
|
"allowed_tools": ["read_gitea_issues", "create_collection"]})
|
|
assert s.status_code == 200
|
|
sid = s.json()["id"]
|
|
skills = client.get("/api/agent/skills").json()["skills"]
|
|
assert any(k["name"] == "Préparation de sprint" for k in skills)
|
|
# apply → creates a conversation ready to run
|
|
apply = client.post(f"/api/agent/skills/{sid}/apply")
|
|
assert apply.status_code == 200
|
|
assert apply.json()["conversation_id"]
|
|
|
|
|
|
def test_router_mentions(client):
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format) "
|
|
"VALUES ('', 'Compte rendu projet', '## Résumé\nDécision prise.', 'markdown')")
|
|
doc_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("INSERT INTO collections (name, icon) VALUES ('Roadmap', '🗺️')")
|
|
coll_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO collection_pages (collection_id, title, property_values_json) "
|
|
"VALUES (?, 'Page Analyse', '{}')", (coll_id,))
|
|
page_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.commit()
|
|
|
|
r = client.get("/api/agent/mentions")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
sections = data["sections"]
|
|
assert isinstance(sections, list) and sections
|
|
tokens = {it["token"] for grp in sections for it in grp["items"]}
|
|
assert f"document:{doc_id}" in tokens
|
|
assert f"collection:{coll_id}" in tokens
|
|
assert f"page:{page_id}" in tokens
|
|
|
|
# search filter
|
|
r2 = client.get("/api/agent/mentions", params={"q": "Compte rendu"})
|
|
found = [it for grp in r2.json()["sections"] for it in grp["items"]]
|
|
assert found and all(it["type"] == "document" and it["label"] == "Compte rendu projet" for it in found)
|
|
|
|
# déduplication : deux fichiers avec le même titre n'apparaissent qu'une fois
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format) "
|
|
"VALUES ('', 'Compte rendu projet', 'Bis', 'markdown')")
|
|
conn.commit()
|
|
r3 = client.get("/api/agent/mentions")
|
|
dup_docs = [it for grp in r3.json()["sections"] for it in grp["items"]
|
|
if it["type"] == "document" and it["label"] == "Compte rendu projet"]
|
|
assert len(dup_docs) == 1, dup_docs
|
|
|
|
# scoping : les fichiers de l'espace courant ouvert sont proposés en section dédiée
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('Espace Scope', ?)", (_admin_id(),))
|
|
ws_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO pages (workspace, workspace_id, title, content, content_format) "
|
|
"VALUES ('Espace Scope', ?, 'Fichier du ws', '', 'markdown')", (ws_id,))
|
|
conn.commit()
|
|
r4 = client.get("/api/agent/mentions", params={"workspace_id": ws_id})
|
|
secs = r4.json()["sections"]
|
|
assert secs[0]["key"] == "files" and secs[0]["label"] == "Espace Scope"
|
|
file_tokens = {it["token"] for it in secs[0]["items"]}
|
|
assert any(t.startswith("document:") and (r4.json()["workspace_id"] == ws_id) for t in file_tokens)
|
|
|
|
|
|
def test_router_feedback(client):
|
|
cid = _make_conversation(client)
|
|
ok = client.post("/api/agent/feedback", json={
|
|
"rating": "up", "conversation_id": cid, "content": "Merci, très utile",
|
|
})
|
|
assert ok.status_code == 200
|
|
assert ok.json()["status"] == "recorded"
|
|
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
rows = conn.execute("SELECT rating, snippet FROM agent_feedback").fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0]["rating"] == "up"
|
|
assert rows[0]["snippet"] == "Merci, très utile"
|
|
|
|
bad = client.post("/api/agent/feedback", json={"rating": "meh"})
|
|
assert bad.status_code == 400
|
|
|
|
|
|
def test_context_builder_document_mention(client):
|
|
import json as _json
|
|
|
|
from app.db import get_conn
|
|
from app.services.context_builder import ContextBuilder
|
|
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format) "
|
|
"VALUES ('ws', 'Note brainstorming', 'Contenu ultra secret du doc.', 'markdown')")
|
|
mid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO pages (workspace, title, content, content_format) "
|
|
"VALUES ('ws', 'Doc blocs', ?, 'blocks')",
|
|
(_json.dumps([
|
|
{"id": "b1", "type": "heading", "content": "Titre du bloc"},
|
|
{"id": "b2", "type": "paragraph", "content": "Paragraphe du bloc.",
|
|
"children": [{"id": "b3", "type": "paragraph", "content": "Enfant aussi."}]},
|
|
]),))
|
|
bid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.commit()
|
|
|
|
ctx = ContextBuilder(_admin_id()).build(mentions=[f"document:{mid}"])
|
|
assert "Note brainstorming" in ctx
|
|
assert "Contenu ultra secret du doc." in ctx
|
|
|
|
ctx2 = ContextBuilder(_admin_id()).build(mentions=[f"document:{bid}"])
|
|
assert "Titre du bloc" in ctx2
|
|
assert "Paragraphe du bloc." in ctx2
|
|
assert "Enfant aussi." in ctx2
|
|
|
|
|
|
def test_router_tools_and_providers(client):
|
|
tools = client.get("/api/agent/tools").json()["tools"]
|
|
assert len(tools) >= 10
|
|
names = {t["name"] for t in tools}
|
|
for expected in ("search_workspace", "create_collection", "create_page",
|
|
"update_page", "add_property", "create_view", "read_page",
|
|
"read_workspaces", "read_document", "create_document"):
|
|
assert expected in names
|
|
|
|
prov = client.get("/api/agent/providers").json()
|
|
assert prov["provider"] == "offline"
|
|
assert prov["available"] is False
|
|
assert "providers" in prov
|
|
|
|
|
|
def test_router_providers_enriched(client):
|
|
prov = client.get("/api/agent/providers").json()
|
|
assert "providers" in prov
|
|
ids = [p["id"] for p in prov["providers"]]
|
|
for expected in ("openai", "anthropic", "google", "ollama", "offline"):
|
|
assert expected in ids
|
|
openai = next(p for p in prov["providers"] if p["id"] == "openai")
|
|
assert openai["default_model"] == "gpt-4o"
|
|
assert openai["models"]
|
|
assert openai["requires_key"] is True
|
|
# offline requires no key and exposes no models
|
|
offline = next(p for p in prov["providers"] if p["id"] == "offline")
|
|
assert offline["requires_key"] is False
|
|
|
|
|
|
def test_router_patch_providers_updates_config(client):
|
|
from app.auth.session import SessionManager
|
|
from app.db import get_conn
|
|
|
|
with get_conn() as conn:
|
|
admin = conn.execute("SELECT * FROM users WHERE login='admin'").fetchone()
|
|
sess = SessionManager.create_session(dict(admin))
|
|
|
|
client.cookies.set("flowdeck_session", sess)
|
|
r = client.patch("/api/agent/providers",
|
|
json={"provider": "openai", "model": "gpt-4o-mini",
|
|
"api_key": "sk-test", "api_base": ""})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["provider"] == "openai"
|
|
assert body["model"] == "gpt-4o-mini"
|
|
assert body["has_api_key"] is True
|
|
|
|
# persisted in llm_config (single row id=1)
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT * FROM llm_config WHERE id=1").fetchone()
|
|
assert row is not None
|
|
assert row["provider"] == "openai"
|
|
assert row["model"] == "gpt-4o-mini"
|
|
assert row["api_key"] == "sk-test"
|
|
|
|
# GET /providers reflects the new config
|
|
prov = client.get("/api/agent/providers").json()
|
|
assert prov["provider"] == "openai"
|
|
assert prov["model"] == "gpt-4o-mini"
|
|
assert prov["has_api_key"] is True
|
|
# a real provider with a key is marked available (no network call made)
|
|
assert prov["available"] is True
|
|
|
|
|
|
def test_router_patch_providers_requires_admin(client):
|
|
from app.auth.session import SessionManager
|
|
from app.db import get_conn
|
|
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, is_admin) VALUES ('user_basic', 'Basic', 0)")
|
|
user = conn.execute("SELECT * FROM users WHERE login='user_basic'").fetchone()
|
|
conn.commit()
|
|
sess = SessionManager.create_session(dict(user))
|
|
client.cookies.set("flowdeck_session", sess)
|
|
|
|
r = client.patch("/api/agent/providers", json={"provider": "openai"})
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_llm_client_default_model_provider_scoped(client):
|
|
# Le modèle global ne doit pas fuir vers un autre provider testé (ex. nvidia
|
|
# alors que deepseek est actif) sinon le provider reçoit un modèle inconnu
|
|
# (« model not found », HTTP 404).
|
|
from app.services.llm_client import LLMClient
|
|
from app.services.llm_config import set_llm_config
|
|
|
|
set_llm_config(provider="deepseek", model="deepseek-v4-flash",
|
|
api_key="sk-global", api_base="https://api.deepseek.com/v1")
|
|
|
|
llm = LLMClient(provider="nvidia", api_key="nvapi-xxx")
|
|
assert llm.provider == "nvidia"
|
|
assert llm.default_model == "nvidia/nemotron-3-super-120b-a12b"
|
|
assert llm.default_model != "deepseek-v4-flash"
|
|
|
|
# le provider global conserve bien son modèle
|
|
llm2 = LLMClient()
|
|
assert llm2.provider == "deepseek"
|
|
assert llm2.default_model == "deepseek-v4-flash"
|
|
|
|
|
|
def test_router_providers_test_offline(client):
|
|
r = client.post("/api/agent/providers/test", json={"provider": "offline"})
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|
|
|
|
|
|
def test_router_providers_test_missing_key(client):
|
|
# openai without a key must report failure (no mock fallback for the ping)
|
|
r = client.post("/api/agent/providers/test",
|
|
json={"provider": "openai", "api_key": ""})
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is False
|
|
assert "clé" in r.json()["error"].lower()
|
|
|
|
|
|
def test_router_run_persists_provider_model(client):
|
|
cid = _make_conversation(client)
|
|
resp = client.post(f"/api/agent/conversations/{cid}/run",
|
|
json={"message": "crée une collection Bilan",
|
|
"provider": "openai", "model": "gpt-4o-mini"})
|
|
assert resp.status_code == 200
|
|
detail = client.get(f"/api/agent/conversations/{cid}").json()
|
|
assert detail["conversation"]["provider"] == "openai"
|
|
assert detail["conversation"]["model"] == "gpt-4o-mini"
|
|
|
|
|
|
def _parse_sse(line: str) -> dict:
|
|
for part in line.split("\n"):
|
|
if part.startswith("data:"):
|
|
try:
|
|
return json.loads(part[len("data:"):].strip())
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
|
|
# ── Per-user provider keys (v4.10.2) ──
|
|
|
|
|
|
def _session_for_login(login: str) -> str:
|
|
from app.auth.session import SessionManager
|
|
from app.db import get_conn
|
|
with get_conn() as conn:
|
|
user = conn.execute("SELECT * FROM users WHERE login=?", (login,)).fetchone()
|
|
return SessionManager.create_session(dict(user))
|
|
|
|
|
|
def test_router_user_keys_crud(client):
|
|
# no session → falls back to the single admin user
|
|
r = client.put("/api/agent/keys/openai",
|
|
json={"api_key": "sk-secret-1", "api_base": "", "default_model": "gpt-4o-mini"})
|
|
assert r.status_code == 200
|
|
assert r.json()["has_key"] is True
|
|
|
|
keys = client.get("/api/agent/keys").json()["keys"]
|
|
entry = [k for k in keys if k["provider"] == "openai"][0]
|
|
assert entry["has_key"] is True
|
|
assert entry["default_model"] == "gpt-4o-mini"
|
|
assert "sk-secret-1" not in json.dumps(entry), "raw api_key must never be exposed"
|
|
|
|
# empty key keeps the stored one
|
|
r = client.put("/api/agent/keys/openai",
|
|
json={"api_key": "", "models": ["gpt-4o", "gpt-4o-custom"]})
|
|
assert r.status_code == 200
|
|
keys = client.get("/api/agent/keys").json()["keys"]
|
|
entry = [k for k in keys if k["provider"] == "openai"][0]
|
|
assert entry["has_key"] is True
|
|
assert "gpt-4o-custom" in entry["models"]
|
|
|
|
# providers endpoint merges saved models + has_key flag
|
|
prov = client.get("/api/agent/providers").json()
|
|
p = [x for x in prov["providers"] if x["id"] == "openai"][0]
|
|
assert p["has_key"] is True
|
|
assert "gpt-4o-custom" in p["models"]
|
|
|
|
r = client.delete("/api/agent/keys/openai")
|
|
assert r.status_code == 200
|
|
keys = client.get("/api/agent/keys").json()["keys"]
|
|
assert all(k["provider"] != "openai" for k in keys)
|
|
|
|
|
|
def test_router_user_keys_scoped_per_user(client):
|
|
from app.db import get_conn
|
|
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO users (login, full_name, is_admin) VALUES ('alice', 'Alice', 0)")
|
|
conn.commit()
|
|
|
|
# admin saves a key
|
|
client.put("/api/agent/keys/deepseek", json={"api_key": "sk-admin-deepseek"})
|
|
assert client.get("/api/agent/keys").json()["keys"]
|
|
|
|
# alice saves her own key and must NOT see the admin's
|
|
client.cookies.set("flowdeck_session", _session_for_login("alice"))
|
|
client.put("/api/agent/keys/deepseek", json={"api_key": "sk-alice-deepseek"})
|
|
keys = client.get("/api/agent/keys").json()["keys"]
|
|
entry = [k for k in keys if k["provider"] == "deepseek"][0]
|
|
assert entry["has_key"] is True
|
|
|
|
from app.db import get_conn as _gc
|
|
with _gc() as conn:
|
|
rows = conn.execute(
|
|
"SELECT user_id, api_key FROM user_llm_keys WHERE provider='deepseek'"
|
|
).fetchall()
|
|
by_user = {r["user_id"]: r["api_key"] for r in rows}
|
|
assert len(by_user) == 2
|
|
assert "sk-admin-deepseek" in by_user.values()
|
|
assert "sk-alice-deepseek" in by_user.values()
|
|
# masked API never leaks a raw key
|
|
assert "sk-alice-deepseek" not in json.dumps(keys)
|
|
|
|
|
|
def test_router_user_keys_invalid_provider(client):
|
|
assert client.put("/api/agent/keys/bogus", json={"api_key": "x"}).status_code == 400
|
|
assert client.post("/api/agent/keys/bogus/models").status_code == 400
|
|
assert client.delete("/api/agent/keys/bogus").status_code == 400
|
|
|
|
|
|
def test_router_fetch_models_uses_stored_key(client, monkeypatch):
|
|
from app.routers import agent as agent_router
|
|
|
|
calls = []
|
|
|
|
async def fake_fetch(provider, *, api_key="", api_base="", timeout=20):
|
|
calls.append({"provider": provider, "api_key": api_key, "api_base": api_base})
|
|
return ["stub-a", "stub-b"]
|
|
|
|
monkeypatch.setattr(agent_router, "fetch_provider_models", fake_fetch)
|
|
|
|
# no stored key yet → empty key is sent
|
|
r = client.post("/api/agent/keys/openai/models", json={})
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|
|
assert r.json()["models"] == ["stub-a", "stub-b"]
|
|
assert calls[-1]["api_key"] == ""
|
|
|
|
# stored key is used as fallback when the body carries none
|
|
client.put("/api/agent/keys/openai", json={"api_key": "sk-stored"})
|
|
r = client.post("/api/agent/keys/openai/models", json={"api_base": "https://proxy.home/v1"})
|
|
assert r.status_code == 200
|
|
assert calls[-1]["api_key"] == "sk-stored"
|
|
assert calls[-1]["api_base"] == "https://proxy.home/v1"
|
|
|
|
# body-provided key wins over the stored one
|
|
r = client.post("/api/agent/keys/openai/models", json={"api_key": "sk-typed"})
|
|
assert calls[-1]["api_key"] == "sk-typed"
|
|
|
|
|
|
def test_router_fetch_models_error_surfaced(client, monkeypatch):
|
|
from app.routers import agent as agent_router
|
|
|
|
async def fake_fetch(provider, *, api_key="", api_base="", timeout=20):
|
|
raise ConnectionError("injoignable")
|
|
|
|
monkeypatch.setattr(agent_router, "fetch_provider_models", fake_fetch)
|
|
r = client.post("/api/agent/keys/openai/models", json={"api_key": "sk-x"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is False
|
|
assert "injoignable" in body["error"]
|
|
|
|
|
|
def test_router_patch_conversation(client):
|
|
cid = _make_conversation(client)
|
|
r = client.patch(f"/api/agent/conversations/{cid}",
|
|
json={"provider": "openai", "model": "gpt-4o-mini", "title": "Sprint préparé"})
|
|
assert r.status_code == 200
|
|
detail = client.get(f"/api/agent/conversations/{cid}").json()
|
|
conv = detail["conversation"]
|
|
assert conv["provider"] == "openai"
|
|
assert conv["model"] == "gpt-4o-mini"
|
|
assert conv["title"] == "Sprint préparé"
|
|
|
|
|
|
def test_router_run_uses_user_key(client, monkeypatch):
|
|
from app.routers import agent as agent_router
|
|
|
|
created = {}
|
|
|
|
class FakeLLM(agent_router.LLMClient):
|
|
def __init__(self, *args, **kwargs):
|
|
created.update(kwargs)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
async def complete(self, messages, *, model=None, tools=None, stream=False):
|
|
return await self._mock_complete(messages, model, tools)
|
|
|
|
monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)
|
|
|
|
# user key saved → run must create the LLM client with that key
|
|
client.put("/api/agent/keys/openai", json={"api_key": "sk-user-secret", "api_base": "https://proxy.home/v1"})
|
|
cid = _make_conversation(client)
|
|
resp = client.post(f"/api/agent/conversations/{cid}/run",
|
|
json={"message": "crée une collection Bilan",
|
|
"provider": "openai", "model": "gpt-4o-mini"})
|
|
assert resp.status_code == 200
|
|
assert created.get("api_key") == "sk-user-secret"
|
|
assert created.get("api_base") == "https://proxy.home/v1"
|
|
|
|
# without a saved key the user key is not injected (global config/environment wins)
|
|
assert client.delete("/api/agent/keys/openai").status_code == 200
|
|
created.clear()
|
|
cid2 = _make_conversation(client)
|
|
resp = client.post(f"/api/agent/conversations/{cid2}/run",
|
|
json={"message": "crée une collection Bilan",
|
|
"provider": "openai", "model": "gpt-4o-mini"})
|
|
assert resp.status_code == 200
|
|
assert created.get("api_key", "") == ""
|
|
return {}
|
|
|
|
|
|
def test_router_user_key_test_verifies(client, monkeypatch):
|
|
"""A successful user-key test flags the provider as functional (verified)."""
|
|
from app.routers import agent as agent_router
|
|
from app.services.llm_client import LLMResponse
|
|
|
|
seen = {}
|
|
|
|
class FakeLLM(agent_router.LLMClient):
|
|
def __init__(self, *args, **kwargs):
|
|
seen.update(kwargs)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
async def ping(self, *, model=None):
|
|
return LLMResponse(text="PONG", model=model or "gpt-4o")
|
|
|
|
monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)
|
|
|
|
client.put("/api/agent/keys/openai", json={"api_key": "sk-user"})
|
|
r = client.post("/api/agent/keys/openai/test", json={"model": "gpt-4o-mini"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["verified"] is True
|
|
assert seen.get("api_key") == "sk-user"
|
|
|
|
# the stored key row now carries verified=True
|
|
entry = [k for k in client.get("/api/agent/keys").json()["keys"]
|
|
if k["provider"] == "openai"][0]
|
|
assert entry["verified"] is True
|
|
|
|
# /providers exposes openai as configured + verified + functional
|
|
prov = client.get("/api/agent/providers").json()
|
|
p = [x for x in prov["providers"] if x["id"] == "openai"][0]
|
|
assert p["has_key"] is True
|
|
assert p["configured"] is True
|
|
assert p["verified"] is True
|
|
assert p["functional"] is True
|
|
|
|
|
|
def test_router_user_key_test_failure(client, monkeypatch):
|
|
"""A failing test keeps the key but marks the provider non-functional."""
|
|
from app.routers import agent as agent_router
|
|
|
|
class FakeLLM(agent_router.LLMClient):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
async def ping(self, *, model=None):
|
|
raise PermissionError("Clé API manquante pour le provider « openai »")
|
|
|
|
monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)
|
|
|
|
client.put("/api/agent/keys/openai", json={"api_key": "sk-user"})
|
|
r = client.post("/api/agent/keys/openai/test", json={})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is False
|
|
assert body["verified"] is False
|
|
|
|
keys = client.get("/api/agent/keys").json()["keys"]
|
|
entry = [k for k in keys if k["provider"] == "openai"][0]
|
|
assert entry["verified"] is False
|
|
assert entry["last_error"]
|
|
|
|
prov = client.get("/api/agent/providers").json()
|
|
p = [x for x in prov["providers"] if x["id"] == "openai"][0]
|
|
assert p["functional"] is False
|
|
|
|
|
|
def test_generate_endpoint_returns_text(client, monkeypatch):
|
|
"""/api/agent/generate performs headless text generation (no tools)."""
|
|
from app.routers import agent as agent_router
|
|
from app.services.llm_client import LLMResponse
|
|
|
|
class FakeLLM(agent_router.LLMClient):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
async def complete(self, messages, *, model=None, tools=None, stream=False):
|
|
# The tool schema must never be offered for content generation.
|
|
assert tools is None
|
|
return LLMResponse(text="Voici le document traduit en anglais.", model="gpt-4o-mini")
|
|
|
|
monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)
|
|
|
|
r = client.post("/api/agent/generate", json={
|
|
"prompt": "Traduis ce document en anglais.",
|
|
"context": "Titre du document : Doc\nContenu en français.",
|
|
"model": "gpt-4o-mini",
|
|
})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert "traduit en anglais" in body["text"]
|
|
|
|
|
|
def test_generate_endpoint_requires_prompt(client):
|
|
r = client.post("/api/agent/generate", json={"prompt": " "})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_mock_planner_ignores_context_keywords(client):
|
|
"""Document context words (ex: 'cherche', 'collection') must not trigger
|
|
tool intents — the planner only inspects the user objective."""
|
|
import asyncio
|
|
|
|
from app.services.llm_client import LLMClient
|
|
|
|
llm = LLMClient(provider="offline")
|
|
resp = asyncio.run(llm.complete(
|
|
[{"role": "system", "content": "x"},
|
|
{"role": "user", "content": "Résume ce document.\n\n# Contexte\n"
|
|
"Le rapport mentionne une recherche sur la collection Z."}],
|
|
model="gpt-4o", tools=[{"type": "function"}],
|
|
))
|
|
assert resp.tool_calls == []
|
|
assert resp.text
|