From 56f7d1fdf59123b45373c285353361b7844a3274 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Sat, 4 Nov 2023 14:21:00 -0100 Subject: [PATCH] Support override of Responses.real_send() (#6988) --- docs/docs/faq.rst | 16 +++++++ moto/core/models.py | 17 +++++++ setup.cfg | 2 +- tests/test_core/test_responses_module.py | 59 +++++++++++++++++++++--- 4 files changed, 87 insertions(+), 7 deletions(-) diff --git a/docs/docs/faq.rst b/docs/docs/faq.rst index 866d88aea..5d79dc1e7 100644 --- a/docs/docs/faq.rst +++ b/docs/docs/faq.rst @@ -30,3 +30,19 @@ If you want to mock the default region, as an additional layer of protection aga os.environ["MOTO_ALLOW_NONEXISTENT_REGION"] = True os.environ["AWS_DEFAULT_REGION"] = "antarctica" + + +How can I mock my own HTTP-requests, using the Responses-module? +################################################################ + +Moto uses it's own Responses-mock to intercept AWS requests, so if you need to intercept custom (non-AWS) request as part of your tests, you may find that Moto 'swallows' any pass-thru's that you have defined. +You can pass your own Responses-mock to Moto, to ensure that any custom (non-AWS) are handled by that Responses-mock. + +.. sourcecode:: python + + from moto.core.models import override_responses_real_send + + my_own_mock = responses.RequestsMock(assert_all_requests_are_fired=True) + override_responses_real_send(my_own_mock) + my_own_mock.start() + my_own_mock.add_passthru("http://some-website.com") diff --git a/moto/core/models.py b/moto/core/models.py index c879fd278..9957c6f35 100644 --- a/moto/core/models.py +++ b/moto/core/models.py @@ -302,6 +302,23 @@ def patch_resource(resource: Any) -> None: raise Exception(f"Argument {resource} should be of type boto3.resource") +def override_responses_real_send(user_mock: Optional[responses.RequestsMock]) -> None: + """ + Moto creates it's own Responses-object responsible for intercepting AWS requests + If a custom Responses-object is created by the user, Moto will hijack any of the pass-thru's set + + Call this method to ensure any requests unknown to Moto are passed through the custom Responses-object. + + Set the user_mock argument to None to reset this behaviour. + + Note that this is only supported from Responses>=0.24.0 + """ + if user_mock is None: + responses_mock._real_send = responses._real_send + else: + responses_mock._real_send = user_mock.unbound_on_send() + + class BotocoreEventMockAWS(BaseMockAWS): def reset(self) -> None: botocore_stubber.reset() diff --git a/setup.cfg b/setup.cfg index 1412fb4c2..323c81606 100644 --- a/setup.cfg +++ b/setup.cfg @@ -273,7 +273,7 @@ disable = W,C,R,E enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import [mypy] -files= moto, tests/test_core/test_mock_all.py, tests/test_core/test_decorator_calls.py +files= moto, tests/test_core/test_mock_all.py, tests/test_core/test_decorator_calls.py, tests/test_core/test_responses_module.py show_column_numbers=True show_error_codes = True disable_error_code=abstract diff --git a/tests/test_core/test_responses_module.py b/tests/test_core/test_responses_module.py index d989ae2e9..1fad92da5 100644 --- a/tests/test_core/test_responses_module.py +++ b/tests/test_core/test_responses_module.py @@ -5,18 +5,21 @@ Ensure that the responses module plays nice with our mocks import boto3 import requests import responses -from moto import mock_s3, settings +from moto import mock_dynamodb, mock_s3, settings +from moto.core.models import override_responses_real_send +from moto.core.versions import RESPONSES_VERSION +from moto.utilities.distutils_version import LooseVersion from unittest import SkipTest, TestCase class TestResponsesModule(TestCase): - def setUp(self): + def setUp(self) -> None: if settings.TEST_SERVER_MODE: raise SkipTest("No point in testing responses-decorator in ServerMode") @mock_s3 @responses.activate - def test_moto_first(self): + def test_moto_first(self) -> None: """ Verify we can activate a user-defined `responses` on top of our Moto mocks """ @@ -24,13 +27,13 @@ class TestResponsesModule(TestCase): @responses.activate @mock_s3 - def test_moto_second(self): + def test_moto_second(self) -> None: """ Verify we can load Moto after activating a `responses`-mock """ self.moto_responses_compatibility() - def moto_responses_compatibility(self): + def moto_responses_compatibility(self) -> None: responses.add( responses.GET, url="http://127.0.0.1/lkdsfjlkdsa", json={"a": "4"} ) @@ -42,7 +45,7 @@ class TestResponsesModule(TestCase): assert r.json() == {"a": "4"} @responses.activate - def test_moto_as_late_as_possible(self): + def test_moto_as_late_as_possible(self) -> None: """ Verify we can load moto after registering a response """ @@ -60,3 +63,47 @@ class TestResponsesModule(TestCase): # And outside of Moto with requests.get("http://127.0.0.1/lkdsfjlkdsa") as r: assert r.json() == {"a": "4"} + + +@mock_dynamodb +class TestResponsesMockWithPassThru(TestCase): + """ + https://github.com/getmoto/moto/issues/6417 + """ + + def setUp(self) -> None: + if RESPONSES_VERSION < LooseVersion("0.24.0"): + raise SkipTest("Can only test this with responses >= 0.24.0") + + self.r_mock = responses.RequestsMock(assert_all_requests_are_fired=True) + override_responses_real_send(self.r_mock) + self.r_mock.start() + self.r_mock.add_passthru("http://ip.jsontest.com") + + def tearDown(self) -> None: + self.r_mock.stop() + self.r_mock.reset() + override_responses_real_send(None) + + def http_requests(self) -> str: + # Mock this website + requests.post("https://example.org") + + # Passthrough this website + assert requests.get("http://ip.jsontest.com").status_code == 200 + + return "OK" + + def aws_and_http_requests(self) -> str: + ddb = boto3.client("dynamodb", "us-east-1") + assert ddb.list_tables()["TableNames"] == [] + self.http_requests() + return "OK" + + def test_http_requests(self) -> None: + self.r_mock.add(responses.POST, "https://example.org", status=200) + self.assertEqual("OK", self.http_requests()) + + def test_aws_and_http_requests(self) -> None: + self.r_mock.add(responses.POST, "https://example.org", status=200) + self.assertEqual("OK", self.aws_and_http_requests())