(sync: bool, client: OpenAI, async_client: AsyncOpenAI)
| 146 | @pytest.mark.asyncio |
| 147 | @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) |
| 148 | async def test_multiple_data_lines(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: |
| 149 | def body() -> Iterator[bytes]: |
| 150 | yield b"event: ping\n" |
| 151 | yield b"data: {\n" |
| 152 | yield b'data: "foo":\n' |
| 153 | yield b"data: true}\n" |
| 154 | yield b"\n\n" |
| 155 | |
| 156 | iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client) |
| 157 | |
| 158 | sse = await iter_next(iterator) |
| 159 | assert sse.event == "ping" |
| 160 | assert sse.json() == {"foo": True} |
| 161 | |
| 162 | await assert_empty_iter(iterator) |
| 163 | |
| 164 | |
| 165 | @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) |
nothing calls this directly
no test coverage detected