How to use the respx.AsyncMockTransport function in respx

To help you get started, we’ve selected a few respx examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github lundberg / respx / tests / test_transports.py View on Github external
async def test_transport_pop():
    url = "https://foo.bar/"
    alias = "get_alias"

    transport = AsyncMockTransport()
    transport.get(url, status_code=404, alias=alias)

    request_pattern = transport.pop(alias)

    assert request_pattern.response.status_code == 404
    assert request_pattern.alias == alias
    assert request_pattern.url == url

    assert not transport.aliases
    assert not transport.patterns

    with pytest.raises(KeyError):
        transport.pop(alias)

    assert transport.pop(alias, "custom default") == "custom default"
github lundberg / respx / tests / test_transports.py View on Github external
async def test_async_transport():
    url = "https://foo.bar/"

    transport = AsyncMockTransport(assert_all_called=False)
    transport.get(url, status_code=404)
    transport.get(url, content={"foo": "bar"})
    transport.post(url, pass_through=True)
    transport.put(url)

    async with httpx.AsyncClient(transport=transport) as client:
        response = await client.get(url)
        assert response.status_code == 404
        response = await client.get(url)
        assert response.status_code == 200
        assert response.json() == {"foo": "bar"}
        with pytest.raises(ValueError, match="pass_through"):
            await client.post(url)
github lundberg / respx / tests / test_transports.py View on Github external
async def test_transport_assertions():
    url = "https://foo.bar/"

    transport = AsyncMockTransport()
    transport.get(url, status_code=404)
    transport.post(url, content={"foo": "bar"})

    with pytest.raises(AssertionError, match="not called"):
        async with httpx.AsyncClient(transport=transport) as client:
            response = await client.get(url)
            assert response.status_code == 404