"""Integration tests for REST API routes."""
import pytest
from navi.llm.base import Message
class TestHealth:
def test_health(self, client):
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "ok"
assert "embed" in data
def test_health_embed(self, client):
response = client.get("/health/embed")
assert response.status_code == 200
data = response.json()
assert "ok" in data
class TestAgents:
def test_list_profiles(self, client):
response = client.get("/agents/profiles")
assert response.status_code == 200
data = response.json()
assert len(data) >= 2
assert any(p["id"] == "secretary" for p in data)
def test_list_tools(self, client):
response = client.get("/agents/tools")
assert response.status_code == 200
data = response.json()
assert len(data) >= 2
names = {t["name"] for t in data}
assert "test_tool" in names
class TestSessions:
async def test_create_session(self, client):
response = client.post("/sessions", json={"profile_id": "secretary"})
assert response.status_code == 201
data = response.json()
assert "session_id" in data
assert data["profile_id"] == "secretary"
def test_create_session_invalid_profile(self, client):
response = client.post("/sessions", json={"profile_id": "nonexistent"})
assert response.status_code == 404
@pytest.mark.anyio
async def test_list_sessions(self, client, make_session):
session = await make_session("secretary", [Message(role="user", content="hi")])
response = client.get("/sessions")
assert response.status_code == 200
data = response.json()
assert any(s["session_id"] == session.id for s in data)
@pytest.mark.anyio
async def test_list_sessions_paginates(self, client, make_session):
sessions = [await make_session("secretary") for _ in range(12)]
response = client.get("/sessions?limit=10&offset=0")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 10
assert data["has_more"] is True
assert data["next_offset"] == 10
response = client.get("/sessions?limit=10&offset=10")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 2
assert data["has_more"] is False
assert {s["session_id"] for s in data["items"]}.issubset({s.id for s in sessions})
@pytest.mark.anyio
async def test_list_sessions_filters_by_profile_before_pagination(self, client, make_session):
pinned = await make_session("developer")
await make_session("secretary")
for _ in range(12):
await make_session("developer")
response = client.patch(f"/sessions/{pinned.id}/pin", json={"pinned": True})
assert response.status_code == 200
response = client.get("/sessions?limit=10&offset=0&profile_id=developer")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 10
assert all(s["profile_id"] == "developer" for s in data["items"])
assert data["items"][0]["session_id"] == pinned.id
assert data["has_more"] is True
@pytest.mark.anyio
async def test_get_session(self, client, make_session):
session = await make_session("secretary")
response = client.get(f"/sessions/{session.id}")
assert response.status_code == 200
data = response.json()
assert data["session_id"] == session.id
assert data["profile_id"] == "secretary"
def test_get_session_not_found(self, client):
response = client.get("/sessions/nonexistent")
assert response.status_code == 404
@pytest.mark.anyio
async def test_pin_session(self, client, make_session):
session = await make_session("secretary")
response = client.patch(f"/sessions/{session.id}/pin", json={"pinned": True})
assert response.status_code == 200
data = response.json()
assert data["pinned"] is True
def test_pin_session_not_found(self, client):
response = client.patch("/sessions/nonexistent/pin", json={"pinned": True})
assert response.status_code == 404
@pytest.mark.anyio
async def test_get_context(self, client, make_session, mock_deps):
session = await make_session("secretary", [Message(role="user", content="hello")])
session.context.append(Message(role="user", content="hello"))
await mock_deps["session_store"].save(session)
response = client.get(f"/sessions/{session.id}/context")
assert response.status_code == 200
data = response.json()
assert data["session_id"] == session.id
assert data["message_count"] == 1
def test_get_context_not_found(self, client):
response = client.get("/sessions/nonexistent/context")
assert response.status_code == 404
@pytest.mark.anyio
async def test_get_planning(self, client, make_session):
session = await make_session("secretary")
session.planning_logs.append({"phases": {}})
response = client.get(f"/sessions/{session.id}/planning")
assert response.status_code == 200
data = response.json()
assert data["session_id"] == session.id
assert len(data["logs"]) == 1
def test_get_planning_not_found(self, client):
response = client.get("/sessions/nonexistent/planning")
assert response.status_code == 404
@pytest.mark.anyio
async def test_delete_session(self, client, make_session):
session = await make_session("secretary")
response = client.delete(f"/sessions/{session.id}")
assert response.status_code == 204
assert client.get(f"/sessions/{session.id}").status_code == 404
def test_delete_session_not_found(self, client):
response = client.delete("/sessions/nonexistent")
assert response.status_code == 404
class TestMessages:
@pytest.mark.anyio
async def test_send_message(self, client, make_session, monkeypatch):
session = await make_session("secretary")
class DummyAgent:
async def run(self, session_id, user_message, images=None):
return "Response text"
# Patch the Agent class in deps so the original get_agent() (captured by
# Depends() at import time) instantiates our dummy when called.
monkeypatch.setattr("navi.api.deps.Agent", lambda *a, **kw: DummyAgent())
response = client.post(f"/sessions/{session.id}/messages", json={"content": "hi"})
assert response.status_code == 200
data = response.json()
assert data["role"] == "assistant"
assert data["content"] == "Response text"
def test_send_message_not_found(self, client):
response = client.post("/sessions/nonexistent/messages", json={"content": "hi"})
assert response.status_code == 404
class TestAdmin:
def test_list_users(self, client, mock_deps):
mock_deps["fake_conn"].enqueue([
{"id": "u1", "email": "a@b.com", "display_name": "A", "role": "admin",
"permissions": "[]", "created_at": __import__("datetime").datetime.now(__import__("datetime").timezone.utc),
"updated_at": __import__("datetime").datetime.now(__import__("datetime").timezone.utc)}
])
response = client.get("/admin/users")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert data[0]["id"] == "u1"
@pytest.mark.anyio
async def test_get_session_detail(self, client, make_session):
session = await make_session("secretary", [Message(role="user", content="hi")])
response = client.get(f"/admin/sessions/{session.id}")
assert response.status_code == 200
data = response.json()
assert data["session_id"] == session.id
assert "messages" in data
def test_get_session_detail_not_found(self, client):
response = client.get("/admin/sessions/nonexistent")
assert response.status_code == 404
@pytest.mark.anyio
async def test_delete_session(self, client, make_session):
session = await make_session("secretary")
response = client.delete(f"/admin/sessions/{session.id}")
assert response.status_code == 204
assert client.get(f"/admin/sessions/{session.id}").status_code == 404
def test_delete_session_not_found(self, client):
response = client.delete("/admin/sessions/nonexistent")
assert response.status_code == 404
@pytest.mark.anyio
async def test_get_user_sessions(self, client, make_session):
session = await make_session("secretary")
response = client.get(f"/admin/users/{session.user_id}/sessions")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
def test_get_user_detail_not_found(self, client, mock_deps):
mock_deps["fake_conn"].enqueue(None)
response = client.get("/admin/users/nonexistent")
assert response.status_code == 404
def test_list_profiles(self, client):
response = client.get("/admin/profiles")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert any(p["id"] == "secretary" for p in data)
def test_update_profile_availability(self, client, mock_deps):
mock_deps["fake_conn"].enqueue("INSERT 0 1")
response = client.patch("/admin/profiles/secretary/availability", json={"is_admin_only": True})
assert response.status_code == 200
data = response.json()
assert data["ok"] is True
@pytest.mark.anyio
async def test_list_sessions_pagination(self, client, make_session):
sessions = [await make_session("secretary") for _ in range(5)]
response = client.get("/admin/sessions?limit=2&offset=0")
assert response.status_code == 200
data = response.json()
assert data["total"] == 5
assert len(data["items"]) == 2
assert data["limit"] == 2
assert data["offset"] == 0
@pytest.mark.anyio
async def test_list_sessions_pagination_offset(self, client, make_session):
sessions = [await make_session("secretary") for _ in range(5)]
response = client.get("/admin/sessions?limit=2&offset=4")
assert response.status_code == 200
data = response.json()
assert data["total"] == 5
assert len(data["items"]) == 1
@pytest.mark.anyio
async def test_list_sessions_search(self, client, make_session, mock_deps):
store = mock_deps["session_store"]
s1 = await make_session("secretary")
s1.name = "alpha session"
await store.save(s1)
s2 = await make_session("developer")
s2.name = "beta session"
await store.save(s2)
# search by name
response = client.get("/admin/sessions?search=alpha")
assert response.status_code == 200
data = response.json()
ids = {i["session_id"] for i in data["items"]}
assert s1.id in ids
assert s2.id not in ids
# search by profile
response = client.get("/admin/sessions?search=developer")
assert response.status_code == 200
data = response.json()
ids = {i["session_id"] for i in data["items"]}
assert s2.id in ids
@pytest.mark.anyio
async def test_list_sessions_sort(self, client, make_session, mock_deps):
store = mock_deps["session_store"]
s1 = await make_session("secretary")
s1.name = "aaa"
await store.save(s1)
s2 = await make_session("secretary")
s2.name = "zzz"
await store.save(s2)
response = client.get("/admin/sessions?sort_by=name&sort_order=asc")
assert response.status_code == 200
data = response.json()
names = [i["name"] for i in data["items"]]
assert names.index("aaa") < names.index("zzz")
response = client.get("/admin/sessions?sort_by=name&sort_order=desc")
assert response.status_code == 200
data = response.json()
names = [i["name"] for i in data["items"]]
assert names.index("zzz") < names.index("aaa")
