92a8f5d894
Tests cover: - Smoke: /health, /auth/status, /distill/status (test_health.py) - Persona validation: path traversal, bad names, list_personas (test_persona.py) - Chat API: persona routing, session persistence, error handling (test_api_chat.py) - Files API: ALLOWED set enforcement, read/write, missing files (test_api_files.py) - Webhooks: NC Talk HMAC accept/reject, Google Chat JWT (test_webhooks.py) - Tools: scratch read/write/append/clear, tasks CRUD, cron parser + tools (test_tools.py) - Security: path traversal, replay attack, known gaps documented (test_security.py) All LLM calls mocked — suite runs in ~1.4s. Run: cd cortex && .venv/bin/pytest Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""
|
|
Tests for POST /chat — persona routing, session handling, LLM mocking.
|
|
"""
|
|
import json
|
|
import pytest
|
|
|
|
|
|
def _parse_sse(text: str) -> list[dict]:
|
|
"""Extract JSON payloads from an SSE response body."""
|
|
events = []
|
|
for line in text.splitlines():
|
|
if line.startswith("data: "):
|
|
try:
|
|
events.append(json.loads(line[6:]))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return events
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_chat_basic(client, mock_llm):
|
|
r = await client.post("/chat", json={"message": "Hello", "persona": "inara"})
|
|
assert r.status_code == 200
|
|
events = _parse_sse(r.text)
|
|
responses = [e for e in events if e.get("type") == "response"]
|
|
assert len(responses) == 1
|
|
assert responses[0]["response"] == "Hello, I am a test response."
|
|
assert responses[0]["backend"] == "claude"
|
|
assert "session_id" in responses[0]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_chat_default_persona(client, mock_llm):
|
|
"""persona defaults to 'inara' when not specified."""
|
|
r = await client.post("/chat", json={"message": "Hi"})
|
|
assert r.status_code == 200
|
|
events = _parse_sse(r.text)
|
|
assert any(e.get("type") == "response" for e in events)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_chat_unknown_persona(client, mock_llm):
|
|
r = await client.post("/chat", json={"message": "Hi", "persona": "nobody"})
|
|
assert r.status_code == 200
|
|
events = _parse_sse(r.text)
|
|
errors = [e for e in events if e.get("type") == "error"]
|
|
assert len(errors) == 1
|
|
assert "nobody" in errors[0]["message"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_chat_path_traversal_persona(client, mock_llm):
|
|
r = await client.post("/chat", json={"message": "Hi", "persona": "../../etc"})
|
|
assert r.status_code == 200
|
|
events = _parse_sse(r.text)
|
|
errors = [e for e in events if e.get("type") == "error"]
|
|
assert len(errors) == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_chat_session_persists(client, mock_llm):
|
|
"""Same session_id reuses history."""
|
|
r1 = await client.post("/chat", json={"message": "First message", "session_id": "test-sess-1"})
|
|
r2 = await client.post("/chat", json={"message": "Second message", "session_id": "test-sess-1"})
|
|
assert r1.status_code == 200
|
|
assert r2.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sessions_list(client, mock_llm):
|
|
"""After a chat, the session appears in /sessions."""
|
|
await client.post("/chat", json={"message": "Hello", "session_id": "test-list-sess"})
|
|
r = await client.get("/sessions")
|
|
assert r.status_code == 200
|
|
sessions = r.json()["sessions"]
|
|
ids = [s["session_id"] for s in sessions]
|
|
assert "test-list-sess" in ids
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_history(client, mock_llm):
|
|
await client.post("/chat", json={"message": "Hello", "session_id": "test-hist-sess"})
|
|
r = await client.get("/history/test-hist-sess")
|
|
assert r.status_code == 200
|
|
msgs = r.json()["messages"]
|
|
assert any(m["role"] == "user" and "Hello" in m["content"] for m in msgs)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_delete(client, mock_llm):
|
|
await client.post("/chat", json={"message": "Hello", "session_id": "test-del-sess"})
|
|
r = await client.delete("/sessions/test-del-sess")
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_delete_unknown(client):
|
|
r = await client.delete("/sessions/does-not-exist")
|
|
assert r.status_code == 404
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_backend_get(client):
|
|
r = await client.get("/backend")
|
|
assert r.status_code == 200
|
|
assert r.json()["primary"] in ("claude", "gemini")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_backend_set(client):
|
|
r = await client.post("/backend", json={"primary": "gemini"})
|
|
assert r.status_code == 200
|
|
assert r.json()["primary"] == "gemini"
|
|
# Reset
|
|
await client.post("/backend", json={"primary": "claude"})
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_backend_set_invalid(client):
|
|
r = await client.post("/backend", json={"primary": "gpt-4"})
|
|
assert r.status_code == 400
|