Support for calls made by fastapi (#188)

* Introduce a `socket.socketpair()` function that acts as the real one

Author: mindflayer

Pipfile

@@ -29,6 +29,7 @@ build = "*"
 wheel = "*"
 twine = "*"
 anaconda-client = "*"
+fastapi = "*"
 
 [requires]
 python_version = "3"

mocket/__init__.py

@@ -3,4 +3,4 @@
 
 __all__ = ("async_mocketize", "mocketize", "Mocket", "MocketEntry", "Mocketizer")
 
-__version__ = "3.10.7"
+__version__ = "3.10.8"

mocket/mocket.py

@@ -51,6 +51,7 @@
 true_gethostbyname = socket.gethostbyname
 true_gethostname = socket.gethostname
 true_getaddrinfo = socket.getaddrinfo
+true_socketpair = socket.socketpair
 true_ssl_wrap_socket = ssl.wrap_socket
 true_ssl_socket = ssl.SSLSocket
 true_ssl_context = ssl.SSLContext
@@ -136,6 +137,13 @@ def create_connection(address, timeout=None, source_address=None):
     return s
 
 
+def socketpair():
+    """Returns a real socketpair() used by asyncio loop for supporting calls made by fastapi and similar services."""
+    import _socket
+
+    return _socket.socketpair()
+
+
 def _hash_request(h, req):
     return h(encode_to_bytes("".join(sorted(req.split("\r\n"))))).hexdigest()
 
@@ -488,6 +496,7 @@ def enable(namespace=None, truesocket_recording_dir=None):
         ] = lambda host, port, family=None, socktype=None, proto=None, flags=None: [
             (2, 1, 6, "", (host, port))
         ]
+        socket.socketpair = socket.__dict__["socketpair"] = socketpair
         ssl.wrap_socket = ssl.__dict__["wrap_socket"] = FakeSSLContext.wrap_socket
         ssl.SSLContext = ssl.__dict__["SSLContext"] = FakeSSLContext
         socket.inet_pton = socket.__dict__["inet_pton"] = lambda family, ip: byte_type(
@@ -520,6 +529,7 @@ def disable():
         socket.gethostname = socket.__dict__["gethostname"] = true_gethostname
         socket.gethostbyname = socket.__dict__["gethostbyname"] = true_gethostbyname
         socket.getaddrinfo = socket.__dict__["getaddrinfo"] = true_getaddrinfo
+        socket.socketpair = socket.__dict__["socketpair"] = true_socketpair
         ssl.wrap_socket = ssl.__dict__["wrap_socket"] = true_ssl_wrap_socket
         ssl.SSLContext = ssl.__dict__["SSLContext"] = true_ssl_context
         socket.inet_pton = socket.__dict__["inet_pton"] = true_inet_pton

tests/tests37/test_httpx.py

@@ -2,8 +2,10 @@
 
 import httpx
 import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
 
-from mocket import Mocketizer, async_mocketize
+from mocket import Mocketizer, async_mocketize, mocketize
 from mocket.mockhttp import Entry
 
 
@@ -48,3 +50,28 @@ async def test_httpx_fixture(httpx_client):
         response = await client.get(url)
 
         assert response.json() == data
+
+
+def create_app() -> FastAPI:
+    app = FastAPI()
+
+    @app.get("/")
+    async def read_main() -> dict:
+        async with httpx.AsyncClient() as client:
+            r = await client.get("https://example.org/")
+            return r.json()
+
+    return app
+
+
+@mocketize
+def test_call_from_fastapi() -> None:
+    app = create_app()
+    client = TestClient(app)
+
+    Entry.single_register(Entry.GET, "https://example.org/", body='{"id": 1}')
+
+    response = client.get("/")
+
+    assert response.status_code == 200
+    assert response.json() == {"id": 1}