content: import asyncio
  import contextlib
  import logging
  import sys
  import unittest
  import unittest.mock
  import warnings
  
  from websockets.exceptions import ConnectionClosed, InvalidState
  from websockets.frames import (
      OP_BINARY,
      OP_CLOSE,
      OP_CONT,
      OP_PING,
      OP_PONG,
      OP_TEXT,
      Close,
      CloseCode,
  )
  from websockets.legacy.framing import Frame
  from websockets.legacy.protocol import WebSocketCommonProtocol, broadcast
  from websockets.protocol import State
  
  from ..utils import MS
  from .utils import AsyncioTestCase
  
  
  async def async_iterable(iterable):
      for item in iterable:
          yield item
  
  
  class TransportMock(unittest.mock.Mock):
      """
      Transport mock to control the protocol's inputs and outputs in tests.
  
      It calls the protocol's connection_made and connection_lost methods like
      actual transports.
  
      It also calls the protocol's connection_open method to bypass the
      WebSocket handshake.
  
      To simulate incoming data, tests call the protocol's data_received and
      eof_received methods directly.
  
      They could also pause_writing and resume_writing to test flow control.
  
      """
  
      # This should happen in __init__ but overriding Mock.__init__ is hard.
      def setup_mock(self, loop, protocol):
          self.loop = loop
          self.protocol = protocol
          self._eof = False
          self._closing = False
          # Simulate a successful TCP handshake.
          self.protocol.connection_made(self)
          # Simulate a successful WebSocket handshake.
          self.protocol.connection_open()
  
      def can_write_eof(self):
          return True
  
      def write_eof(self):
          # When the protocol half-closes the TCP connection, it expects the
          # other end to close it. Simulate that.
          if not self._eof:
              self.loop.call_soon(self.close)
          self._eof = True
  
      def close(self):
          # Simulate how actual transports drop the connection.
          if not self._closing:
              self.loop.call_soon(self.protocol.connection_lost, None)
          self._closing = True
  
      def abort(self):
          # Change this to an `if` if tests call abort() multiple times.
          assert self.protocol.state is not State.CLOSED
          self.loop.call_soon(self.protocol.connection_lost, None)
  
  
  class CommonTests:
      """
      Mixin that defines most tests but doesn't inherit unittest.TestCase.
  
      Tests are run by the ServerTests and ClientTests subclasses.
  
      """
  
      def setUp(self):
          super().setUp()
  
          # This logic is encapsulated in a coroutine to prevent it from executing
          # before the event loop is running which causes asyncio.get_event_loop()
          # to raise a DeprecationWarning on Python ≥ 3.10.
  
          async def create_protocol():
              # Disable pings to make it easier to test what frames are sent exactly.
              return WebSocketCommonProtocol(ping_interval=None)
  
          self.protocol = self.loop.run_until_complete(create_protocol())
          self.transport = TransportMock()
          self.transport.setup_mock(self.loop, self.protocol)
  
      def tearDown(self):
          self.transport.close()
          self.loop.run_until_complete(self.protocol.close())
          super().tearDown()
  
      # Utilities for writing tests.
  
      def make_drain_slow(self, delay=MS):
          # Process connection_made in order to initialize self.protocol.transport.
          self.run_loop_once()
  
          original_drain = self.protocol._drain
  
          async def delayed_drain():
              await asyncio.sleep(delay)
              await original_drain()
  
          self.protocol._drain = delayed_drain
  
      close_frame = Frame(
          True,
          OP_CLOSE,
          Close(CloseCode.NORMAL_CLOSURE, "close").serialize(),
      )
      local_close = Frame(
          True,
          OP_CLOSE,
          Close(CloseCode.NORMAL_CLOSURE, "local").serialize(),
      )
      remote_close = Frame(
          True,
          OP_CLOSE,
          Close(CloseCode.NORMAL_CLOSURE, "remote").serialize(),
      )
  
      def receive_frame(self, frame):
          """
          Make the protocol receive a frame.
  
          """
          write = self.protocol.data_received
          mask = not self.protocol.is_client
          frame.write(write, mask=mask)
  
      def receive_eof(self):
          """
          Make the protocol receive the end of the data stream.
  
          Since ``WebSocketCommonProtocol.eof_received`` returns ``None``, an
          actual transport would close itself after calling it. This function
          emulates that behavior.
  
          """
          self.protocol.eof_received()
          self.loop.call_soon(self.transport.close)
  
      def receive_eof_if_client(self):
          """
          Like receive_eof, but only if this is the client side.
  
          Since the server is supposed to initiate the termination of the TCP
          connection, this method helps making tests work for both sides.
  
          """
          if self.protocol.is_client:
              self.receive_eof()
  
      def close_connection(self, code=CloseCode.NORMAL_CLOSURE, reason="close"):
          """
          Execute a closing handshake.
  
          This puts the connection in the CLOSED state.
  
          """
          close_frame_data = Close(code, reason).serialize()
          # Prepare the response to the closing handshake from the remote side.
          self.receive_frame(Frame(True, OP_CLOSE, close_frame_data))
          self.receive_eof_if_client()
          # Trigger the closing handshake from the local side and complete it.
          self.loop.run_until_complete(self.protocol.close(code, reason))
          # Empty the outgoing data stream so we can make assertions later on.
          self.assertOneFrameSent(True, OP_CLOSE, close_frame_data)
  
          assert self.protocol.state is State.CLOSED
  
      def half_close_connection_local(
          self,
          code=CloseCode.NORMAL_CLOSURE,
          reason="close",
      ):
          """
          Start a closing handshake but do not complete it.
  
          The main difference with `close_connection` is that the connection is
          left in the CLOSING state until the event loop runs again.
  
          The current implementation returns a task that must be awaited or
          canceled, else asyncio complains about destroying a pending task.
  
          """
          close_frame_data = Close(code, reason).serialize()
          # Trigger the closing handshake from the local endpoint.
          close_task = self.loop.create_task(self.protocol.close(code, reason))
          self.run_loop_once()  # write_frame executes
          # Empty the outgoing data stream so we can make assertions later on.
          self.assertOneFrameSent(True, OP_CLOSE, close_frame_data)
  
          assert self.protocol.state is State.CLOSING
  
          # Complete the closing sequence at 1ms intervals so the test can run
          # at each point even it goes back to the event loop several times.
          self.loop.call_later(
              MS, self.receive_frame, Frame(True, OP_CLOSE, close_frame_data)
          )
          self.loop.call_later(2 * MS, self.receive_eof_if_client)
  
          # This task must be awaited or canceled by the caller.
          return close_task
  
      def half_close_connection_remote(
          self,
          code=CloseCode.NORMAL_CLOSURE,
          reason="close",
      ):
          """
          Receive a closing handshake but do not complete it.
  
          The main difference with `close_connection` is that the connection is
          left in the CLOSING state until the event loop runs again.
  
          """
          # On the server side, websockets completes the closing handshake and
          # closes the TCP connection immediately. Yield to the event loop after
          # sending the close frame to run the test while the connection is in
          # the CLOSING state.
          if not self.protocol.is_client:
              self.make_drain_slow()
  
          close_frame_data = Close(code, reason).serialize()
          # Trigger the closing handshake from the remote endpoint.
          self.receive_frame(Frame(True, OP_CLOSE, close_frame_data))
          self.run_loop_once()  # read_frame executes
          # Empty the outgoing data stream so we can make assertions later on.
          self.assertOneFrameSent(True, OP_CLOSE, close_frame_data)
  
          assert self.protocol.state is State.CLOSING
  
          # Complete the closing sequence at 1ms intervals so the test can run
          # at each point even it goes back to the event loop several times.
          self.loop.call_later(2 * MS, self.receive_eof_if_client)
  
      def process_invalid_frames(self):
          """
          Make the protocol fail quickly after simulating invalid data.
  
          To achieve this, this function triggers the protocol's eof_received,
          which interrupts pending reads waiting for more data.
  
          """
          self.run_loop_once()
          self.receive_eof()
          self.loop.run_until_complete(self.protocol.close_connection_task)
  
      def sent_frames(self):
          """
          Read all frames sent to the transport.
  
          """
          stream = asyncio.StreamReader(loop=self.loop)
  
          for (data,), kw in self.transport.write.call_args_list:
              stream.feed_data(data)
          self.transport.write.call_args_list = []
          stream.feed_eof()
  
          frames = []
          while not stream.at_eof():
              frames.append(
                  self.loop.run_until_complete(
                      Frame.read(stream.readexactly, mask=self.protocol.is_client)
                  )
              )
          return frames
  
      def last_sent_frame(self):
          """
          Read the last frame sent to the transport.
  
          This method assumes that at most one frame was sent. It raises an
          AssertionError otherwise.
  
          """
          frames = self.sent_frames()
          if frames:
              assert len(frames) == 1
              return frames[0]
  
      def assertFramesSent(self, *frames):
          self.assertEqual(self.sent_frames(), [Frame(*args) for args in frames])
  
      def assertOneFrameSent(self, *args):
          self.assertEqual(self.last_sent_frame(), Frame(*args))
  
      def assertNoFrameSent(self):
          self.assertIsNone(self.last_sent_frame())
  
      def assertConnectionClosed(self, code, message):
          # The following line guarantees that connection_lost was called.
          self.assertEqual(self.protocol.state, State.CLOSED)
          # A close frame was received.
          self.assertEqual(self.protocol.close_code, code)
          self.assertEqual(self.protocol.close_reason, message)
  
      def assertConnectionFailed(self, code, message):
          # The following line guarantees that connection_lost was called.
          self.assertEqual(self.protocol.state, State.CLOSED)
          # No close frame was received.
          self.assertEqual(self.protocol.close_code, CloseCode.ABNORMAL_CLOSURE)
          self.assertEqual(self.protocol.close_reason, "")
          # A close frame was sent -- unless the connection was already lost.
          if code == CloseCode.ABNORMAL_CLOSURE:
              self.assertNoFrameSent()
          else:
              self.assertOneFrameSent(True, OP_CLOSE, Close(code, message).serialize())
  
      @contextlib.contextmanager
      def assertCompletesWithin(self, min_time, max_time):
          t0 = self.loop.time()
          yield
          t1 = self.loop.time()
          dt = t1 - t0
          self.assertGreaterEqual(dt, min_time, f"Too fast: {dt} < {min_time}")
          self.assertLess(dt, max_time, f"Too slow: {dt} >= {max_time}")
  
      # Test constructor.
  
      def test_timeout_backwards_compatibility(self):
          async def create_protocol():
              return WebSocketCommonProtocol(ping_interval=None, timeout=5)
  
          with warnings.catch_warnings(record=True) as recorded:
              warnings.simplefilter("always")
              protocol = self.loop.run_until_complete(create_protocol())
  
          self.assertEqual(protocol.close_timeout, 5)
          self.assertDeprecationWarnings(recorded, ["rename timeout to close_timeout"])
  
      def test_loop_backwards_compatibility(self):
          loop = asyncio.new_event_loop()
          self.addCleanup(loop.close)
  
          with warnings.catch_warnings(record=True) as recorded:
              warnings.simplefilter("always")
              protocol = WebSocketCommonProtocol(ping_interval=None, loop=loop)
  
          self.assertEqual(protocol.loop, loop)
          self.assertDeprecationWarnings(recorded, ["remove loop argument"])
  
      # Test public attributes.
  
      def test_local_address(self):
          get_extra_info = unittest.mock.Mock(return_value=("host", 4312))
          self.transport.get_extra_info = get_extra_info
  
          self.assertEqual(self.protocol.local_address, ("host", 4312))
          get_extra_info.assert_called_with("sockname")
  
      def test_local_address_before_connection(self):
          # Emulate the situation before connection_open() runs.
          _transport = self.protocol.transport
          del self.protocol.transport
          try:
              self.assertEqual(self.protocol.local_address, None)
          finally:
              self.protocol.transport = _transport
  
      def test_remote_address(self):
          get_extra_info = unittest.mock.Mock(return_value=("host", 4312))
          self.transport.get_extra_info = get_extra_info
  
          self.assertEqual(self.protocol.remote_address, ("host", 4312))
          get_extra_info.assert_called_with("peername")
  
      def test_remote_address_before_connection(self):
          # Emulate the situation before connection_open() runs.
          _transport = self.protocol.transport
          del self.protocol.transport
          try:
              self.assertEqual(self.protocol.remote_address, None)
          finally:
              self.protocol.transport = _transport
  
      def test_open(self):
          self.assertTrue(self.protocol.open)
          self.close_connection()
          self.assertFalse(self.protocol.open)
  
      def test_closed(self):
          self.assertFalse(self.protocol.closed)
          self.close_connection()
          self.assertTrue(self.protocol.closed)
  
      def test_wait_closed(self):
          wait_closed = self.loop.create_task(self.protocol.wait_closed())
          self.assertFalse(wait_closed.done())
          self.close_connection()
          self.assertTrue(wait_closed.done())
  
      def test_close_code(self):
          self.close_connection(CloseCode.GOING_AWAY, "Bye!")
          self.assertEqual(self.protocol.close_code, CloseCode.GOING_AWAY)
  
      def test_close_reason(self):
          self.close_connection(CloseCode.GOING_AWAY, "Bye!")
          self.assertEqual(self.protocol.close_reason, "Bye!")
  
      def test_close_code_not_set(self):
          self.assertIsNone(self.protocol.close_code)
  
      def test_close_reason_not_set(self):
          self.assertIsNone(self.protocol.close_reason)
  
      # Test the recv coroutine.
  
      def test_recv_text(self):
          self.receive_frame(Frame(True, OP_TEXT, "café".encode()))
          data = self.loop.run_until_complete(self.protocol.recv())
          self.assertEqual(data, "café")
  
      def test_recv_binary(self):
          self.receive_frame(Frame(True, OP_BINARY, b"tea"))
          data = self.loop.run_until_complete(self.protocol.recv())
          self.assertEqual(data, b"tea")
  
      def test_recv_on_closing_connection_local(self):
          close_task = self.half_close_connection_local()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.recv())
  
          self.loop.run_until_complete(close_task)  # cleanup
  
      def test_recv_on_closing_connection_remote(self):
          self.half_close_connection_remote()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.recv())
  
      def test_recv_on_closed_connection(self):
          self.close_connection()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.recv())
  
      def test_recv_protocol_error(self):
          self.receive_frame(Frame(True, OP_CONT, "café".encode()))
          self.process_invalid_frames()
          self.assertConnectionFailed(CloseCode.PROTOCOL_ERROR, "")
  
      def test_recv_unicode_error(self):
          self.receive_frame(Frame(True, OP_TEXT, "café".encode("latin-1")))
          self.process_invalid_frames()
          self.assertConnectionFailed(CloseCode.INVALID_DATA, "")
  
      def test_recv_text_payload_too_big(self):
          self.protocol.max_size = 1024
          self.receive_frame(Frame(True, OP_TEXT, "café".encode() * 205))
          self.process_invalid_frames()
          self.assertConnectionFailed(CloseCode.MESSAGE_TOO_BIG, "")
  
      def test_recv_binary_payload_too_big(self):
          self.protocol.max_size = 1024
          self.receive_frame(Frame(True, OP_BINARY, b"tea" * 342))
          self.process_invalid_frames()
          self.assertConnectionFailed(CloseCode.MESSAGE_TOO_BIG, "")
  
      def test_recv_text_no_max_size(self):
          self.protocol.max_size = None  # for test coverage
          self.receive_frame(Frame(True, OP_TEXT, "café".encode() * 205))
          data = self.loop.run_until_complete(self.protocol.recv())
          self.assertEqual(data, "café" * 205)
  
      def test_recv_binary_no_max_size(self):
          self.protocol.max_size = None  # for test coverage
          self.receive_frame(Frame(True, OP_BINARY, b"tea" * 342))
          data = self.loop.run_until_complete(self.protocol.recv())
          self.assertEqual(data, b"tea" * 342)
  
      def test_recv_queue_empty(self):
          recv = self.loop.create_task(self.protocol.recv())
          with self.assertRaises(asyncio.TimeoutError):
              self.loop.run_until_complete(
                  asyncio.wait_for(asyncio.shield(recv), timeout=MS)
              )
  
          self.receive_frame(Frame(True, OP_TEXT, "café".encode()))
          data = self.loop.run_until_complete(recv)
          self.assertEqual(data, "café")
  
      def test_recv_queue_full(self):
          self.protocol.max_queue = 2
          # Test internals because it's hard to verify buffers from the outside.
          self.assertEqual(list(self.protocol.messages), [])
  
          self.receive_frame(Frame(True, OP_TEXT, "café".encode()))
          self.run_loop_once()
          self.assertEqual(list(self.protocol.messages), ["café"])
  
          self.receive_frame(Frame(True, OP_BINARY, b"tea"))
          self.run_loop_once()
          self.assertEqual(list(self.protocol.messages), ["café", b"tea"])
  
          self.receive_frame(Frame(True, OP_BINARY, b"milk"))
          self.run_loop_once()
          self.assertEqual(list(self.protocol.messages), ["café", b"tea"])
  
          self.loop.run_until_complete(self.protocol.recv())
          self.run_loop_once()
          self.assertEqual(list(self.protocol.messages), [b"tea", b"milk"])
  
          self.loop.run_until_complete(self.protocol.recv())
          self.run_loop_once()
          self.assertEqual(list(self.protocol.messages), [b"milk"])
  
          self.loop.run_until_complete(self.protocol.recv())
          self.run_loop_once()
          self.assertEqual(list(self.protocol.messages), [])
  
      def test_recv_queue_no_limit(self):
          self.protocol.max_queue = None
  
          for _ in range(100):
              self.receive_frame(Frame(True, OP_TEXT, "café".encode()))
              self.run_loop_once()
  
          # Incoming message queue can contain at least 100 messages.
          self.assertEqual(list(self.protocol.messages), ["café"] * 100)
  
          for _ in range(100):
              self.loop.run_until_complete(self.protocol.recv())
  
          self.assertEqual(list(self.protocol.messages), [])
  
      def test_recv_other_error(self):
          async def read_message():
              raise Exception("BOOM")
  
          self.protocol.read_message = read_message
          self.process_invalid_frames()
          self.assertConnectionFailed(CloseCode.INTERNAL_ERROR, "")
  
      def test_recv_canceled(self):
          recv = self.loop.create_task(self.protocol.recv())
          self.loop.call_soon(recv.cancel)
  
          with self.assertRaises(asyncio.CancelledError):
              self.loop.run_until_complete(recv)
  
          # The next frame doesn't disappear in a vacuum (it used to).
          self.receive_frame(Frame(True, OP_TEXT, "café".encode()))
          data = self.loop.run_until_complete(self.protocol.recv())
          self.assertEqual(data, "café")
  
      def test_recv_canceled_race_condition(self):
          recv = self.loop.create_task(
              asyncio.wait_for(self.protocol.recv(), timeout=0.000_001)
          )
          self.loop.call_soon(self.receive_frame, Frame(True, OP_TEXT, "café".encode()))
  
          with self.assertRaises(asyncio.TimeoutError):
              self.loop.run_until_complete(recv)
  
          # The previous frame doesn't disappear in a vacuum (it used to).
          self.receive_frame(Frame(True, OP_TEXT, "tea".encode()))
          data = self.loop.run_until_complete(self.protocol.recv())
          # If we're getting "tea" there, it means "café" was swallowed (ha, ha).
          self.assertEqual(data, "café")
  
      def test_recv_when_transfer_data_cancelled(self):
          # Clog incoming queue.
          self.protocol.max_queue = 1
          self.receive_frame(Frame(True, OP_TEXT, "café".encode()))
          self.receive_frame(Frame(True, OP_BINARY, b"tea"))
          self.run_loop_once()
  
          # Flow control kicks in (check with an implementation detail).
          self.assertFalse(self.protocol._put_message_waiter.done())
  
          # Schedule recv().
          recv = self.loop.create_task(self.protocol.recv())
  
          # Cancel transfer_data_task (again, implementation detail).
          self.protocol.fail_connection()
          self.run_loop_once()
          self.assertTrue(self.protocol.transfer_data_task.cancelled())
  
          # recv() completes properly.
          self.assertEqual(self.loop.run_until_complete(recv), "café")
  
      def test_recv_prevents_concurrent_calls(self):
          recv = self.loop.create_task(self.protocol.recv())
  
          with self.assertRaises(RuntimeError) as raised:
              self.loop.run_until_complete(self.protocol.recv())
          self.assertEqual(
              str(raised.exception),
              "cannot call recv while another coroutine "
              "is already waiting for the next message",
          )
          recv.cancel()
  
      # Test the send coroutine.
  
      def test_send_text(self):
          self.loop.run_until_complete(self.protocol.send("café"))
          self.assertOneFrameSent(True, OP_TEXT, "café".encode())
  
      def test_send_binary(self):
          self.loop.run_until_complete(self.protocol.send(b"tea"))
          self.assertOneFrameSent(True, OP_BINARY, b"tea")
  
      def test_send_binary_from_bytearray(self):
          self.loop.run_until_complete(self.protocol.send(bytearray(b"tea")))
          self.assertOneFrameSent(True, OP_BINARY, b"tea")
  
      def test_send_binary_from_memoryview(self):
          self.loop.run_until_complete(self.protocol.send(memoryview(b"tea")))
          self.assertOneFrameSent(True, OP_BINARY, b"tea")
  
      def test_send_dict(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.send({"not": "encoded"}))
          self.assertNoFrameSent()
  
      def test_send_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.send(42))
          self.assertNoFrameSent()
  
      def test_send_iterable_text(self):
          self.loop.run_until_complete(self.protocol.send(["ca", "fé"]))
          self.assertFramesSent(
              (False, OP_TEXT, "ca".encode()),
              (False, OP_CONT, "fé".encode()),
              (True, OP_CONT, "".encode()),
          )
  
      def test_send_iterable_binary(self):
          self.loop.run_until_complete(self.protocol.send([b"te", b"a"]))
          self.assertFramesSent(
              (False, OP_BINARY, b"te"), (False, OP_CONT, b"a"), (True, OP_CONT, b"")
          )
  
      def test_send_iterable_binary_from_bytearray(self):
          self.loop.run_until_complete(
              self.protocol.send([bytearray(b"te"), bytearray(b"a")])
          )
          self.assertFramesSent(
              (False, OP_BINARY, b"te"), (False, OP_CONT, b"a"), (True, OP_CONT, b"")
          )
  
      def test_send_iterable_binary_from_memoryview(self):
          self.loop.run_until_complete(
              self.protocol.send([memoryview(b"te"), memoryview(b"a")])
          )
          self.assertFramesSent(
              (False, OP_BINARY, b"te"), (False, OP_CONT, b"a"), (True, OP_CONT, b"")
          )
  
      def test_send_empty_iterable(self):
          self.loop.run_until_complete(self.protocol.send([]))
          self.assertNoFrameSent()
  
      def test_send_iterable_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.send([42]))
          self.assertNoFrameSent()
  
      def test_send_iterable_mixed_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.send(["café", b"tea"]))
          self.assertFramesSent(
              (False, OP_TEXT, "café".encode()),
              (True, OP_CLOSE, Close(CloseCode.INTERNAL_ERROR, "").serialize()),
          )
  
      def test_send_iterable_prevents_concurrent_send(self):
          self.make_drain_slow(2 * MS)
  
          async def send_iterable():
              await self.protocol.send(["ca", "fé"])
  
          async def send_concurrent():
              await asyncio.sleep(MS)
              await self.protocol.send(b"tea")
  
          async def run_concurrently():
              await asyncio.gather(
                  send_iterable(),
                  send_concurrent(),
              )
  
          self.loop.run_until_complete(run_concurrently())
  
          self.assertFramesSent(
              (False, OP_TEXT, "ca".encode()),
              (False, OP_CONT, "fé".encode()),
              (True, OP_CONT, "".encode()),
              (True, OP_BINARY, b"tea"),
          )
  
      def test_send_async_iterable_text(self):
          self.loop.run_until_complete(self.protocol.send(async_iterable(["ca", "fé"])))
          self.assertFramesSent(
              (False, OP_TEXT, "ca".encode()),
              (False, OP_CONT, "fé".encode()),
              (True, OP_CONT, "".encode()),
          )
  
      def test_send_async_iterable_binary(self):
          self.loop.run_until_complete(self.protocol.send(async_iterable([b"te", b"a"])))
          self.assertFramesSent(
              (False, OP_BINARY, b"te"), (False, OP_CONT, b"a"), (True, OP_CONT, b"")
          )
  
      def test_send_async_iterable_binary_from_bytearray(self):
          self.loop.run_until_complete(
              self.protocol.send(async_iterable([bytearray(b"te"), bytearray(b"a")]))
          )
          self.assertFramesSent(
              (False, OP_BINARY, b"te"), (False, OP_CONT, b"a"), (True, OP_CONT, b"")
          )
  
      def test_send_async_iterable_binary_from_memoryview(self):
          self.loop.run_until_complete(
              self.protocol.send(async_iterable([memoryview(b"te"), memoryview(b"a")]))
          )
          self.assertFramesSent(
              (False, OP_BINARY, b"te"), (False, OP_CONT, b"a"), (True, OP_CONT, b"")
          )
  
      def test_send_empty_async_iterable(self):
          self.loop.run_until_complete(self.protocol.send(async_iterable([])))
          self.assertNoFrameSent()
  
      def test_send_async_iterable_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.send(async_iterable([42])))
          self.assertNoFrameSent()
  
      def test_send_async_iterable_mixed_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(
                  self.protocol.send(async_iterable(["café", b"tea"]))
              )
          self.assertFramesSent(
              (False, OP_TEXT, "café".encode()),
              (True, OP_CLOSE, Close(CloseCode.INTERNAL_ERROR, "").serialize()),
          )
  
      def test_send_async_iterable_prevents_concurrent_send(self):
          self.make_drain_slow(2 * MS)
  
          async def send_async_iterable():
              await self.protocol.send(async_iterable(["ca", "fé"]))
  
          async def send_concurrent():
              await asyncio.sleep(MS)
              await self.protocol.send(b"tea")
  
          async def run_concurrently():
              await asyncio.gather(
                  send_async_iterable(),
                  send_concurrent(),
              )
  
          self.loop.run_until_complete(run_concurrently())
  
          self.assertFramesSent(
              (False, OP_TEXT, "ca".encode()),
              (False, OP_CONT, "fé".encode()),
              (True, OP_CONT, "".encode()),
              (True, OP_BINARY, b"tea"),
          )
  
      def test_send_on_closing_connection_local(self):
          close_task = self.half_close_connection_local()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.send("foobar"))
  
          self.assertNoFrameSent()
  
          self.loop.run_until_complete(close_task)  # cleanup
  
      def test_send_on_closing_connection_remote(self):
          self.half_close_connection_remote()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.send("foobar"))
  
          self.assertNoFrameSent()
  
      def test_send_on_closed_connection(self):
          self.close_connection()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.send("foobar"))
  
          self.assertNoFrameSent()
  
      # Test the ping coroutine.
  
      def test_ping_default(self):
          self.loop.run_until_complete(self.protocol.ping())
          # With our testing tools, it's more convenient to extract the expected
          # ping data from the library's internals than from the frame sent.
          ping_data = next(iter(self.protocol.pings))
          self.assertIsInstance(ping_data, bytes)
          self.assertEqual(len(ping_data), 4)
          self.assertOneFrameSent(True, OP_PING, ping_data)
  
      def test_ping_text(self):
          self.loop.run_until_complete(self.protocol.ping("café"))
          self.assertOneFrameSent(True, OP_PING, "café".encode())
  
      def test_ping_binary(self):
          self.loop.run_until_complete(self.protocol.ping(b"tea"))
          self.assertOneFrameSent(True, OP_PING, b"tea")
  
      def test_ping_binary_from_bytearray(self):
          self.loop.run_until_complete(self.protocol.ping(bytearray(b"tea")))
          self.assertOneFrameSent(True, OP_PING, b"tea")
  
      def test_ping_binary_from_memoryview(self):
          self.loop.run_until_complete(self.protocol.ping(memoryview(b"tea")))
          self.assertOneFrameSent(True, OP_PING, b"tea")
  
      def test_ping_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.ping(42))
          self.assertNoFrameSent()
  
      def test_ping_on_closing_connection_local(self):
          close_task = self.half_close_connection_local()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.ping())
  
          self.assertNoFrameSent()
  
          self.loop.run_until_complete(close_task)  # cleanup
  
      def test_ping_on_closing_connection_remote(self):
          self.half_close_connection_remote()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.ping())
  
          self.assertNoFrameSent()
  
      def test_ping_on_closed_connection(self):
          self.close_connection()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.ping())
  
          self.assertNoFrameSent()
  
      # Test the pong coroutine.
  
      def test_pong_default(self):
          self.loop.run_until_complete(self.protocol.pong())
          self.assertOneFrameSent(True, OP_PONG, b"")
  
      def test_pong_text(self):
          self.loop.run_until_complete(self.protocol.pong("café"))
          self.assertOneFrameSent(True, OP_PONG, "café".encode())
  
      def test_pong_binary(self):
          self.loop.run_until_complete(self.protocol.pong(b"tea"))
          self.assertOneFrameSent(True, OP_PONG, b"tea")
  
      def test_pong_binary_from_bytearray(self):
          self.loop.run_until_complete(self.protocol.pong(bytearray(b"tea")))
          self.assertOneFrameSent(True, OP_PONG, b"tea")
  
      def test_pong_binary_from_memoryview(self):
          self.loop.run_until_complete(self.protocol.pong(memoryview(b"tea")))
          self.assertOneFrameSent(True, OP_PONG, b"tea")
  
      def test_pong_type_error(self):
          with self.assertRaises(TypeError):
              self.loop.run_until_complete(self.protocol.pong(42))
          self.assertNoFrameSent()
  
      def test_pong_on_closing_connection_local(self):
          close_task = self.half_close_connection_local()
  
          with self.assertRaises(ConnectionClosed):
              self.loop.run_until_complete(self.protocol.pong())
  
          self.assertNoFrameSent()
  
          self.loop.run_until_complete(close_task)  # cleanup
  
  