"""Tests for GET /api/runs/{run_id}/messages and GET /api/runs/{run_id}/feedback endpoints.""" from __future__ import annotations from unittest.mock import AsyncMock, MagicMock import pytest from _router_auth_helpers import make_authed_test_app from _run_message_pagination_helpers import assert_run_message_page from fastapi.testclient import TestClient from app.gateway.routers import runs # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_app(run_store=None, event_store=None, feedback_repo=None): """Build a test FastAPI app with stub auth and mocked state.""" app = make_authed_test_app() app.include_router(runs.router) if run_store is not None: app.state.run_store = run_store if event_store is not None: app.state.run_event_store = event_store if feedback_repo is not None: app.state.feedback_repo = feedback_repo return app def _make_run_store(run_record: dict | None): """Return an AsyncMock run store whose get() returns run_record.""" store = MagicMock() store.get = AsyncMock(return_value=run_record) return store def _make_event_store(rows: list[dict]): """Return an AsyncMock event store whose list_messages_by_run() returns rows.""" store = MagicMock() store.list_messages_by_run = AsyncMock(return_value=rows) return store def _make_message(seq: int) -> dict: return {"seq": seq, "event_type": "on_chat_model_stream", "category": "message", "content": f"msg-{seq}"} # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- def test_run_messages_returns_envelope(): """GET /api/runs/{run_id}/messages returns {data: [...], has_more: bool}.""" rows = [_make_message(i) for i in range(1, 4)] run_record = {"run_id": "run-1", "thread_id": "thread-1"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store(rows), ) with TestClient(app) as client: response = client.get("/api/runs/run-1/messages") assert response.status_code == 200 body = response.json() assert "data" in body assert "has_more" in body assert body["has_more"] is False assert len(body["data"]) == 3 def test_run_messages_404_when_run_not_found(): """Returns 404 when the run store returns None.""" app = _make_app( run_store=_make_run_store(None), event_store=_make_event_store([]), ) with TestClient(app) as client: response = client.get("/api/runs/missing-run/messages") assert response.status_code == 404 assert "missing-run" in response.json()["detail"] def test_run_messages_has_more_true_when_extra_row_returned(): """has_more=True when event store returns limit+1 rows.""" # Default limit is 50; provide 51 rows rows = [_make_message(i) for i in range(1, 52)] # 51 rows run_record = {"run_id": "run-2", "thread_id": "thread-2"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store(rows), ) with TestClient(app) as client: response = client.get("/api/runs/run-2/messages") assert response.status_code == 200 body = response.json() assert body["has_more"] is True assert len(body["data"]) == 50 # trimmed to limit assert [m["seq"] for m in body["data"]] == list(range(2, 52)) def test_run_messages_default_page_keeps_newest_messages_when_extra_row_returned(): """Default latest-page trimming drops the older sentinel row, not the newest message.""" rows = [_make_message(i) for i in range(16, 67)] run_record = {"run_id": "run-2", "thread_id": "thread-2"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store(rows), ) with TestClient(app) as client: assert_run_message_page(client, "/api/runs/run-2/messages", expected_seq=list(range(17, 67))) def test_run_messages_before_seq_page_keeps_newest_side_when_extra_row_returned(): """Backward pagination trims the older sentinel so adjacent pages do not miss the boundary message.""" rows = [_make_message(i) for i in range(1, 18)] run_record = {"run_id": "run-2", "thread_id": "thread-2"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store(rows), ) with TestClient(app) as client: assert_run_message_page( client, "/api/runs/run-2/messages?before_seq=18&limit=16", expected_seq=list(range(2, 18)), ) def test_run_messages_after_seq_page_keeps_oldest_side_when_extra_row_returned(): """Forward pagination still trims the newer sentinel row.""" rows = [_make_message(i) for i in range(11, 62)] run_record = {"run_id": "run-2", "thread_id": "thread-2"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store(rows), ) with TestClient(app) as client: assert_run_message_page( client, "/api/runs/run-2/messages?after_seq=10", expected_seq=list(range(11, 61)), ) def test_run_messages_passes_after_seq_to_event_store(): """after_seq query param is forwarded to event_store.list_messages_by_run.""" rows = [_make_message(10)] run_record = {"run_id": "run-3", "thread_id": "thread-3"} event_store = _make_event_store(rows) app = _make_app( run_store=_make_run_store(run_record), event_store=event_store, ) with TestClient(app) as client: response = client.get("/api/runs/run-3/messages?after_seq=5") assert response.status_code == 200 event_store.list_messages_by_run.assert_awaited_once_with( "thread-3", "run-3", limit=51, # default limit(50) + 1 before_seq=None, after_seq=5, ) def test_run_messages_respects_custom_limit(): """Custom limit is respected and capped at 200.""" rows = [_make_message(i) for i in range(1, 6)] run_record = {"run_id": "run-4", "thread_id": "thread-4"} event_store = _make_event_store(rows) app = _make_app( run_store=_make_run_store(run_record), event_store=event_store, ) with TestClient(app) as client: response = client.get("/api/runs/run-4/messages?limit=10") assert response.status_code == 200 event_store.list_messages_by_run.assert_awaited_once_with( "thread-4", "run-4", limit=11, # 10 + 1 before_seq=None, after_seq=None, ) def test_run_messages_passes_before_seq_to_event_store(): """before_seq query param is forwarded to event_store.list_messages_by_run.""" rows = [_make_message(3)] run_record = {"run_id": "run-5", "thread_id": "thread-5"} event_store = _make_event_store(rows) app = _make_app( run_store=_make_run_store(run_record), event_store=event_store, ) with TestClient(app) as client: response = client.get("/api/runs/run-5/messages?before_seq=10") assert response.status_code == 200 event_store.list_messages_by_run.assert_awaited_once_with( "thread-5", "run-5", limit=51, before_seq=10, after_seq=None, ) @pytest.mark.parametrize("cursor", ["before_seq", "after_seq"]) @pytest.mark.parametrize("value", [0, -1]) def test_run_messages_rejects_non_positive_seq_cursors(cursor: str, value: int): run_record = {"run_id": "run-6", "thread_id": "thread-6"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store([]), ) with TestClient(app) as client: response = client.get("/api/runs/run-6/messages", params={cursor: value}) assert response.status_code == 422 def test_run_messages_empty_data(): """Returns empty data list when no messages exist.""" run_record = {"run_id": "run-6", "thread_id": "thread-6"} app = _make_app( run_store=_make_run_store(run_record), event_store=_make_event_store([]), ) with TestClient(app) as client: response = client.get("/api/runs/run-6/messages") assert response.status_code == 200 body = response.json() assert body["data"] == [] assert body["has_more"] is False def _make_feedback_repo(rows: list[dict]): """Return an AsyncMock feedback repo whose list_by_run() returns rows.""" repo = MagicMock() repo.list_by_run = AsyncMock(return_value=rows) return repo def test_resolve_thread_id_handles_null_configurable(): """A client may send ``config.configurable`` as JSON ``null``. The key is then present with value ``None``, so the old ``.get("configurable", {}).get("thread_id")`` raised ``AttributeError`` (an unhandled HTTP 500). Per the docstring it should generate a new id. """ import uuid from app.gateway.routers.thread_runs import RunCreateRequest tid = runs._resolve_thread_id(RunCreateRequest(config={"configurable": None})) uuid.UUID(tid) # a freshly generated id, not a crash # working inputs are unaffected assert runs._resolve_thread_id(RunCreateRequest(config={"configurable": {"thread_id": "t1"}})) == "t1" def test_build_run_config_handles_null_configurable(): """A null ``configurable`` must also survive ``build_run_config``. ``_resolve_thread_id`` is not the only place that reads it: ``build_run_config`` does ``configurable.update(request_config.get("configurable", {}))`` and, in the ``context`` branch, ``request_config.get("configurable", {}).keys()``. With the key present and ``None``, ``.get(..., {})`` returns ``None``, so both raised (``dict.update(None)`` / ``None.keys()``) -- an unhandled HTTP 500 that the isolated ``_resolve_thread_id`` test could not catch. """ from app.gateway.services import build_run_config config = build_run_config("t1", {"configurable": None}, None) assert config["configurable"]["thread_id"] == "t1" # the context branch logs the caller's configurable keys; a null value must not crash config = build_run_config("t1", {"context": {}, "configurable": None}, None) assert config["configurable"]["thread_id"] == "t1"