Repository: encode/httpx
File: tests/client/test_auth.py (Python)

Code before:

def test_sync_auth_reads_response_body() -> None:
    """
    Test that we can read the response body in an auth flow if `requires_response_body`
    is set.
    """
    url = "https://example.org/"
    auth = ResponseBodyAuth("xyz")
    client = Client(transport=SyncMockTransport())

    response = client.get(url, auth=auth)
    assert response.status_code == 200
    assert response.json() == {"auth": '{"auth": "xyz"}'}


@pytest.mark.asyncio
async def test_sync_async_auth() -> None:
    """
    Test that we can use a different auth flow implementation in the async case, to
    support cases that require performing I/O or using concurrency primitives (such
    as checking a disk-based cache or fetching a token from a remote auth server).
    """
    url = "https://example.org/"
    auth = SyncOrAsyncAuth()

    client = AsyncClient(transport=AsyncMockTransport())
    response = await client.get(url, auth=auth)
    assert response.status_code == 200
    assert response.json() == {"auth": "async-auth"}

    response = Client(transport=SyncMockTransport()).get(url, auth=auth)
    assert response.status_code == 200
    assert response.json() == {"auth": "sync-auth"}


Review comment (suggestion):
Let's use `async with AsyncClient(...) as client` here, since we're starting to get strict about that now.

May as well also use the same for the sync case too, just for consistency within the test case.

Code after:


@pytest.mark.asyncio
async def test_digest_auth_unavailable_streaming_body():
    url = "https://example.org/"
    auth = DigestAuth(username="tomchristie", password="password123")

    async def streaming_body():
        yield b"Example request body"  # pragma: nocover

    async with httpx.AsyncClient(transport=AsyncMockTransport()) as client:
        with pytest.raises(RequestBodyUnavailable):
            await client.post(url, data=streaming_body(), auth=auth)


@pytest.mark.asyncio
async def test_async_auth_reads_response_body() -> None:
    """
    Test that we can read the response body in an auth flow if `requires_response_body`
    is set.
    """
    url = "https://example.org/"
    auth = ResponseBodyAuth("xyz")
    async with httpx.AsyncClient(transport=AsyncMockTransport()) as client:
        response = await client.get(url, auth=auth)

    assert response.status_code == 200
    assert response.json() == {"auth": '{"auth": "xyz"}'}


def test_sync_auth_reads_response_body() -> None:
    """
    Test that we can read the response body in an auth flow if `requires_response_body`
    is set.
    """
    url = "https://example.org/"
    auth = ResponseBodyAuth("xyz")

    with httpx.Client(transport=SyncMockTransport()) as client:
        response = client.get(url, auth=auth)

    assert response.status_code == 200
    assert response.json() == {"auth": '{"auth": "xyz"}'}


@pytest.mark.asyncio
async def test_async_auth() -> None:
    """
    Test that we can use an auth implementation specific to the async case, to
    support cases that require performing I/O or using concurrency primitives (such
    as checking a disk-based cache or fetching a token from a remote auth server).