(sync: bool, client: Anthropic, async_client: AsyncAnthropic)
| 64 | @pytest.mark.asyncio |
| 65 | @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) |
| 66 | async def test_multiple_events(sync: bool, client: Anthropic, async_client: AsyncAnthropic) -> None: |
| 67 | def body() -> Iterator[bytes]: |
| 68 | yield b"event: ping\n" |
| 69 | yield b"\n" |
| 70 | yield b"event: completion\n" |
| 71 | yield b"\n" |
| 72 | |
| 73 | iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client) |
| 74 | |
| 75 | sse = await iter_next(iterator) |
| 76 | assert sse.event == "ping" |
| 77 | assert sse.data == "" |
| 78 | |
| 79 | sse = await iter_next(iterator) |
| 80 | assert sse.event == "completion" |
| 81 | assert sse.data == "" |
| 82 | |
| 83 | await assert_empty_iter(iterator) |
| 84 | |
| 85 | |
| 86 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected