How to mock and test request with content passed as byte iterator

[adamko147]

Hello team,

I have httpx code that passes content to request using byte iterator. When trying the test code using pytest_httpx and match_content on mocked request, I get assertion error and the aiter_bytes is never called. Here's what I'm trying...

import httpx
import pytest


async def aiter_bytes():
    yield b"h"
    yield b"e"
    yield b"l"
    yield b"l"
    yield b"o"


@pytest.mark.asyncio
async def test_httpx_get(httpx_mock):
    httpx_mock.add_response(method="PUT", url="http://put", content=b"world", match_content=b"hello")
    async with httpx.AsyncClient() as client:
        res = await client.put("http://put", content=aiter_bytes())
        assert res.text == "world"

and the exception when runnning the test

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Request('PUT', 'http://put')>

    def read(self) -> bytes:
        """
        Read and return the request content.
        """
        if not hasattr(self, "_content"):
>           assert isinstance(self.stream, typing.Iterable)
E           AssertionError

When I remove the match_content parameter, the test passes, but the aiter_bytes is never called. Is there anything I'm missing? I'm not sure whether this is pytest_httpx or httpx issue, although when I run the code without pytest in real program, it works and sends the data properly.

Thank you,
Adam

[Colin-b]

Hello,

The documentation is already explaining how to stream chunks

If something is not working in the provided example, feel free to report it.

Thanks

[adamko147]

Hello,

The documentation is already explaining how to stream chunks

If something is not working in the provided example, feel free to report it.

Thanks

Hi @Colin-b, thanks for reaching out. I'am aware of the documentation, although the docs covers the case when mocked url produces the stream, which is not what I'm trying to achieve.

Following code passes fine

async def test_httpx_put(httpx_mock):
    httpx_mock.add_response(method="PUT", url="http://put", content=b"world", match_content=b"hello")
    async with httpx.AsyncClient() as client:
        res = await client.put("http://put", content=b"hello")
        assert res.text == "world"

and checks that I call http://put with "hello" content.

when I change the way of sending the data to async iterator as follows

async def aiter_bytes():
    yield b"h"
    yield b"e"
    yield b"l"
    yield b"l"
    yield b"o"


@pytest.mark.asyncio
async def test_httpx_put_iter(httpx_mock):
    httpx_mock.add_response(method="PUT", url="http://put", content=b"world", match_content=b"hello")
    async with httpx.AsyncClient() as client:
        res = await client.put("http://put", content=aiter_bytes())
        assert res.text == "world"

the execution fails with error

self = <Request('PUT', 'http://put')>

    def read(self) -> bytes:
        """
        Read and return the request content.
        """
        if not hasattr(self, "_content"):
>           assert isinstance(self.stream, typing.Iterable)
E           AssertionError

comming from pytest_httpx/_request_matcher.py:101: in _content_match

The code should work, based on httpx documentation you can pass content as Binary content to include in the body of the request, as bytes or a byte iterator.

The content is typing.AsyncIterable in this case and therefore _RequestMatcher._content_match should use

if self.content is not None:
    return await request.aread() == self.content

to compare the content.

Should I open new issue or we can re-open this one?

Thank you!

[Colin-b]

My bad, I read this too fast. Indeed this looks like a bug !

Thanks for reporting, I'll try to find some time

[Colin-b]

Fixed in version 0.31.1 available on pypi