diff --git a/README.rst b/README.rst index ae63650..407f1bc 100644 --- a/README.rst +++ b/README.rst @@ -232,6 +232,22 @@ E.g. for cases you want to test retrying mechanisms # this will actually perform a request resp = loop.run_until_complete(session.get('http://backend/api')) +**also you can passthrough all requests except specified by mocking object** + +.. code:: python + import asyncio + import aiohttp + from aioresponses import aioresponses + + @aioresponses(passthrough_unmatched=True) + def test_passthrough_unmatched(m, test_client): + url = 'https://httpbin.org/get' + m.get(url, status=200) + session = aiohttp.ClientSession() + # this will actually perform a request + resp = loop.run_until_complete(session.get('http://backend/api')) + # this will not perform a request and resp2.status will return 200 + resp2 = loop.run_until_complete(session.get(url)) **aioresponses allows to throw an exception** diff --git a/aioresponses/core.py b/aioresponses/core.py index 1edb7bd..a22fad6 100644 --- a/aioresponses/core.py +++ b/aioresponses/core.py @@ -225,6 +225,7 @@ class aioresponses(object): def __init__(self, **kwargs: Any): self._param = kwargs.pop('param', None) self._passthrough = kwargs.pop('passthrough', []) + self.passthrough_unmatched = kwargs.pop('passthrough_unmatched', False) self.patcher = patch('aiohttp.client.ClientSession._request', side_effect=self._request_mock, autospec=True) @@ -512,6 +513,10 @@ async def _request_mock(self, orig_self: ClientSession, response = await self.match(method, url, **kwargs) if response is None: + if self.passthrough_unmatched: + return (await self.patcher.temp_original( + orig_self, method, url_origin, *args, **kwargs + )) raise ClientConnectionError( 'Connection refused: {} {}'.format(method, url) ) diff --git a/tests/test_aioresponses.py b/tests/test_aioresponses.py index 706e89e..22b1bc4 100644 --- a/tests/test_aioresponses.py +++ b/tests/test_aioresponses.py @@ -796,3 +796,18 @@ async def test_relative_url_redirect_followed(self, rsps): self.assertEqual(str(response.url), f"{base_url}/baz") self.assertEqual(len(response.history), 1) self.assertEqual(str(response.history[0].url), url) + + async def test_pass_through_unmatched_requests(self): + matched_url = "https://matched_example.org" + unmatched_url = "https://httpbin.org/get" + params_unmatched = {'foo': 'bar'} + + with aioresponses(passthrough_unmatched=True) as m: + m.post(URL(matched_url), status=200) + mocked_response = await self.session.post(URL(matched_url)) + response = await self.session.get( + URL(unmatched_url), params=params_unmatched + ) + self.assertEqual(response.status, 200) + self.assertEqual(str(response.url), 'https://httpbin.org/get?foo=bar') + self.assertEqual(mocked_response.status, 200)