Merge pull request #233 from outp1/master

added passthrough_unmatched flag for aioresponses class (solve #190)

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**
 

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)
             )

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)