From 724eef5ae853f019621168071105c76962b378d4 Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Thu, 7 May 2026 15:50:22 -0700 Subject: [PATCH] test: cover list_sessions endpoint Three cases: no filter (returns all), agent_id filter (passes through to where clause), and empty result. --- .../test_endpoints_sessions.py | 47 ++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/tests/test_litellm/proxy/managed_agents_endpoints/test_endpoints_sessions.py b/tests/test_litellm/proxy/managed_agents_endpoints/test_endpoints_sessions.py index be217fa8542..1afd950c1e2 100644 --- a/tests/test_litellm/proxy/managed_agents_endpoints/test_endpoints_sessions.py +++ b/tests/test_litellm/proxy/managed_agents_endpoints/test_endpoints_sessions.py @@ -97,7 +97,7 @@ def _make_session(session_id="sess-1", **kw): return SimpleNamespace(**base) -def _make_prisma(agent=None, session=None): +def _make_prisma(agent=None, session=None, sessions=None): p = MagicMock() agent_t = MagicMock() agent_t.find_unique = AsyncMock(return_value=agent) @@ -111,6 +111,7 @@ def _make_prisma(agent=None, session=None): sess_t.create = AsyncMock() sess_t.find_unique = AsyncMock(return_value=None) sess_t.update = AsyncMock() + sess_t.find_many = AsyncMock(return_value=list(sessions) if sessions else []) p.db.litellm_managedagentsessiontable = sess_t return p @@ -456,3 +457,47 @@ def test_delete_session_no_task_arn_skips_stop(app_factory, user): resp = client.delete("/v1/managed_agents/sessions/sess-9") assert resp.status_code == 200 stop_mock.assert_not_called() + + +# --------------------------------------------------------------------------- +# list_sessions +# --------------------------------------------------------------------------- + + +def test_list_sessions_returns_all_when_no_filter(app_factory, user): + client = app_factory(user) + rows = [ + _make_session(session_id="s1", agent_id="agt-1", status="ready"), + _make_session(session_id="s2", agent_id="agt-2", status="dead"), + ] + prisma = _make_prisma(sessions=rows) + with patch("litellm.proxy.proxy_server.prisma_client", prisma): + resp = client.get("/v1/managed_agents/sessions") + assert resp.status_code == 200 + body = resp.json() + assert [r["id"] for r in body] == ["s1", "s2"] + # No filter: where is empty dict + _, kwargs = prisma.db.litellm_managedagentsessiontable.find_many.call_args + assert kwargs["where"] == {} + assert kwargs["order"] == {"created_at": "desc"} + + +def test_list_sessions_filters_by_agent_id(app_factory, user): + client = app_factory(user) + rows = [_make_session(session_id="s1", agent_id="agt-1", status="ready")] + prisma = _make_prisma(sessions=rows) + with patch("litellm.proxy.proxy_server.prisma_client", prisma): + resp = client.get("/v1/managed_agents/sessions?agent_id=agt-1") + assert resp.status_code == 200 + assert [r["agent_id"] for r in resp.json()] == ["agt-1"] + _, kwargs = prisma.db.litellm_managedagentsessiontable.find_many.call_args + assert kwargs["where"] == {"agent_id": "agt-1"} + + +def test_list_sessions_empty(app_factory, user): + client = app_factory(user) + prisma = _make_prisma(sessions=[]) + with patch("litellm.proxy.proxy_server.prisma_client", prisma): + resp = client.get("/v1/managed_agents/sessions") + assert resp.status_code == 200 + assert resp.json() == []