Improve tests for sync implementation slightly.

Author: aaugustin

src/websockets/sync/connection.py

@@ -373,6 +373,7 @@ def send(self, message: Data | Iterable[Data]) -> None:
 
             except RuntimeError:
                 # We didn't start sending a fragmented message.
+                # The connection is still usable.
                 raise
 
             except Exception:
@@ -756,7 +757,7 @@ def set_recv_exc(self, exc: BaseException | None) -> None:
 
         """
         assert self.protocol_mutex.locked()
-        if self.recv_exc is None:
+        if self.recv_exc is None:  # pragma: no branch
             self.recv_exc = exc
 
     def close_socket(self) -> None:

tests/sync/connection.py

@@ -8,7 +8,7 @@ class InterceptingConnection(Connection):
     """
     Connection subclass that can intercept outgoing packets.
 
-    By interfacing with this connection, you can simulate network conditions
+    By interfacing with this connection, we simulate network conditions
     affecting what the component being tested receives during a test.
 
     """

tests/sync/test_connection.py

@@ -246,21 +246,24 @@ def test_recv_streaming_connection_closed_ok(self):
         """recv_streaming raises ConnectionClosedOK after a normal closure."""
         self.remote_connection.close()
         with self.assertRaises(ConnectionClosedOK):
-            list(self.connection.recv_streaming())
+            for _ in self.connection.recv_streaming():
+                self.fail("did not raise")
 
     def test_recv_streaming_connection_closed_error(self):
         """recv_streaming raises ConnectionClosedError after an error."""
         self.remote_connection.close(code=CloseCode.INTERNAL_ERROR)
         with self.assertRaises(ConnectionClosedError):
-            list(self.connection.recv_streaming())
+            for _ in self.connection.recv_streaming():
+                self.fail("did not raise")
 
     def test_recv_streaming_during_recv(self):
         """recv_streaming raises RuntimeError when called concurrently with recv."""
         recv_thread = threading.Thread(target=self.connection.recv)
         recv_thread.start()
 
         with self.assertRaises(RuntimeError) as raised:
-            list(self.connection.recv_streaming())
+            for _ in self.connection.recv_streaming():
+                self.fail("did not raise")
         self.assertEqual(
             str(raised.exception),
             "cannot call recv_streaming while another thread "
@@ -278,7 +281,8 @@ def test_recv_streaming_during_recv_streaming(self):
         recv_streaming_thread.start()
 
         with self.assertRaises(RuntimeError) as raised:
-            list(self.connection.recv_streaming())
+            for _ in self.connection.recv_streaming():
+                self.fail("did not raise")
         self.assertEqual(
             str(raised.exception),
             r"cannot call recv_streaming while another thread "
@@ -374,7 +378,7 @@ def test_send_empty_iterable(self):
         """send does nothing when called with an empty iterable."""
         self.connection.send([])
         self.connection.close()
-        self.assertEqual(list(iter(self.remote_connection)), [])
+        self.assertEqual(list(self.remote_connection), [])
 
     def test_send_mixed_iterable(self):
         """send raises TypeError when called with an iterable of inconsistent types."""
@@ -437,7 +441,7 @@ def test_close_waits_for_connection_closed(self):
 
     def test_close_timeout_waiting_for_close_frame(self):
         """close times out if no close frame is received."""
-        with self.drop_eof_rcvd(), self.drop_frames_rcvd():
+        with self.drop_frames_rcvd(), self.drop_eof_rcvd():
             self.connection.close()
 
         with self.assertRaises(ConnectionClosedError) as raised:
@@ -464,6 +468,10 @@ def test_close_timeout_waiting_for_connection_closed(self):
         self.assertIsInstance(exc.__cause__, (socket.timeout, TimeoutError))
 
     def test_close_waits_for_recv(self):
+        # The sync implementation doesn't have a buffer for incoming messsages.
+        # It requires reading incoming frames until the close frame is reached.
+        # This behavior — close() blocks until recv() is called — is less than
+        # ideal and inconsistent with the asyncio implementation.
         self.remote_connection.send("😀")
 
         close_thread = threading.Thread(target=self.connection.close)
@@ -547,6 +555,25 @@ def closer():
 
         close_thread.join()
 
+    def test_close_during_recv(self):
+        """close aborts recv when called concurrently with recv."""
+
+        def closer():
+            time.sleep(MS)
+            self.connection.close()
+
+        close_thread = threading.Thread(target=closer)
+        close_thread.start()
+
+        with self.assertRaises(ConnectionClosedOK) as raised:
+            self.connection.recv()
+
+        exc = raised.exception
+        self.assertEqual(str(exc), "sent 1000 (OK); then received 1000 (OK)")
+        self.assertIsNone(exc.__cause__)
+
+        close_thread.join()
+
     def test_close_during_send(self):
         """close fails the connection when called concurrently with send."""
         close_gate = threading.Event()
@@ -599,42 +626,45 @@ def test_ping_explicit_binary(self):
         self.connection.ping(b"ping")
         self.assertFrameSent(Frame(Opcode.PING, b"ping"))
 
-    def test_ping_duplicate_payload(self):
-        """ping rejects the same payload until receiving the pong."""
-        with self.remote_connection.protocol_mutex:  # block response to ping
-            pong_waiter = self.connection.ping("idem")
-            with self.assertRaises(RuntimeError) as raised:
-                self.connection.ping("idem")
-            self.assertEqual(
-                str(raised.exception),
-                "already waiting for a pong with the same data",
-            )
-        self.assertTrue(pong_waiter.wait(MS))
-        self.connection.ping("idem")  # doesn't raise an exception
-
     def test_acknowledge_ping(self):
         """ping is acknowledged by a pong with the same payload."""
-        with self.drop_frames_rcvd():
+        with self.drop_frames_rcvd():  # drop automatic response to ping
             pong_waiter = self.connection.ping("this")
-            self.assertFalse(pong_waiter.wait(MS))
         self.remote_connection.pong("this")
         self.assertTrue(pong_waiter.wait(MS))
 
     def test_acknowledge_ping_non_matching_pong(self):
         """ping isn't acknowledged by a pong with a different payload."""
-        with self.drop_frames_rcvd():
+        with self.drop_frames_rcvd():  # drop automatic response to ping
             pong_waiter = self.connection.ping("this")
         self.remote_connection.pong("that")
         self.assertFalse(pong_waiter.wait(MS))
 
     def test_acknowledge_previous_ping(self):
         """ping is acknowledged by a pong with the same payload as a later ping."""
-        with self.drop_frames_rcvd():
+        with self.drop_frames_rcvd():  # drop automatic response to ping
             pong_waiter = self.connection.ping("this")
             self.connection.ping("that")
         self.remote_connection.pong("that")
         self.assertTrue(pong_waiter.wait(MS))
 
+    def test_ping_duplicate_payload(self):
+        """ping rejects the same payload until receiving the pong."""
+        with self.drop_frames_rcvd():  # drop response to ping
+            pong_waiter = self.connection.ping("idem")
+
+        with self.assertRaises(RuntimeError) as raised:
+            self.connection.ping("idem")
+        self.assertEqual(
+            str(raised.exception),
+            "already waiting for a pong with the same data",
+        )
+
+        self.remote_connection.pong("idem")
+        self.assertTrue(pong_waiter.wait(MS))
+
+        self.connection.ping("idem")  # doesn't raise an exception
+
     # Test pong.
 
     def test_pong(self):