Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_pass_through(client, parameters, expected):
async with MockTransport() as respx_mock:
request = respx_mock.add(**parameters)
with asynctest.mock.patch(
"asyncio.open_connection",
side_effect=ConnectionRefusedError("test request blocked"),
) as open_connection:
with pytest.raises(NetworkError):
await client.get("https://example.org/")
assert open_connection.called is True
assert request.called is True
assert request.pass_through is expected
with MockTransport() as respx_mock:
request = respx_mock.add(**parameters)
with asynctest.mock.patch(
"socket.socket.connect", side_effect=socket.error("test request blocked"),
) as connect:
with pytest.raises(NetworkError):
httpx.get("https://example.org/")
assert connect.called is True
assert request.called is True
assert request.pass_through is expected
async def test_request_callback(client):
def callback(request, response):
if request.url.host == "foo.bar":
response.headers["X-Foo"] = "bar"
response.content = lambda request, name: f"hello {name}"
response.context["name"] = "lundberg"
return response
async with MockTransport(assert_all_called=False) as respx_mock:
request = respx_mock.add(callback, status_code=202, headers={"X-Ham": "spam"})
response = await client.get("https://foo.bar/")
assert request.called is True
assert request.pass_through is None
assert response.status_code == 202
assert response.headers == httpx.Headers(
{"Content-Type": "text/plain", "X-Ham": "spam", "X-Foo": "bar"}
)
assert response.text == "hello lundberg"
with pytest.raises(ValueError):
respx_mock.add(lambda req, res: "invalid")
await client.get("https://ham.spam/")
async def test_assert_all_mocked(client, assert_all_mocked, raises):
with raises:
with MockTransport(assert_all_mocked=assert_all_mocked) as respx_mock:
response = httpx.get("https://foo.bar/")
assert respx_mock.stats.call_count == 1
assert response.status_code == 200
with raises:
async with MockTransport(assert_all_mocked=assert_all_mocked) as respx_mock:
response = await client.get("https://foo.bar/")
assert respx_mock.stats.call_count == 1
assert response.status_code == 200
assert respx_mock.stats.call_count == 0
async def test_url_match(client, url, pattern):
async with MockTransport(assert_all_mocked=False) as respx_mock:
request = respx_mock.get(pattern, content="baz")
response = await client.get(url)
assert request.called is True
assert response.status_code == 200
assert response.text == "baz"
async def test_assert_all_mocked(client, assert_all_mocked, raises):
with raises:
with MockTransport(assert_all_mocked=assert_all_mocked) as respx_mock:
response = httpx.get("https://foo.bar/")
assert respx_mock.stats.call_count == 1
assert response.status_code == 200
with raises:
async with MockTransport(assert_all_mocked=assert_all_mocked) as respx_mock:
response = await client.get("https://foo.bar/")
assert respx_mock.stats.call_count == 1
assert response.status_code == 200
assert respx_mock.stats.call_count == 0
async def test_assert_all_called(client, assert_all_called, do_post, raises):
with raises:
async with MockTransport(assert_all_called=assert_all_called) as respx_mock:
request1 = respx_mock.get("https://foo.bar/1/", status_code=404)
request2 = respx_mock.post("https://foo.bar/", status_code=201)
await client.get("https://foo.bar/1/")
if do_post:
await client.post("https://foo.bar/")
assert request1.called is True
assert request2.called is do_post