We're attempting to mock HTTP responses requested from within a Python thread pool. Is that supported? The following example suggest it does not work, unfortunately:
import concurrent.futures
import httpretty as _httpretty
import pytest
import requests
@pytest.fixture(scope="session")
def httpretty():
_httpretty.enable(verbose=True, allow_net_connect=False)
try:
yield _httpretty
finally:
_httpretty.disable()
_httpretty.reset()
@pytest.fixture
def mock_response(httpretty):
def request_callback(request, uri, response_headers):
return 200, response_headers, request.body # Pass through the requested body
httpretty.register_uri(method="POST", uri="https://my-server.com/", body=request_callback)
@pytest.fixture
def example_payloads():
return [0, 1, 3]
def post(payload):
r = requests.post("https://my-server.com/", json=payload)
return r.json()
def test_simple_map_passes(mock_response, example_payloads):
results = map(post, example_payloads)
assert list(results) == example_payloads
def test_threadpool_map_fails(mock_response, example_payloads):
executor = concurrent.futures.ThreadPoolExecutor()
results = executor.map(post, example_payloads)
# Responses appear to be randomly drawn (with duplicates) from the expected list
assert list(results) == example_payloads
Interestingly, occasionally, the response body appears to be the full string like this:
We're attempting to mock HTTP responses requested from within a Python thread pool. Is that supported? The following example suggest it does not work, unfortunately:
Interestingly, occasionally, the response body appears to be the full string like this:
which is really not expected at all since it should just contain the JSON data, e.g.
0
,1
.I believe this works with Python 3.8 and 3.9, but fails from Python 3.10 onwards.