How to use with pytest?

[adriantorrie]

Using pytest, I've tried to create a mock fixture for an app written with FastAPI/HTTPX (FastAPI is an interface microservice receiving requests, and forwarding onto a backend service using an HTTPX Async client).

The fixture and unit test

from httpx_gssapi import HTTPSPNEGOAuth
from pytest_httpx import HTTPXMock
from unittest.mock import Mock

class TestClient(object):
    @ pytest.fixture
    def test_client(self, test_settings):
        """Returns a test-configured client to use in tests"""
        test_client = MyClient(test_settings)
        test_client._auth = Mock(spec=HTTPSPNEGOAuth)
        return test_client

    @pytest.mark.asyncio
    async def test_client_get_id(self, test_app, test_client, httpx_mock: HTTPXMock):
        # Mocks
        httpx_mock.add_response(
            url="http://test/id?path=foo&selectedFields=Id",
            json={"Id": "TEST"}
        )

        async with AsyncClient(app=test_app, base_url="http://test") as ac:
            id = await test_client._get_id(ac, "foo")
            assert id == 'TEST'

MyCllient has the auth set as a member, using the following prod defaults, which I thought the mock replacing in the fixture would be enough for things to "just work".

self._auth = HTTPSPNEGOAuth(
            mutual_authentication=OPTIONAL,
            opportunistic_auth=True,
            delegate=False)

Within the function being tested, a get request is being made using a HTTPSPNEGOAuth auth object, with an httpx.AsyncClient instance

response = await client.get(url, params=params, auth=self._auth)     # Line that is failing in the test

I receive the error:

self = <httpx.AsyncClient object at 0x00000232160A6640>
request = <Request('GET', 'http://test/id?path=foo&selectedFields=Id')>
auth = <Mock spec='HTTPSPNEGOAuth' id='2414147621552'>
timeout = Timeout(timeout=5.0), allow_redirects = True, history = []

    async def _send_handling_auth(
        self,
        request: Request,
        auth: Auth,
        timeout: Timeout,
        allow_redirects: bool,
        history: typing.List[Response],
    ) -> Response:
        auth_flow = auth.async_auth_flow(request)
        try:
            request = await auth_flow.__anext__()
    
            for hook in self._event_hooks["request"]:
                await hook(request)
    
            while True:
                response = await self._send_handling_redirects(
                    request,
                    timeout=timeout,
                    allow_redirects=allow_redirects,
                    history=history,
                )
                try:
                    try:
                        next_request = await auth_flow.asend(response)
                    except StopAsyncIteration:
                        return response
    
                    response.history = list(history)
                    await response.aread()
                    request = next_request
                    history.append(response)
    
                except Exception as exc:
                    await response.aclose()
                    raise exc
        finally:
>           await auth_flow.aclose()
E           TypeError: object Mock can't be used in 'await' expression

I treid swapping the Mock for an AsyncMock in the fixture:

    @ pytest.fixture
    def test_client(self, test_settings):
        """Returns a test-configured client to use in tests"""
        test_client = MyClient(test_settings)
        test_client._auth = AsyncMock(spec=HTTPSPNEGOAuth)   # AsyncMock used
        return test_client

Which gives a similar error still:

self = <httpx.AsyncClient object at 0x000001D67640A3D0>
request = <AsyncMock name='mock.async_auth_flow().__anext__()' id='2020620739584'>
auth = <AsyncMock spec='HTTPSPNEGOAuth' id='2020617927120'>
timeout = Timeout(timeout=5.0), allow_redirects = True, history = []

    async def _send_handling_auth(
        self,
        request: Request,
        auth: Auth,
        timeout: Timeout,
        allow_redirects: bool,
        history: typing.List[Response],
    ) -> Response:
        auth_flow = auth.async_auth_flow(request)
        try:
            request = await auth_flow.__anext__()
    
            for hook in self._event_hooks["request"]:
                await hook(request)
    
            while True:
                response = await self._send_handling_redirects(
                    request,
                    timeout=timeout,
                    allow_redirects=allow_redirects,
                    history=history,
                )
                try:
                    try:
                        next_request = await auth_flow.asend(response)
                    except StopAsyncIteration:
                        return response
    
                    response.history = list(history)
                    await response.aread()
                    request = next_request
                    history.append(response)
    
                except Exception as exc:
                    await response.aclose()
                    raise exc
        finally:
>           await auth_flow.aclose()
E           TypeError: object MagicMock can't be used in 'await' expression

I'm not sure where to take it from here to get the auth object mocked out correctly for use with pytest.

[aiudirog]

Sorry for the slow response, I was traveling. The issue is AsyncMock not handling async generators properly. It is making the asend() and aclose() methods synchronous for some reason:

In [1]: from httpx_gssapi import HTTPSPNEGOAuth
In [2]: from unittest.mock import AsyncMock
In [3]: auth = AsyncMock(spec=HTTPSPNEGOAuth)
In [4]: auth_flow = auth.async_auth_flow(...)

In [5]: auth_flow
Out[5]: <MagicMock name='mock.async_auth_flow()' id='140057769280368'>

In [6]: auth_flow.__anext__()  # Correctly a coroutine
Out[6]: <coroutine object AsyncMockMixin._execute_mock_call at 0x7f61bd9731c0>

In [7]: auth_flow.asend
Out[7]: <MagicMock name='mock.async_auth_flow().asend' id='140057769692224'>

In [8]: auth_flow.asend()  # Not a coroutine
Out[8]: <MagicMock name='mock.async_auth_flow().asend()' id='140057769324208'>

In [9]: await auth_flow.asend()
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-9-4d8d419a061f> in <module>
----> 1 await auth_flow.asend()

TypeError: object MagicMock can't be used in 'await' expression

To be honest, I'm not sure how to handle this simply, without mocking out some of the internal HTTPX auth flow details themselves. A better approach may be to use K5TEST like we do in the end_to_end tests and setup a fake Kerberos environment instead of mocking the authentication.