Trace how MocketSocket is instantiated and file descriptors are created

mocket/utils.py

@@ -20,6 +20,15 @@ def __init__(self, initial_bytes=None, w_fd=None):
     def write(self, content):
         super(MocketSocketCore, self).write(content)
 
+        import sys
+
+        print(
+            __name__,
+            "MocketSocketCore.write",
+            "write_fd",
+            type(self.write_fd),
+            file=sys.stderr,
+        )
         if self.write_fd:
             os.write(self.write_fd, content)
 

tests/tests38/test_http_aiohttp.py

@@ -111,6 +111,26 @@ async def test_https_session(self):
         async def test_no_verify(self):
             Entry.single_register(Entry.GET, self.target_url, status=404)
 
+            import hunter
+            from hunter import Q
+            from hunter.actions import CallPrinter, StackPrinter
+
+            # Predicates for tracing relevant function calls
+            # hunter_predicates = Q(module_startswith="ssl") | Q(module_startswith="aiohttp") | Q(module_startswith="asyncio") | Q(module_startswith="mocket") | Q(module_startswith="http")
+            # hunter.trace(hunter_predicates, action=CallPrinter())
+
+            def is_interesting_call(event):
+                if event.kind != "call":
+                    return False
+                if event.function in ("__init__", "fileno", "fd", "write"):
+                    return True
+                return False
+
+            hunter_predicates = Q(is_interesting_call, module_startswith="mocket")
+            hunter.trace(
+                hunter_predicates, actions=[StackPrinter(depth=5), CallPrinter()]
+            )
+
             async with aiohttp.ClientSession(timeout=self.timeout) as session:
                 async with session.get(self.target_url, ssl=False) as get_response:
                     assert get_response.status == 404