From b40f6060bc280e89c2339220aa9acd337a5e8860 Mon Sep 17 00:00:00 2001 From: Chris Date: Fri, 9 Oct 2026 10:12:03 -0500 Subject: [PATCH 1/2] fix(conformance): record what the aiosendspin SDK reads through receive_timed aiosendspin b3aa88b reads every encrypted connection through EncryptedWebSocket.receive_timed(), which returns the message with the time its last frame arrived and does not call receive(). The control message recorder and the received binary frame recorder wrap receive() only, so from that commit on they recorded nothing a connection received. Every case that judges what the server received then failed, whichever implementation was under test: the stream/start gate found no client/state, and the time exchange found a server/time answering no client/time. Both recorders now wrap receive_timed() as well, where the SDK has it, and restore it on uninstall. --- .../_aiosendspin_protocol_evidence.py | 34 ++++- tests/test_receive_timed_recording.py | 140 ++++++++++++++++++ 2 files changed, 170 insertions(+), 4 deletions(-) create mode 100644 tests/test_receive_timed_recording.py diff --git a/src/conformance/adapters/_aiosendspin_protocol_evidence.py b/src/conformance/adapters/_aiosendspin_protocol_evidence.py index b15fd3e..e296412 100644 --- a/src/conformance/adapters/_aiosendspin_protocol_evidence.py +++ b/src/conformance/adapters/_aiosendspin_protocol_evidence.py @@ -187,8 +187,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 +213,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 +310,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 +489,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 +514,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( diff --git a/tests/test_receive_timed_recording.py b/tests/test_receive_timed_recording.py new file mode 100644 index 0000000..5f37c67 --- /dev/null +++ b/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() From dd77509a8e03acd4489aee3cc9592b37fe5df79d Mon Sep 17 00:00:00 2001 From: Chris Date: Fri, 9 Oct 2026 10:12:18 -0500 Subject: [PATCH 2/2] docs(conformance): say the recorder observes receive_timed too --- src/conformance/adapters/_aiosendspin_protocol_evidence.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/conformance/adapters/_aiosendspin_protocol_evidence.py b/src/conformance/adapters/_aiosendspin_protocol_evidence.py index e296412..a5a584e 100644 --- a/src/conformance/adapters/_aiosendspin_protocol_evidence.py +++ b/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