Merge pull request #133 from clslgrnc/akeeman_patch

releasenotes/notes/session-scoped-mock-7f1c98d9a91bffc8.yaml

@@ -0,0 +1,8 @@
+---
+features:
+  - |
+    Allow mocking of only a single Session object. While there has always been
+    the option of adding just the Adapter to an existing Session, this is more
+    difficult to work with than the Mocker workflow that is typically used. You
+    can now pass a Session object to a Mocker and it will provide the standard
+    workflow but limited to just that object.

requests_mock/mocker.py

@@ -13,6 +13,7 @@
 import functools
 
 import requests
+import six
 
 from requests_mock import adapter
 from requests_mock import exceptions
@@ -28,6 +29,18 @@
 _original_send = requests.Session.send
 
 
+def _set_method(target, name, method):
+    """ Set a mocked method onto the target.
+
+    Target may be either an instance of a Session object of the
+    requests.Session class. First we Bind the method if it's an instance.
+    """
+    if not isinstance(target, type):
+        method = six.create_bound_method(method, target)
+
+    setattr(target, name, method)
+
+
 class MockerCore(object):
     """A wrapper around common mocking functions.
 
@@ -68,7 +81,11 @@ class MockerCore(object):
     This will become the default in a 2.X release. See bug: #1584008.
     """
 
-    def __init__(self, **kwargs):
+    def __init__(self, session=None, **kwargs):
+        if session and not isinstance(session, requests.Session):
+            raise TypeError("Only a requests.Session object can be mocked")
+
+        self._mock_target = session or requests.Session
         self.case_sensitive = kwargs.pop('case_sensitive', self.case_sensitive)
         self._adapter = (
             kwargs.pop('adapter', None) or
@@ -89,15 +106,16 @@ def start(self):
         if self._last_send:
             raise RuntimeError('Mocker has already been started')
 
-        self._last_send = requests.Session.send
-        self._last_get_adapter = requests.Session.get_adapter
+        # backup last `send` for restoration on `self.stop`
+        self._last_send = self._mock_target.send
+        self._last_get_adapter = self._mock_target.get_adapter
 
         def _fake_get_adapter(session, url):
             return self._adapter
 
         def _fake_send(session, request, **kwargs):
             # mock get_adapter
-            requests.Session.get_adapter = _fake_get_adapter
+            _set_method(session, "get_adapter", _fake_get_adapter)
 
             # NOTE(jamielennox): self._last_send vs _original_send. Whilst it
             # seems like here we would use _last_send there is the possibility
@@ -120,14 +138,17 @@ def _fake_send(session, request, **kwargs):
                 pass
             finally:
                 # restore get_adapter
-                requests.Session.get_adapter = self._last_get_adapter
+                _set_method(session, "get_adapter", self._last_get_adapter)
 
             # if we are here it means we must run the real http request
             # Or, with nested mocks, to the parent mock, that is why we use
             # _last_send here instead of _original_send
-            return self._last_send(session, request, **kwargs)
+            if isinstance(self._mock_target, type):
+                return self._last_send(session, request, **kwargs)
+            else:
+                return self._last_send(request, **kwargs)
 
-        requests.Session.send = _fake_send
+        _set_method(self._mock_target, "send", _fake_send)
 
     def stop(self):
         """Stop mocking requests.
@@ -136,7 +157,7 @@ def stop(self):
         When nesting mockers, make sure to stop the innermost first.
         """
         if self._last_send:
-            requests.Session.send = self._last_send
+            self._mock_target.send = self._last_send
             self._last_send = None
 
     # for familiarity with MagicMock

tests/pytest/test_with_pytest.py

@@ -50,6 +50,19 @@ def test_redirect_and_nesting():
         assert 'outer' + url_outer == requests.get(url_outer).text  # nosec
 
 
+def test_mixed_mocks():
+    url = 'mock://example.test/'
+    with requests_mock.Mocker() as global_mock:
+        global_mock.get(url, text='global')
+        session = requests.Session()
+        text = session.get(url).text
+        assert text == 'global'  # nosec
+        with requests_mock.Mocker(session=session) as session_mock:
+            session_mock.get(url, real_http=True)
+            text = session.get(url).text
+            assert text == 'global'  # nosec
+
+
 class TestClass(object):
 
     def configure(self, requests_mock):

tests/test_mocker.py

@@ -49,6 +49,66 @@ def test_multiple_starts(self):
         self.assertMockStopped()
         mocker.stop()
 
+    def test_with_session(self):
+        url = 'http://test.url/path'
+        url_inner = 'http://test.url/inner'
+        url_outer = 'http://test.url/outer'
+        with requests_mock.Mocker() as global_mock:
+            global_mock.get(url_outer, text='global')
+
+            session_a = requests.Session()
+            session_b = requests.Session()
+
+            session_a_original_send = session_a.send
+            session_b_original_send = session_b.send
+            self.assertNotEqual(session_a_original_send, session_b_original_send)
+
+            mocker_a = requests_mock.Mocker(session=session_a)
+            mocker_b = requests_mock.Mocker(session=session_b)
+
+            mocker_a.start()
+            mocker_b.start()
+
+            mocker_a.register_uri('GET', url, text='resp_a')
+            mocker_a.register_uri('GET', url_outer, real_http=True)
+            mocker_b.register_uri('GET', url, text='resp_b')
+
+            with requests_mock.Mocker(session=session_b) as mocker_b_inner:
+                mocker_b_inner.register_uri('GET', url, real_http=True)
+                mocker_b_inner.register_uri('GET', url_inner, text='resp_b_inner')
+
+                self.assertEqual('resp_a', session_a.get(url).text)
+                self.assertEqual('resp_b', session_b.get(url).text)
+                self.assertRaises(exceptions.NoMockAddress,
+                                session_a.get,
+                                url_inner)
+                self.assertEqual('resp_b_inner', session_b.get(url_inner).text)
+
+            self.assertEqual('resp_a', session_a.get(url).text)
+            self.assertEqual('resp_b', session_b.get(url).text)
+            self.assertRaises(exceptions.NoMockAddress,
+                                session_a.get,
+                                url_inner)
+            self.assertRaises(exceptions.NoMockAddress,
+                                session_b.get,
+                                url_inner)
+            self.assertEqual('global', session_a.get(url_outer).text)
+            self.assertRaises(exceptions.NoMockAddress,
+                                session_b.get,
+                                url_outer)
+
+            self.assertNotEqual(session_a.send, session_a_original_send)
+            self.assertNotEqual(session_b.send, session_b_original_send)
+            self.assertNotEqual(session_a.send, session_b.send)
+
+            mocker_a.stop()
+            mocker_b.stop()
+
+            self.assertEqual(session_a.send, session_a_original_send)
+            self.assertEqual(session_b.send, session_b_original_send)
+        self.assertEqual(requests.Session.send, original_send)
+
+
     def test_with_context_manager(self):
         self.assertMockStopped()
         with requests_mock.Mocker() as m: