content: import asyncio
  import contextlib
  import itertools
  import logging
  import socket
  import sys
  import unittest
  import uuid
  from unittest.mock import Mock, patch
  
  from websockets.asyncio.compatibility import TimeoutError, aiter, anext, asyncio_timeout
  from websockets.asyncio.connection import *
  from websockets.asyncio.connection import broadcast
  from websockets.exceptions import (
      ConcurrencyError,
      ConnectionClosedError,
      ConnectionClosedOK,
  )
  from websockets.frames import CloseCode, Frame, Opcode
  from websockets.protocol import CLIENT, SERVER, Protocol, State
  
  from ..protocol import RecordingProtocol
  from ..utils import MS, alist
  from .connection import InterceptingConnection
  
  
  # Connection implements symmetrical behavior between clients and servers.
  # All tests run on the client side and the server side to validate this.
  
  
  class ClientConnectionTests(unittest.IsolatedAsyncioTestCase):
      LOCAL = CLIENT
      REMOTE = SERVER
  
      async def asyncSetUp(self):
          self.loop = asyncio.get_running_loop()
          socket_, remote_socket = socket.socketpair()
          protocol = Protocol(self.LOCAL)
          remote_protocol = RecordingProtocol(self.REMOTE)
          self.transport, self.connection = await self.loop.create_connection(
              lambda: Connection(protocol, close_timeout=2 * MS),
              sock=socket_,
          )
          _remote_transport, self.remote_connection = await self.loop.create_connection(
              lambda: InterceptingConnection(remote_protocol),
              sock=remote_socket,
          )
  
      async def asyncTearDown(self):
          await self.remote_connection.close()
          await self.connection.close()
  
      # Test helpers built upon RecordingProtocol and InterceptingConnection.
  
      async def wait_for_remote_side(self):
          """Wait for the remote side to process messages."""
          # Two runs of the event loop are required for answering pings.
          await asyncio.sleep(0)
          await asyncio.sleep(0)
  
      async def assertFrameSent(self, frame):
          """Check that a single frame was sent."""
          await self.wait_for_remote_side()
          self.assertEqual(self.remote_connection.protocol.get_frames_rcvd(), [frame])
  
      async def assertFramesSent(self, frames):
          """Check that several frames were sent."""
          await self.wait_for_remote_side()
          self.assertEqual(self.remote_connection.protocol.get_frames_rcvd(), frames)
  
      async def assertNoFrameSent(self):
          """Check that no frame was sent."""
          await self.wait_for_remote_side()
          self.assertEqual(self.remote_connection.protocol.get_frames_rcvd(), [])
  
      @contextlib.asynccontextmanager
      async def delay_frames_rcvd(self, delay):
          """Delay frames before they're received by the connection."""
          with self.remote_connection.delay_frames_sent(delay):
              yield
              await self.wait_for_remote_side()
  
      @contextlib.asynccontextmanager
      async def delay_eof_rcvd(self, delay):
          """Delay EOF before it's received by the connection."""
          with self.remote_connection.delay_eof_sent(delay):
              yield
              await self.wait_for_remote_side()
  
      @contextlib.asynccontextmanager
      async def drop_frames_rcvd(self):
          """Drop frames before they're received by the connection."""
          with self.remote_connection.drop_frames_sent():
              yield
              await self.wait_for_remote_side()
  
      @contextlib.asynccontextmanager
      async def drop_eof_rcvd(self):
          """Drop EOF before it's received by the connection."""
          with self.remote_connection.drop_eof_sent():
              yield
              await self.wait_for_remote_side()
  
      # Test __aenter__ and __aexit__.
  
      async def test_aenter(self):
          """__aenter__ returns the connection itself."""
          async with self.connection as connection:
              self.assertIs(connection, self.connection)
  
      async def test_aexit(self):
          """__aexit__ closes the connection with code 1000."""
          async with self.connection:
              await self.assertNoFrameSent()
          await self.assertFrameSent(Frame(Opcode.CLOSE, b"\x03\xe8"))
  
      async def test_aexit_with_exception(self):
          """__aexit__ with an exception closes the connection with code 1011."""
          with self.assertRaises(RuntimeError):
              async with self.connection:
                  raise RuntimeError
          await self.assertFrameSent(Frame(Opcode.CLOSE, b"\x03\xf3"))
  
      # Test __aiter__.
  
      async def test_aiter_text(self):
          """__aiter__ yields text messages."""
          iterator = aiter(self.connection)
          async with contextlib.aclosing(iterator):
              await self.remote_connection.send("😀")
              self.assertEqual(await anext(iterator), "😀")
              await self.remote_connection.send("😀")
              self.assertEqual(await anext(iterator), "😀")
  
      async def test_aiter_binary(self):
          """__aiter__ yields binary messages."""
          iterator = aiter(self.connection)
          async with contextlib.aclosing(iterator):
              await self.remote_connection.send(b"\x01\x02\xfe\xff")
              self.assertEqual(await anext(iterator), b"\x01\x02\xfe\xff")
              await self.remote_connection.send(b"\x01\x02\xfe\xff")
              self.assertEqual(await anext(iterator), b"\x01\x02\xfe\xff")
  
      async def test_aiter_mixed(self):
          """__aiter__ yields a mix of text and binary messages."""
          iterator = aiter(self.connection)
          async with contextlib.aclosing(iterator):
              await self.remote_connection.send("😀")
              self.assertEqual(await anext(iterator), "😀")
              await self.remote_connection.send(b"\x01\x02\xfe\xff")
              self.assertEqual(await anext(iterator), b"\x01\x02\xfe\xff")
  
      async def test_aiter_connection_closed_ok(self):
          """__aiter__ terminates after a normal closure."""
          iterator = aiter(self.connection)
          async with contextlib.aclosing(iterator):
              await self.remote_connection.close()
              with self.assertRaises(StopAsyncIteration):
                  await anext(iterator)
  
      async def test_aiter_connection_closed_error(self):
          """__aiter__ raises ConnectionClosedError after an error."""
          iterator = aiter(self.connection)
          async with contextlib.aclosing(iterator):
              await self.remote_connection.close(code=CloseCode.INTERNAL_ERROR)
              with self.assertRaises(ConnectionClosedError):
                  await anext(iterator)
  
      # Test recv.
  
      async def test_recv_text(self):
          """recv receives a text message."""
          await self.remote_connection.send("😀")
          self.assertEqual(await self.connection.recv(), "😀")
  
      async def test_recv_binary(self):
          """recv receives a binary message."""
          await self.remote_connection.send(b"\x01\x02\xfe\xff")
          self.assertEqual(await self.connection.recv(), b"\x01\x02\xfe\xff")
  
      async def test_recv_text_as_bytes(self):
          """recv receives a text message as bytes."""
          await self.remote_connection.send("😀")
          self.assertEqual(await self.connection.recv(decode=False), "😀".encode())
  
      async def test_recv_binary_as_text(self):
          """recv receives a binary message as a str."""
          await self.remote_connection.send("😀".encode())
          self.assertEqual(await self.connection.recv(decode=True), "😀")
  
      async def test_recv_fragmented_text(self):
          """recv receives a fragmented text message."""
          await self.remote_connection.send(["😀", "😀"])
          self.assertEqual(await self.connection.recv(), "😀😀")
  
      async def test_recv_fragmented_binary(self):
          """recv receives a fragmented binary message."""
          await self.remote_connection.send([b"\x01\x02", b"\xfe\xff"])
          self.assertEqual(await self.connection.recv(), b"\x01\x02\xfe\xff")
  
      async def test_recv_connection_closed_ok(self):
          """recv raises ConnectionClosedOK after a normal closure."""
          await self.remote_connection.close()
          with self.assertRaises(ConnectionClosedOK):
              await self.connection.recv()
  
      async def test_recv_connection_closed_error(self):
          """recv raises ConnectionClosedError after an error."""
          await self.remote_connection.close(code=CloseCode.INTERNAL_ERROR)
          with self.assertRaises(ConnectionClosedError):
              await self.connection.recv()
  
      async def test_recv_non_utf8_text(self):
          """recv receives a non-UTF-8 text message."""
          await self.remote_connection.send(b"\x01\x02\xfe\xff", text=True)
          with self.assertRaises(ConnectionClosedError) as raised:
              await self.connection.recv()
          self.assertEqual(raised.exception.sent.code, CloseCode.INVALID_DATA)
  
      async def test_recv_during_recv(self):
          """recv raises ConcurrencyError when called concurrently."""
          recv_task = asyncio.create_task(self.connection.recv())
          await asyncio.sleep(0)  # let the event loop start recv_task
          try:
              with self.assertRaises(ConcurrencyError) as raised:
                  await self.connection.recv()
          finally:
              recv_task.cancel()
          self.assertEqual(
              str(raised.exception),
              "cannot call recv while another coroutine "
              "is already running recv or recv_streaming",
          )
  
      async def test_recv_during_recv_streaming(self):
          """recv raises ConcurrencyError when called concurrently with recv_streaming."""
          recv_streaming_task = asyncio.create_task(
              alist(self.connection.recv_streaming())
          )
          await asyncio.sleep(0)  # let the event loop start recv_streaming_task
          try:
              with self.assertRaises(ConcurrencyError) as raised:
                  await self.connection.recv()
          finally:
              recv_streaming_task.cancel()
          self.assertEqual(
              str(raised.exception),
              "cannot call recv while another coroutine "
              "is already running recv or recv_streaming",
          )
  
      async def test_recv_cancellation_before_receiving(self):
          """recv can be canceled before receiving a message."""
          recv_task = asyncio.create_task(self.connection.recv())
          await asyncio.sleep(0)  # let the event loop start recv_task
          recv_task.cancel()
          await asyncio.sleep(0)  # let the event loop cancel recv_task
  
          # Running recv again receives the next message.
          await self.remote_connection.send("😀")
          self.assertEqual(await self.connection.recv(), "😀")
  
      async def test_recv_cancellation_while_receiving(self):
          """recv can be canceled while receiving a fragmented message."""
          gate = asyncio.Event()
  
          async def fragments():
              yield "⏳"
              await gate.wait()
              yield "⌛️"
  
          asyncio.create_task(self.remote_connection.send(fragments()))
          await asyncio.sleep(0)
  
          recv_task = asyncio.create_task(self.connection.recv())
          await asyncio.sleep(0)  # let the event loop start recv_task
          recv_task.cancel()
          await asyncio.sleep(0)  # let the event loop cancel recv_task
  
          gate.set()
  
          # Running recv again receives the complete message.
          self.assertEqual(await self.connection.recv(), "⏳⌛️")
  
      # Test recv_streaming.
  
      async def test_recv_streaming_text(self):
          """recv_streaming receives a text message."""
          await self.remote_connection.send("😀")
          self.assertEqual(
              await alist(self.connection.recv_streaming()),
              ["😀"],
          )
  
      async def test_recv_streaming_binary(self):
          """recv_streaming receives a binary message."""
          await self.remote_connection.send(b"\x01\x02\xfe\xff")
          self.assertEqual(
              await alist(self.connection.recv_streaming()),
              [b"\x01\x02\xfe\xff"],
          )
  
      async def test_recv_streaming_text_as_bytes(self):
          """recv_streaming receives a text message as bytes."""
          await self.remote_connection.send("😀")
          self.assertEqual(
              await alist(self.connection.recv_streaming(decode=False)),
              ["😀".encode()],
          )
  
      async def test_recv_streaming_binary_as_str(self):
          """recv_streaming receives a binary message as a str."""
          await self.remote_connection.send("😀".encode())
          self.assertEqual(
              await alist(self.connection.recv_streaming(decode=True)),
              ["😀"],
          )
  
      async def test_recv_streaming_fragmented_text(self):
          """recv_streaming receives a fragmented text message."""
          await self.remote_connection.send(["😀", "😀"])
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.connection.recv_streaming()),
              ["😀", "😀", ""],
          )
  
      async def test_recv_streaming_fragmented_binary(self):
          """recv_streaming receives a fragmented binary message."""
          await self.remote_connection.send([b"\x01\x02", b"\xfe\xff"])
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.connection.recv_streaming()),
              [b"\x01\x02", b"\xfe\xff", b""],
          )
  
      async def test_recv_streaming_connection_closed_ok(self):
          """recv_streaming raises ConnectionClosedOK after a normal closure."""
          await self.remote_connection.close()
          with self.assertRaises(ConnectionClosedOK):
              async for _ in self.connection.recv_streaming():
                  self.fail("did not raise")
  
      async def test_recv_streaming_connection_closed_error(self):
          """recv_streaming raises ConnectionClosedError after an error."""
          await self.remote_connection.close(code=CloseCode.INTERNAL_ERROR)
          with self.assertRaises(ConnectionClosedError):
              async for _ in self.connection.recv_streaming():
                  self.fail("did not raise")
  
      async def test_recv_streaming_non_utf8_text(self):
          """recv_streaming receives a non-UTF-8 text message."""
          await self.remote_connection.send(b"\x01\x02\xfe\xff", text=True)
          with self.assertRaises(ConnectionClosedError) as raised:
              await alist(self.connection.recv_streaming())
          self.assertEqual(raised.exception.sent.code, CloseCode.INVALID_DATA)
  
      async def test_recv_streaming_during_recv(self):
          """recv_streaming raises ConcurrencyError when called concurrently with recv."""
          recv_task = asyncio.create_task(self.connection.recv())
          await asyncio.sleep(0)  # let the event loop start recv_task
          try:
              with self.assertRaises(ConcurrencyError) as raised:
                  async for _ in self.connection.recv_streaming():
                      self.fail("did not raise")
          finally:
              recv_task.cancel()
          self.assertEqual(
              str(raised.exception),
              "cannot call recv_streaming while another coroutine "
              "is already running recv or recv_streaming",
          )
  
      async def test_recv_streaming_during_recv_streaming(self):
          """recv_streaming raises ConcurrencyError when called concurrently."""
          recv_streaming_task = asyncio.create_task(
              alist(self.connection.recv_streaming())
          )
          await asyncio.sleep(0)  # let the event loop start recv_streaming_task
          try:
              with self.assertRaises(ConcurrencyError) as raised:
                  async for _ in self.connection.recv_streaming():
                      self.fail("did not raise")
          finally:
              recv_streaming_task.cancel()
          self.assertEqual(
              str(raised.exception),
              "cannot call recv_streaming while another coroutine "
              "is already running recv or recv_streaming",
          )
  
      async def test_recv_streaming_cancellation_before_receiving(self):
          """recv_streaming can be canceled before receiving a message."""
          recv_streaming_task = asyncio.create_task(
              alist(self.connection.recv_streaming())
          )
          await asyncio.sleep(0)  # let the event loop start recv_streaming_task
          recv_streaming_task.cancel()
          await asyncio.sleep(0)  # let the event loop cancel recv_streaming_task
  
          # Running recv_streaming again receives the next message.
          await self.remote_connection.send(["😀", "😀"])
          self.assertEqual(
              await alist(self.connection.recv_streaming()),
              ["😀", "😀", ""],
          )
  
      async def test_recv_streaming_cancellation_while_receiving(self):
          """recv_streaming cannot be canceled while receiving a fragmented message."""
          gate = asyncio.Event()
  
          async def fragments():
              yield "⏳"
              await gate.wait()
              yield "⌛️"
  
          asyncio.create_task(self.remote_connection.send(fragments()))
          await asyncio.sleep(0)
  
          recv_streaming_task = asyncio.create_task(
              alist(self.connection.recv_streaming())
          )
          await asyncio.sleep(0)  # let the event loop start recv_streaming_task
          await asyncio.sleep(0)  # experimentally, two runs of the event loop
          await asyncio.sleep(0)  # are needed to receive the first fragment
          recv_streaming_task.cancel()
          await asyncio.sleep(0)  # let the event loop cancel recv_streaming_task
  
          gate.set()
          await asyncio.sleep(0)
  
          # Running recv_streaming again fails.
          with self.assertRaises(ConcurrencyError):
              async for _ in self.connection.recv_streaming():
                  self.fail("did not raise")
  
      # Test send.
  
      async def test_send_text(self):
          """send sends a text message."""
          await self.connection.send("😀")
          self.assertEqual(await self.remote_connection.recv(), "😀")
  
      async def test_send_binary(self):
          """send sends a binary message."""
          await self.connection.send(b"\x01\x02\xfe\xff")
          self.assertEqual(await self.remote_connection.recv(), b"\x01\x02\xfe\xff")
  
      async def test_send_binary_from_str(self):
          """send sends a binary message from a str."""
          await self.connection.send("😀", text=False)
          self.assertEqual(await self.remote_connection.recv(), "😀".encode())
  
      async def test_send_text_from_bytes(self):
          """send sends a text message from bytes."""
          await self.connection.send("😀".encode(), text=True)
          self.assertEqual(await self.remote_connection.recv(), "😀")
  
      async def test_send_fragmented_text(self):
          """send sends a fragmented text message."""
          await self.connection.send(["😀", "😀"])
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              ["😀", "😀", ""],
          )
  
      async def test_send_fragmented_binary(self):
          """send sends a fragmented binary message."""
          await self.connection.send([b"\x01\x02", b"\xfe\xff"])
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              [b"\x01\x02", b"\xfe\xff", b""],
          )
  
      async def test_send_fragmented_binary_from_str(self):
          """send sends a fragmented binary message from a str."""
          await self.connection.send(["😀", "😀"], text=False)
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              ["😀".encode(), "😀".encode(), b""],
          )
  
      async def test_send_fragmented_text_from_bytes(self):
          """send sends a fragmented text message from bytes."""
          await self.connection.send(["😀".encode(), "😀".encode()], text=True)
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              ["😀", "😀", ""],
          )
  
      async def test_send_async_fragmented_text(self):
          """send sends a fragmented text message asynchronously."""
  
          async def fragments():
              yield "😀"
              yield "😀"
  
          await self.connection.send(fragments())
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              ["😀", "😀", ""],
          )
  
      async def test_send_async_fragmented_binary(self):
          """send sends a fragmented binary message asynchronously."""
  
          async def fragments():
              yield b"\x01\x02"
              yield b"\xfe\xff"
  
          await self.connection.send(fragments())
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              [b"\x01\x02", b"\xfe\xff", b""],
          )
  
      async def test_send_async_fragmented_binary_from_str(self):
          """send sends a fragmented binary message from a str asynchronously."""
  
          async def fragments():
              yield "😀"
              yield "😀"
  
          await self.connection.send(fragments(), text=False)
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              ["😀".encode(), "😀".encode(), b""],
          )
  
      async def test_send_async_fragmented_text_from_bytes(self):
          """send sends a fragmented text message from bytes asynchronously."""
  
          async def fragments():
              yield "😀".encode()
              yield "😀".encode()
  
          await self.connection.send(fragments(), text=True)
          # websockets sends an trailing empty fragment. That's an implementation detail.
          self.assertEqual(
              await alist(self.remote_connection.recv_streaming()),
              ["😀", "😀", ""],
          )
  
      async def test_send_connection_closed_ok(self):
          """send raises ConnectionClosedOK after a normal closure."""
          await self.remote_connection.close()
          with self.assertRaises(ConnectionClosedOK):
              await self.connection.send("😀")
  
      async def test_send_connection_closed_error(self):
          """send raises ConnectionClosedError after an error."""
          await self.remote_connection.close(code=CloseCode.INTERNAL_ERROR)
          with self.assertRaises(ConnectionClosedError):
              await self.connection.send("😀")
  
      async def test_send_during_send(self):
          """send waits for a previous call to send to complete."""
          # This test fails if the guard with send_in_progress is removed
          # from send() in the case when message is an AsyncIterable.
          gate = asyncio.Event()
  
          async def fragments():
              yield "⏳"
              await gate.wait()
              yield "⌛️"
  
          asyncio.create_task(self.connection.send(fragments()))
          await asyncio.sleep(0)  # let the event loop start the task
          await self.assertFrameSent(
              Frame(Opcode.TEXT, "⏳".encode(), fin=False),
          )
  
          asyncio.create_task(self.connection.send("✅"))
          await asyncio.sleep(0)  # let the event loop start the task
          await self.assertNoFrameSent()
  
          gate.set()
          await asyncio.sleep(0)  # run the event loop
          await asyncio.sleep(0)  # three times in order
          await asyncio.sleep(0)  # to send three frames
          await self.assertFramesSent(
              [
                  Frame(Opcode.CONT, "⌛️".encode(), fin=False),
                  Frame(Opcode.CONT, b"", fin=True),
                  Frame(Opcode.TEXT, "✅".encode()),
              ]
          )
  
      async def test_send_while_send_blocked(self):
          """send waits for a blocked call to send to complete."""
          # This test fails if the guard with send_in_progress is removed
          # from send() in the case when message is an Iterable.
          self.connection.pause_writing()
  
          asyncio.create_task(self.connection.send(["⏳", "⌛️"]))
          await asyncio.sleep(0)  # let the event loop start the task
          await self.assertFrameSent(
              Frame(Opcode.TEXT, "⏳".encode(), fin=False),
          )
  
          asyncio.create_task(self.connection.send("✅"))
          await asyncio.sleep(0)  # let the event loop start the task
          await self.assertNoFrameSent()
  
          self.connection.resume_writing()
          await asyncio.sleep(0)  # run the event loop
          await asyncio.sleep(0)  # three times in order
          await asyncio.sleep(0)  # to send three frames
          await self.assertFramesSent(
              [
                  Frame(Opcode.CONT, "⌛️".encode(), fin=False),
                  Frame(Opcode.CONT, b"", fin=True),
                  Frame(Opcode.TEXT, "✅".encode()),
              ]
          )
  
      async def test_send_while_send_async_blocked(self):
          """send waits for a blocked call to send to complete."""
          # This test fails if the guard with send_in_progress is removed
          # from send() in the case when message is an AsyncIterable.
          self.connection.pause_writing()
  
          async def fragments():
              yield "⏳"
              yield "⌛️"
  
          asyncio.create_task(self.connection.send(fragments()))
          await asyncio.sleep(0)  # let the event loop start the task
          await self.assertFrameSent(
              Frame(Opcode.TEXT, "⏳".encode(), fin=False),
          )
  
          asyncio.create_task(self.connection.send("✅"))
          await asyncio.sleep(0)  # let the event loop start the task
          await self.assertNoFrameSent()
  
          self.connection.resume_writing()
          await asyncio.sleep(0)  # run the event loop
          await asyncio.sleep(0)  # three times in order
          await asyncio.sleep(0)  # to send three frames
          await self.assertFramesSent(
              [
                  Frame(Opcode.CONT, "⌛️".encode(), fin=False),
                  Frame(Opcode.CONT, b"", fin=True),
                  Frame(Opcode.TEXT, "✅".encode()),
              ]
          )
  
      async def test_send_empty_iterable(self):
          """send does nothing when called with an empty iterable."""
          await self.connection.send([])
          await self.connection.close()
          self.assertEqual(await alist(self.remote_connection), [])
  
      async def test_send_mixed_iterable(self):
          """send raises TypeError when called with an iterable of inconsistent types."""
          with self.assertRaises(TypeError):
              await self.connection.send(["😀", b"\xfe\xff"])
  
      async def test_send_unsupported_iterable(self):
          """send raises TypeError when called with an iterable of unsupported type."""
          with self.assertRaises(TypeError):
              await self.connection.send([None])
  
      async def test_send_empty_async_iterable(self):
          """send does nothing when called with an empty async iterable."""
  
          async def fragments():
              return
              yield  # pragma: no cover
  
          await self.connection.send(fragments())
          await self.connection.close()
          self.assertEqual(await alist(self.remote_connection), [])
  
      async def test_send_mixed_async_iterable(self):
          """send raises TypeError when called with an iterable of inconsistent types."""
  
          async def fragments():
              yield "😀"
              yield b"\xfe\xff"
  
          iterator = fragments()
          async with contextlib.aclosing(iterator):
              with self.assertRaises(TypeError):
                  await self.connection.send(iterator)
  
      async def test_send_unsupported_async_iterable(self):
          """send raises TypeError when called with an iterable of unsupported type."""
  
          async def fragments():
              yield None
  
          iterator = fragments()
          async with contextlib.aclosing(iterator):
              with self.assertRaises(TypeError):
                  await self.connection.send(iterator)
  
      async def test_send_dict(self):
          """send raises TypeError when called with a dict."""
          with self.assertRaises(TypeError):
              await self.connection.send({"type": "object"})
  
      async def test_send_unsupported_type(self):
          """send raises TypeError when called with an unsupported type."""
          with self.assertRaises(TypeError):
              await self.connection.send(None)
  
      # Test close.
  
      async def test_close(self):
          """close sends a close frame."""
          await self.connection.close()
          await self.assertFrameSent(Frame(Opcode.CLOSE, b"\x03\xe8"))
  
      async def test_close_explicit_code_reason(self):
          """close sends a close frame with a given code and reason."""
          await self.connection.close(CloseCode.GOING_AWAY, "bye!")
          await self.assertFrameSent(Frame(Opcode.CLOSE, b"\x03\xe9bye!"))
  
      async def test_close_waits_for_close_frame(self):
          """close waits for a close frame then EOF before returning."""
          t0 = self.loop.time()
          async with self.delay_frames_rcvd(MS), self.delay_eof_rcvd(MS):
              await self.connection.close()
          t1 = self.loop.time()
  
          self.assertEqual(self.connection.state, State.CLOSED)
          self.assertEqual(self.connection.close_code, CloseCode.NORMAL_CLOSURE)
          self.assertGreater(t1 - t0, MS)
  
          with self.assertRaises(ConnectionClosedOK) as raised:
              await self.connection.recv()
  
          exc = raised.exception
          self.assertEqual(str(exc), "sent 1000 (OK); then received 1000 (OK)")
          self.assertIsNone(exc.__cause__)
  
      async def test_close_waits_for_connection_closed(self):
          """close waits for EOF before returning."""
          if self.LOCAL is SERVER:
              self.skipTest("only relevant on the client-side")
  
          t0 = self.loop.time()
          async with self.delay_eof_rcvd(MS):
              await self.connection.close()
          t1 = self.loop.time()
  
          self.assertEqual(self.connection.state, State.CLOSED)
          self.assertEqual(self.connection.close_code, CloseCode.NORMAL_CLOSURE)
          self.assertGreater(t1 - t0, MS)
  
          with self.assertRaises(ConnectionClosedOK) as raised:
              await self.connection.recv()
  
          exc = raised.exception
          self.assertEqual(str(exc), "sent 1000 (OK); then received 1000 (OK)")
          self.assertIsNone(exc.__cause__)
  
      async def test_close_no_timeout_waits_for_close_frame(self):
          """close without timeout waits for a close frame then EOF before returning."""
          self.connection.close_timeout = None
  
          t0 = self.loop.time()
          async with self.delay_frames_rcvd(MS), self.delay_eof_rcvd(MS):
              await self.connection.close()
          t1 = self.loop.time()
  
          self.assertEqual(self.connection.state, State.CLOSED)
          self.assertEqual(self.connection.close_code, CloseCode.NORMAL_CLOSURE)
          self.assertGreater(t1 - t0, MS)
  
          with self.assertRaises(ConnectionClosedOK) as raised:
              await self.connection.recv()
  
          exc = raised.exception
          self.assertEqual(str(exc), "sent 1000 (OK); then received 1000 (OK)")
          self.assertIsNone(exc.__cause__)
  
      async def test_close_no_timeout_waits_for_connection_closed(self):
          """close without timeout waits for EOF before returning."""
          if self.LOCAL is SERVER:
              self.skipTest("only relevant on the client-side")
  
          self.connection.close_timeout = None
  
          t0 = self.loop.time()
          async with self.delay_eof_rcvd(MS):
              await self.connection.close()
          t1 = self.loop.time()
  
          self.assertEqual(self.connection.state, State.CLOSED)
          self.assertEqual(self.connection.close_code, CloseCode.NORMAL_CLOSURE)
          self.assertGreater(t1 - t0, MS)
  
          with self.assertRaises(ConnectionClosedOK) as raised:
              await self.connection.recv()
  
          exc = raised.exception
          self.assertEqual(str(exc), "sent 1000 (OK); then received 1000 (OK)")
          self.assertIsNone(exc.__cause__)
  
      async def test_close_timeout_waiting_for_close_frame(self):
          """close times out if no close frame is received."""
          t0 = self.loop.time()
  