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