Repository: getmoto/moto (python)

Issue:
Responses as context manager gets overridden if Moto context managers are set up afterwards
If both Moto and responses are set up as context managers, if responses is set up prior to Moto, then any mocked responses from the requests library will be dropped.

Python version: 3.9.16

```text
moto==4.1.11
responses==0.23.1
boto3==1.26.153
requests==2.31.0
```

```python
import boto3
import moto
import pytest
import requests
import responses

url = "https://www.example1234567890.com"


@pytest.fixture(autouse=True)
def aws_credentials(monkeypatch):
    """Mocked AWS Credentials for moto."""

    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
    monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")


def code_under_test():
    client = boto3.resource("s3")
    bucket = client.Bucket("mybucket")
    bucket.Object("my_file.json").get()["Body"].read()
    data = requests.get(url).json()
    assert data == {"foo": "bar"}


@responses.activate
@moto.mock_s3
def test_with_decorators():
    client = boto3.client("s3")
    client.create_bucket(Bucket="mybucket")
    client.put_object(Bucket="mybucket", Key="my_file.json", Body=b"{}")
    responses.add(responses.GET, url, json={"foo": "bar"})

    code_under_test()


@moto.mock_s3
@responses.activate
def test_with_decorators_alternate():
    client = boto3.client("s3")
    client.create_bucket(Bucket="mybucket")
    client.put_object(Bucket="mybucket", Key="my_file.json", Body=b"{}")
    responses.add(responses.GET, url, json={"foo": "bar"})

    code_under_test()


def test_with_context_managers():
    with moto.mock_s3():
        with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
            client = boto3.client("s3")
            client.create_bucket(Bucket="mybucket")
            client.put_object(Bucket="mybucket", Key="my_file.json", Body=b"{}")
            rsps.add(responses.GET, url, json={"foo": "bar"})

            code_under_test()


# When responses is run as a context manager first, alongside moto as a context manager,
# the mocked requests get overriden.
def test_with_context_managers_alternate():
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
        with moto.mock_s3():
            client = boto3.client("s3")
            client.create_bucket(Bucket="mybucket")
            client.put_object(Bucket="mybucket", Key="my_file.json", Body=b"{}")
            rsps.add(responses.GET, url, json={"foo": "bar"})

            code_under_test()
```

This is closely related to #4870 


Pull request:
Support override of Responses.real_send()
Fixes #6417 

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.

Note that this feature requires `responses>=0.24.0`, [released today](https://pypi.org/project/responses/0.24.0/).

```
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")
```

Patch:
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


Test patch:
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())