pythongssapi / httpx-gssapi Goto Github PK

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'

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

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

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

    @ 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

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