Function test_send_async_bytes_stream

tests/request_test.py:80–92

()

@pytest.mark.asyncio
@pytest.mark.flaky(reruns=3, reruns_delay=2)
async def test_send_async_bytes_stream():
    async def file_bytes_stream():
        with open("README.md", "rb") as f:
            while True:
                chunk = f.read(1024)
                if not chunk:
                    break
                yield chunk

    url = "https://httpbin.org/post"
    response = await client.post(url, body=file_bytes_stream())
    json = await response.json()
    assert json["data"] in open("README.md").read()


@pytest.mark.asyncio