Core: Use internal server to test requests passthrough

CHANGELOG.md

@@ -1,6 +1,16 @@
 Moto Changelog
 ==============
 
+5.0.24
+-----
+Docker Digest for 5.0.24: <autopopulateddigest>
+
+    Miscellaneous:
+        * EC2: Terminating instances will now release private ip addresses from the NIC's attached to the interface
+        * S3: Fixes a bug in complete_multipart_upload() where it was no longer possible to overwrite an earlier multipart upload (Broken in 5.0.23)
+        * S3: get_object_cors() now correctly returns the ExposeHeader-value
+
+
 5.0.23
 -----
 Docker Digest for 5.0.23: _sha256:d41e007bb1f7d41b530959ae9cbed1edf42737ee839faf8da7e925bf19f63105_

tests/test_core/test_responses_module.py

@@ -2,6 +2,9 @@
 Ensure that the responses module plays nice with our mocks
 """
 
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from threading import Event, Thread
+from typing import Optional, Tuple
 from unittest import SkipTest, TestCase
 
 import boto3
@@ -67,6 +70,45 @@ def test_moto_as_late_as_possible(self) -> None:
             assert r.json() == {"a": "4"}
 
 
+class WebRequestHandler(BaseHTTPRequestHandler):
+    def do_GET(self) -> None:
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(b"real response")
+
+
+class SimpleServer:
+    def __init__(self) -> None:
+        self._port = 0
+
+        self._thread: Optional[Thread] = None
+        self._ip_address = "0.0.0.0"
+        self._server: Optional[HTTPServer] = None
+        self._server_ready_event = Event()
+
+    def _server_entry(self) -> None:
+        self._server = HTTPServer(("0.0.0.0", 0), WebRequestHandler)
+        self._server_ready_event.set()
+        self._server.serve_forever()
+
+    def start(self) -> None:
+        self._thread = Thread(target=self._server_entry, daemon=True)
+        self._thread.start()
+        self._server_ready_event.wait()
+
+    def get_host_and_port(self) -> Tuple[str, int]:
+        assert self._server
+        host, port = self._server.server_address
+        return (str(host), port)
+
+    def stop(self) -> None:
+        if self._server:
+            self._server.shutdown()
+        if self._thread:
+            self._thread.join()
+
+
 @mock_aws
 class TestResponsesMockWithPassThru(TestCase):
     """
@@ -77,24 +119,29 @@ def setUp(self) -> None:
         if RESPONSES_VERSION < LooseVersion("0.24.0"):
             raise SkipTest("Can only test this with responses >= 0.24.0")
 
-        raise SkipTest("Endpoint that we're trying to reach is down")
+        self.server = SimpleServer()
+        self.server.start()
+        host, port = self.server.get_host_and_port()
+        self.server_url = f"http://{host}:{port}"
 
         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")
+        self.r_mock.add_passthru(self.server_url)
 
     def tearDown(self) -> None:
         self.r_mock.stop()  # type: ignore
         self.r_mock.reset()  # type: ignore
         override_responses_real_send(None)
 
+        self.server.stop()
+
     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
+        assert requests.get(self.server_url).content == b"real response"
 
         return "OK"