diff --git a/releasenotes/notes/session-scoped-mock-7f1c98d9a91bffc8.yaml b/releasenotes/notes/session-scoped-mock-7f1c98d9a91bffc8.yaml new file mode 100644 index 0000000..c21d76f --- /dev/null +++ b/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. diff --git a/requests_mock/mocker.py b/requests_mock/mocker.py index a83103a..a3abaf2 100644 --- a/requests_mock/mocker.py +++ b/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 diff --git a/tests/pytest/test_with_pytest.py b/tests/pytest/test_with_pytest.py index 334d73a..dc58b47 100644 --- a/tests/pytest/test_with_pytest.py +++ b/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): diff --git a/tests/test_mocker.py b/tests/test_mocker.py index 8f70373..cca8f23 100644 --- a/tests/test_mocker.py +++ b/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: