Test basic pagination for spans endpoint.
(
server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str],
)
| 951 | |
| 952 | @pytest.mark.asyncio |
| 953 | async def test_spans_pagination_basic( |
| 954 | server_client: Tuple[LightningStoreServer, LightningStoreClient, aiohttp.ClientSession, str], |
| 955 | ) -> None: |
| 956 | """Test basic pagination for spans endpoint.""" |
| 957 | server, _client, session, api_endpoint = server_client |
| 958 | |
| 959 | # Create rollout and attempt |
| 960 | rollout = await server.start_rollout(input={"test": "data"}) |
| 961 | |
| 962 | # Add multiple spans |
| 963 | for i in range(10): |
| 964 | span = _make_span(rollout.rollout_id, rollout.attempt.attempt_id, i + 1, f"span-{i}") |
| 965 | await server.add_span(span) |
| 966 | |
| 967 | # Get first page |
| 968 | async with session.get( |
| 969 | f"{api_endpoint}/spans", params={"rollout_id": rollout.rollout_id, "limit": 3, "offset": 0} |
| 970 | ) as resp: |
| 971 | assert resp.status == 200 |
| 972 | data = await resp.json() |
| 973 | assert data["total"] == 10 |
| 974 | assert data["limit"] == 3 |
| 975 | assert len(data["items"]) == 3 |
| 976 | |
| 977 | # Get all spans |
| 978 | async with session.get(f"{api_endpoint}/spans", params={"rollout_id": rollout.rollout_id, "limit": -1}) as resp: |
| 979 | assert resp.status == 200 |
| 980 | data = await resp.json() |
| 981 | assert data["total"] == 10 |
| 982 | assert len(data["items"]) == 10 |
| 983 | |
| 984 | |
| 985 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected