713 lines
27 KiB
Python
713 lines
27 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 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.main import app
|
|
from app.db import init_db, get_conn
|
|
from app.password_utils import hash_password
|
|
from app.config import settings
|
|
|
|
# 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
|
|
|
|
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 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_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
|
|
|
|
|
|
# ── DB schema ──
|
|
|
|
def test_agent_tables_exist(client):
|
|
tables = {"agents", "agent_conversations", "agent_messages",
|
|
"agent_actions", "agent_skills", "agent_triggers"}
|
|
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"
|
|
|
|
|
|
# ── Permissions ──
|
|
|
|
def test_permission_manager_roles(client):
|
|
from app.services.permission_manager import PermissionManager
|
|
from app.db import get_conn
|
|
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.services.agent_engine import AgentEngine
|
|
from app.db import get_conn
|
|
|
|
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.services.agent_engine import AgentEngine, undo_action
|
|
from app.db import get_conn
|
|
|
|
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 _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_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"):
|
|
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_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 |