Environment
Using the latest respx==0.21.1 with pytest==8.0.0 and pytest-mock==3.12.0
Issue
Following the respx docs for mocking a response using both a side_effect and return_value raises a StopIteration error.
While the docs state the following as an example, the test does not actually pass.
Once the iterable is exhausted, the route will fallback and respond with the return_value, if set.
import httpx
import respx
@respx.mock
def test_stacked_responses():
respx.post("https://example.org/").mock(
side_effect=[httpx.Response(201)],
return_value=httpx.Response(200)
)
response1 = httpx.post("https://example.org/")
response2 = httpx.post("https://example.org/")
response3 = httpx.post("https://example.org/")
assert response1.status_code == 201
assert response2.status_code == 200
assert response3.status_code == 200Running the above via pytest I see:
_______________________________________________________________________________________________________________________________________________________________________________________ test_stacked_responses ________________________________________________________________________________________________________________________________________________________________________________________
@respx.mock
def test_stacked_responses():
respx.post("https://example.org/").mock(
side_effect=[httpx.Response(201)],
return_value=httpx.Response(200)
)
response1 = httpx.post("https://example.org/")
> response2 = httpx.post("https://example.org/")
tests\test_client.py:82:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv\Lib\site-packages\httpx\_api.py:319: in post
return request(
.venv\Lib\site-packages\httpx\_api.py:106: in request
return client.request(
.venv\Lib\site-packages\httpx\_client.py:827: in request
return self.send(request, auth=auth, follow_redirects=follow_redirects)
.venv\Lib\site-packages\httpx\_client.py:914: in send
response = self._send_handling_auth(
.venv\Lib\site-packages\httpx\_client.py:942: in _send_handling_auth
response = self._send_handling_redirects(
.venv\Lib\site-packages\httpx\_client.py:979: in _send_handling_redirects
response = self._send_single_request(request)
.venv\Lib\site-packages\httpx\_client.py:1015: in _send_single_request
response = transport.handle_request(request)
.venv\Lib\site-packages\httpx\_transports\default.py:233: in handle_request
resp = self._pool.handle_request(req)
.venv\Lib\site-packages\respx\mocks.py:181: in mock
response = cls._send_sync_request(
.venv\Lib\site-packages\respx\mocks.py:212: in _send_sync_request
httpx_response = cls.handler(httpx_request)
.venv\Lib\site-packages\respx\mocks.py:113: in handler
httpx_response = router.handler(httpx_request)
.venv\Lib\site-packages\respx\router.py:313: in handler
resolved = self.resolve(request)
.venv\Lib\site-packages\respx\router.py:279: in resolve
prospect = route.match(request)
.venv\Lib\site-packages\respx\models.py:426: in match
result = self.resolve(request, **context)
.venv\Lib\site-packages\respx\models.py:391: in resolve
result = self._resolve_side_effect(request, **kwargs)
.venv\Lib\site-packages\respx\models.py:362: in _resolve_side_effect
effect = self._next_side_effect()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Route <Scheme eq 'https'> AND <Host eq 'example.org'> AND <Path eq '/'> AND <Method eq 'POST'>>
def _next_side_effect(
self,
) -> Union[CallableSideEffect, Exception, Type[Exception], httpx.Response]:
assert self._side_effect is not None
effect: Union[CallableSideEffect, Exception, Type[Exception], httpx.Response]
if isinstance(self._side_effect, Iterator):
> effect = next(self._side_effect)
E StopIteration
.venv\Lib\site-packages\respx\models.py:323: StopIteration
The above exception was the direct cause of the following exception:
cls = <class '_pytest.runner.CallInfo'>, func = <function call_runtest_hook.<locals>.<lambda> at 0x000002207332DC60>, when = 'call', reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>)
@classmethod
def from_call(
cls,
func: Callable[[], TResult],
when: Literal["collect", "setup", "call", "teardown"],
reraise: Optional[
Union[Type[BaseException], Tuple[Type[BaseException], ...]]
] = None,
) -> "CallInfo[TResult]":
"""Call func, wrapping the result in a CallInfo.
:param func:
The function to call. Called without arguments.
:param when:
The phase in which the function is called.
:param reraise:
Exception or exceptions that shall propagate if raised by the
function, instead of being wrapped in the CallInfo.
"""
excinfo = None
start = timing.time()
precise_start = timing.perf_counter()
try:
> result: Optional[TResult] = func()
.venv\Lib\site-packages\_pytest\runner.py:345:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv\Lib\site-packages\_pytest\runner.py:266: in <lambda>
lambda: ihook(item=item, **kwds), when=when, reraise=reraise
.venv\Lib\site-packages\pluggy\_hooks.py:501: in __call__
return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
.venv\Lib\site-packages\pluggy\_manager.py:119: in _hookexec
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
.venv\Lib\site-packages\_pytest\threadexception.py:87: in pytest_runtest_call
yield from thread_exception_runtest_hook()
.venv\Lib\site-packages\_pytest\threadexception.py:63: in thread_exception_runtest_hook
yield
.venv\Lib\site-packages\_pytest\unraisableexception.py:90: in pytest_runtest_call
yield from unraisable_exception_runtest_hook()
.venv\Lib\site-packages\_pytest\unraisableexception.py:65: in unraisable_exception_runtest_hook
yield
.venv\Lib\site-packages\_pytest\logging.py:839: in pytest_runtest_call
yield from self._runtest_for(item, "call")
.venv\Lib\site-packages\_pytest\logging.py:822: in _runtest_for
yield
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <CaptureManager _method='fd' _global_capturing=<MultiCapture out=<FDCapture 1 oldfd=9 _state='suspended' tmpfile=<_io...._io.TextIOWrapper name='nul' mode='r' encoding='utf-8'>> _state='suspended' _in_suspended=False> _capture_fixture=None>, item = <Function test_stacked_responses>
@hookimpl(wrapper=True)
def pytest_runtest_call(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("call", item):
> return (yield)
E RuntimeError: generator raised StopIteration
.venv\Lib\site-packages\_pytest\capture.py:882: RuntimeErrorExpectation
I would expect respx to catch this StopIteration error when a return_value is specified and return that value instead of propagating the error.
Environment
Using the latest
respx==0.21.1withpytest==8.0.0andpytest-mock==3.12.0Issue
Following the
respxdocs for mocking a response using both aside_effectandreturn_valueraises aStopIterationerror.While the docs state the following as an example, the test does not actually pass.
Running the above via pytest I see:
_______________________________________________________________________________________________________________________________________________________________________________________ test_stacked_responses ________________________________________________________________________________________________________________________________________________________________________________________ @respx.mock def test_stacked_responses(): respx.post("https://example.org/").mock( side_effect=[httpx.Response(201)], return_value=httpx.Response(200) ) response1 = httpx.post("https://example.org/") > response2 = httpx.post("https://example.org/") tests\test_client.py:82: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .venv\Lib\site-packages\httpx\_api.py:319: in post return request( .venv\Lib\site-packages\httpx\_api.py:106: in request return client.request( .venv\Lib\site-packages\httpx\_client.py:827: in request return self.send(request, auth=auth, follow_redirects=follow_redirects) .venv\Lib\site-packages\httpx\_client.py:914: in send response = self._send_handling_auth( .venv\Lib\site-packages\httpx\_client.py:942: in _send_handling_auth response = self._send_handling_redirects( .venv\Lib\site-packages\httpx\_client.py:979: in _send_handling_redirects response = self._send_single_request(request) .venv\Lib\site-packages\httpx\_client.py:1015: in _send_single_request response = transport.handle_request(request) .venv\Lib\site-packages\httpx\_transports\default.py:233: in handle_request resp = self._pool.handle_request(req) .venv\Lib\site-packages\respx\mocks.py:181: in mock response = cls._send_sync_request( .venv\Lib\site-packages\respx\mocks.py:212: in _send_sync_request httpx_response = cls.handler(httpx_request) .venv\Lib\site-packages\respx\mocks.py:113: in handler httpx_response = router.handler(httpx_request) .venv\Lib\site-packages\respx\router.py:313: in handler resolved = self.resolve(request) .venv\Lib\site-packages\respx\router.py:279: in resolve prospect = route.match(request) .venv\Lib\site-packages\respx\models.py:426: in match result = self.resolve(request, **context) .venv\Lib\site-packages\respx\models.py:391: in resolve result = self._resolve_side_effect(request, **kwargs) .venv\Lib\site-packages\respx\models.py:362: in _resolve_side_effect effect = self._next_side_effect() _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <Route <Scheme eq 'https'> AND <Host eq 'example.org'> AND <Path eq '/'> AND <Method eq 'POST'>> def _next_side_effect( self, ) -> Union[CallableSideEffect, Exception, Type[Exception], httpx.Response]: assert self._side_effect is not None effect: Union[CallableSideEffect, Exception, Type[Exception], httpx.Response] if isinstance(self._side_effect, Iterator): > effect = next(self._side_effect) E StopIteration .venv\Lib\site-packages\respx\models.py:323: StopIteration The above exception was the direct cause of the following exception: cls = <class '_pytest.runner.CallInfo'>, func = <function call_runtest_hook.<locals>.<lambda> at 0x000002207332DC60>, when = 'call', reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>) @classmethod def from_call( cls, func: Callable[[], TResult], when: Literal["collect", "setup", "call", "teardown"], reraise: Optional[ Union[Type[BaseException], Tuple[Type[BaseException], ...]] ] = None, ) -> "CallInfo[TResult]": """Call func, wrapping the result in a CallInfo. :param func: The function to call. Called without arguments. :param when: The phase in which the function is called. :param reraise: Exception or exceptions that shall propagate if raised by the function, instead of being wrapped in the CallInfo. """ excinfo = None start = timing.time() precise_start = timing.perf_counter() try: > result: Optional[TResult] = func() .venv\Lib\site-packages\_pytest\runner.py:345: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .venv\Lib\site-packages\_pytest\runner.py:266: in <lambda> lambda: ihook(item=item, **kwds), when=when, reraise=reraise .venv\Lib\site-packages\pluggy\_hooks.py:501: in __call__ return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult) .venv\Lib\site-packages\pluggy\_manager.py:119: in _hookexec return self._inner_hookexec(hook_name, methods, kwargs, firstresult) .venv\Lib\site-packages\_pytest\threadexception.py:87: in pytest_runtest_call yield from thread_exception_runtest_hook() .venv\Lib\site-packages\_pytest\threadexception.py:63: in thread_exception_runtest_hook yield .venv\Lib\site-packages\_pytest\unraisableexception.py:90: in pytest_runtest_call yield from unraisable_exception_runtest_hook() .venv\Lib\site-packages\_pytest\unraisableexception.py:65: in unraisable_exception_runtest_hook yield .venv\Lib\site-packages\_pytest\logging.py:839: in pytest_runtest_call yield from self._runtest_for(item, "call") .venv\Lib\site-packages\_pytest\logging.py:822: in _runtest_for yield _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <CaptureManager _method='fd' _global_capturing=<MultiCapture out=<FDCapture 1 oldfd=9 _state='suspended' tmpfile=<_io...._io.TextIOWrapper name='nul' mode='r' encoding='utf-8'>> _state='suspended' _in_suspended=False> _capture_fixture=None>, item = <Function test_stacked_responses> @hookimpl(wrapper=True) def pytest_runtest_call(self, item: Item) -> Generator[None, None, None]: with self.item_capture("call", item): > return (yield) E RuntimeError: generator raised StopIteration .venv\Lib\site-packages\_pytest\capture.py:882: RuntimeErrorExpectation
I would expect
respxto catch thisStopIterationerror when areturn_valueis specified and return that value instead of propagating the error.