Support override of Responses.real_send() (getmoto#6988)

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

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

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

tests/test_core/test_responses_module.py

@@ -5,32 +5,35 @@
 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
         """
         self.moto_responses_compatibility()
 
     @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 @@ def moto_responses_compatibility(self):
             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 @@ def test_moto_as_late_as_possible(self):
         # 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())