fix(conformance): record what the aiosendspin SDK reads through receive_timed

src/conformance/adapters/_aiosendspin_protocol_evidence.py

@@ -40,7 +40,8 @@ class level and keeps the JSON body handed to the encrypting transport.
   the same wrapper, which is what orders them.
 - ``client/state`` received and ``stream/start`` sent, in order: full fidelity
   on an encrypted connection, from the same recorder, which also wraps
-  ``EncryptedWebSocket.receive``. The SDK applies a ``client/state`` and sends
+  ``EncryptedWebSocket.receive`` and ``receive_timed``, whichever the SDK
+  reads with. The SDK applies a ``client/state`` and sends
   a ``stream/start`` from its own loops, and its parser rewrites a legacy
   ``state`` into ``available``, so only the transport shows what arrived and
   when. An unencrypted legacy connection bypasses that transport and is not
@@ -187,8 +188,7 @@ async def send_str(transport: Any, data: str) -> None:
             trace.append({**entry, "phase": "sent"})
             self._record_other_sent(socket, began_at)
 
-        async def receive(transport: Any) -> Any:
-            received = await self._original_receive(transport)
+        def observe(transport: Any, received: Any) -> Any:
             if received.type is not WSMsgType.TEXT:
                 return received
             # Marks the socket as one whose incoming text is observed, so that
@@ -214,8 +214,21 @@ async def receive(transport: Any) -> Any:
                 time_exchange.append(message)
             return received
 
+        async def receive(transport: Any) -> Any:
+            return observe(transport, await self._original_receive(transport))
+
+        # The SDK reads an encrypted connection through receive_timed(), which
+        # does not go through receive(); both are observed.
+        self._original_receive_timed = getattr(EncryptedWebSocket, "receive_timed", None)
+
+        async def receive_timed(transport: Any, clock: Any) -> Any:
+            received, received_at = await self._original_receive_timed(transport, clock)
+            return observe(transport, received), received_at
+
         EncryptedWebSocket.send_str = send_str  # type: ignore[method-assign]
         EncryptedWebSocket.receive = receive  # type: ignore[method-assign]
+        if self._original_receive_timed is not None:
+            EncryptedWebSocket.receive_timed = receive_timed  # type: ignore[method-assign]
 
     def initial_activation(self, connection: Any) -> dict[str, Any] | None:
         """
@@ -298,6 +311,8 @@ def uninstall(self) -> None:
         """Restore the SDK transport's own ``send_str`` and ``receive``."""
         self._transport_class.send_str = self._original_send_str  # type: ignore[method-assign]
         self._transport_class.receive = self._original_receive  # type: ignore[method-assign]
+        if self._original_receive_timed is not None:
+            self._transport_class.receive_timed = self._original_receive_timed  # type: ignore[method-assign]
 
     def _record_other_sent(self, socket: Any, began_at: int) -> None:
         # Nothing precedes position 0 for the entry to follow, and one entry
@@ -475,13 +490,23 @@ def __init__(self) -> None:
         self._original_receive = EncryptedWebSocket.receive
         self._frames: list[dict[str, Any]] = []
 
-        async def receive(transport: Any) -> Any:
-            message = await self._original_receive(transport)
+        def observe(message: Any) -> Any:
             if message.type is WSMsgType.BINARY:
                 self._frames.append(binary_frame_record(message.data))
             return message
 
+        async def receive(transport: Any) -> Any:
+            return observe(await self._original_receive(transport))
+
+        self._original_receive_timed = getattr(EncryptedWebSocket, "receive_timed", None)
+
+        async def receive_timed(transport: Any, clock: Any) -> Any:
+            message, received_at = await self._original_receive_timed(transport, clock)
+            return observe(message), received_at
+
         EncryptedWebSocket.receive = receive  # type: ignore[method-assign]
+        if self._original_receive_timed is not None:
+            EncryptedWebSocket.receive_timed = receive_timed  # type: ignore[method-assign]
 
     def frames(self) -> list[dict[str, Any]]:
         """Return one ``{"byte_count": ..., "leading_hex": ...}`` record per frame received."""
@@ -490,6 +515,8 @@ def frames(self) -> list[dict[str, Any]]:
     def uninstall(self) -> None:
         """Restore the SDK transport's own ``receive``."""
         self._transport_class.receive = self._original_receive  # type: ignore[method-assign]
+        if self._original_receive_timed is not None:
+            self._transport_class.receive_timed = self._original_receive_timed  # type: ignore[method-assign]
 
 
 def record_handshake_evidence_server(

tests/test_receive_timed_recording.py

@@ -0,0 +1,140 @@
+"""
+Messages the SDK reads through ``receive_timed`` reach the recorders.
+
+``EncryptedWebSocket.receive_timed`` returns a decrypted message together with
+the time its last frame arrived, and does not go through ``receive``. The SDK
+reads every encrypted connection that way, so a recorder that observes only
+``receive`` sees nothing a connection receives: no ``client/state``, no
+``client/time`` and no binary frame.
+"""
+
+from __future__ import annotations
+
+import json
+import unittest
+from types import SimpleNamespace
+from typing import Any
+from unittest import mock
+
+from aiohttp import WSMsgType
+from aiosendspin.noise.wire import EncryptedWebSocket
+
+from conformance.adapters._aiosendspin_protocol_evidence import (
+    ControlMessageRecorder,
+    ReceivedBinaryFrameRecorder,
+    binary_frame_record,
+)
+
+RECEIVED_AT_US = 1_234_567
+
+
+def _transport(socket: object) -> EncryptedWebSocket:
+    """Build a transport bound to ``socket`` without running a Noise handshake."""
+    transport = object.__new__(EncryptedWebSocket)
+    transport._ws = socket
+    return transport
+
+
+def _connection(socket: object) -> Any:
+    return SimpleNamespace(_wsock_server=socket, _wsock_client=None, is_encrypted=True)
+
+
+def _message(message_type: str, **payload: Any) -> str:
+    return json.dumps({"type": message_type, "payload": payload})
+
+
+class _TimedReceiveCase(unittest.IsolatedAsyncioTestCase):
+    """Stand in for the SDK's ``receive_timed`` before a recorder wraps it."""
+
+    def setUp(self) -> None:
+        self.incoming: list[Any] = []
+
+        async def receive_timed(_transport: EncryptedWebSocket, _clock: Any) -> Any:
+            return self.incoming.pop(0), RECEIVED_AT_US
+
+        self.stand_in = receive_timed
+        patcher = mock.patch.object(EncryptedWebSocket, "receive_timed", receive_timed, create=True)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    async def _receive_timed(
+        self, transport: EncryptedWebSocket, data: Any, kind: Any = WSMsgType.TEXT
+    ) -> Any:
+        message = SimpleNamespace(type=kind, data=data)
+        self.incoming.append(message)
+        returned = await transport.receive_timed(object())
+        # The caller gets back exactly what the SDK's own method returned.
+        self.assertEqual(returned, (message, RECEIVED_AT_US))
+        return returned
+
+
+class ControlMessageRecorderTimedReceiveTests(_TimedReceiveCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.recorder = ControlMessageRecorder()
+        self.addCleanup(self.recorder.uninstall)
+
+    async def test_records_a_client_state_read_through_receive_timed(self) -> None:
+        socket = object()
+        await self._receive_timed(_transport(socket), _message("client/state", available=True))
+
+        self.assertEqual(
+            self.recorder.availability_trace(_connection(socket)),
+            [{"type": "client/state", "available": True}],
+        )
+
+    async def test_records_a_client_time_read_through_receive_timed(self) -> None:
+        socket = object()
+        client_time = _message("client/time", client_transmitted=42)
+        await self._receive_timed(_transport(socket), client_time)
+
+        self.assertEqual(
+            self.recorder.time_exchange(_connection(socket)), [json.loads(client_time)]
+        )
+
+    async def test_uninstall_restores_the_sdks_own_receive_timed(self) -> None:
+        self.assertIsNot(EncryptedWebSocket.receive_timed, self.stand_in)
+
+        self.recorder.uninstall()
+
+        self.assertIs(EncryptedWebSocket.receive_timed, self.stand_in)
+
+
+class ReceivedBinaryFrameRecorderTimedReceiveTests(_TimedReceiveCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.recorder = ReceivedBinaryFrameRecorder()
+        self.addCleanup(self.recorder.uninstall)
+
+    async def test_records_a_binary_frame_read_through_receive_timed(self) -> None:
+        frame = bytes([4]) + bytes(12) + b"audio"
+        await self._receive_timed(_transport(object()), frame, WSMsgType.BINARY)
+
+        self.assertEqual(self.recorder.frames(), [binary_frame_record(frame)])
+
+    async def test_uninstall_restores_the_sdks_own_receive_timed(self) -> None:
+        self.assertIsNot(EncryptedWebSocket.receive_timed, self.stand_in)
+
+        self.recorder.uninstall()
+
+        self.assertIs(EncryptedWebSocket.receive_timed, self.stand_in)
+
+
+class SdkWithoutReceiveTimedTests(unittest.TestCase):
+    """An SDK that predates ``receive_timed`` is recorded through ``receive`` as before."""
+
+    def test_recorders_install_and_uninstall_without_it(self) -> None:
+        original = EncryptedWebSocket.__dict__.get("receive_timed")
+        if original is not None:
+            delattr(EncryptedWebSocket, "receive_timed")
+            self.addCleanup(setattr, EncryptedWebSocket, "receive_timed", original)
+
+        for recorder_class in (ControlMessageRecorder, ReceivedBinaryFrameRecorder):
+            with self.subTest(recorder_class.__name__):
+                recorder = recorder_class()
+                recorder.uninstall()
+                self.assertFalse(hasattr(EncryptedWebSocket, "receive_timed"))
+
+
+if __name__ == "__main__":
+    unittest.main()