content: import asyncio
  import contextlib
  import functools
  import http
  import logging
  import platform
  import random
  import re
  import socket
  import ssl
  import sys
  import unittest
  import urllib.error
  import urllib.request
  import warnings
  from unittest.mock import patch
  
  from websockets.asyncio.compatibility import asyncio_timeout
  from websockets.datastructures import Headers
  from websockets.exceptions import (
      ConnectionClosed,
      InvalidHandshake,
      InvalidHeader,
      NegotiationError,
  )
  from websockets.extensions.permessage_deflate import (
      ClientPerMessageDeflateFactory,
      PerMessageDeflate,
      ServerPerMessageDeflateFactory,
  )
  from websockets.frames import CloseCode
  from websockets.http11 import USER_AGENT
  from websockets.legacy.client import *
  from websockets.legacy.exceptions import InvalidStatusCode
  from websockets.legacy.handshake import build_response
  from websockets.legacy.http import read_response
  from websockets.legacy.server import *
  from websockets.protocol import State
  from websockets.uri import parse_uri
  
  from ..extensions.utils import (
      ClientNoOpExtensionFactory,
      NoOpExtension,
      ServerNoOpExtensionFactory,
  )
  from ..utils import CERTIFICATE, MS, temp_unix_socket_path
  from .utils import AsyncioTestCase
  
  
  async def default_handler(ws):
      if ws.path == "/deprecated_attributes":
          await ws.recv()  # delay that allows catching warnings
          await ws.send(repr((ws.host, ws.port, ws.secure)))
      elif ws.path == "/close_timeout":
          await ws.send(repr(ws.close_timeout))
      elif ws.path == "/path":
          await ws.send(str(ws.path))
      elif ws.path == "/headers":
          await ws.send(repr(ws.request_headers))
          await ws.send(repr(ws.response_headers))
      elif ws.path == "/extensions":
          await ws.send(repr(ws.extensions))
      elif ws.path == "/subprotocol":
          await ws.send(repr(ws.subprotocol))
      elif ws.path == "/slow_stop":
          await ws.wait_closed()
          await asyncio.sleep(2 * MS)
      else:
          await ws.send(await ws.recv())
  
  
  async def redirect_request(path, headers, test, status):
      if path == "/absolute_redirect":
          location = get_server_uri(test.server, test.secure, "/")
      elif path == "/relative_redirect":
          location = "/"
      elif path == "/infinite":
          location = get_server_uri(test.server, test.secure, "/infinite")
      elif path == "/force_secure":
          location = get_server_uri(test.server, True, "/")
      elif path == "/force_insecure":
          location = get_server_uri(test.server, False, "/")
      elif path == "/missing_location":
          return status, {}, b""
      else:
          return None
      return status, {"Location": location}, b""
  
  
  @contextlib.contextmanager
  def temp_test_server(test, **kwargs):
      test.start_server(**kwargs)
      try:
          yield
      finally:
          test.stop_server()
  
  
  def temp_test_redirecting_server(test, status=http.HTTPStatus.FOUND, **kwargs):
      process_request = functools.partial(redirect_request, test=test, status=status)
      return temp_test_server(test, process_request=process_request, **kwargs)
  
  
  @contextlib.contextmanager
  def temp_test_client(test, *args, **kwargs):
      test.start_client(*args, **kwargs)
      try:
          yield
      finally:
          test.stop_client()
  
  
  def with_manager(manager, *args, **kwargs):
      """
      Return a decorator that wraps a function with a context manager.
  
      """
  
      def decorate(func):
          @functools.wraps(func)
          def _decorate(self, *_args, **_kwargs):
              with manager(self, *args, **kwargs):
                  return func(self, *_args, **_kwargs)
  
          return _decorate
  
      return decorate
  
  
  def with_server(**kwargs):
      """
      Return a decorator for TestCase methods that starts and stops a server.
  
      """
      return with_manager(temp_test_server, **kwargs)
  
  
  def with_client(*args, **kwargs):
      """
      Return a decorator for TestCase methods that starts and stops a client.
  
      """
      return with_manager(temp_test_client, *args, **kwargs)
  
  
  def get_server_address(server):
      """
      Return an address on which the given server listens.
  
      """
      # Pick a random socket in order to test both IPv4 and IPv6 on systems
      # where both are available. Randomizing tests is usually a bad idea. If
      # needed, either use the first socket, or test separately IPv4 and IPv6.
      server_socket = random.choice(server.sockets)
  
      if server_socket.family == socket.AF_INET6:  # pragma: no cover
          return server_socket.getsockname()[:2]  # (no IPv6 on CI)
      elif server_socket.family == socket.AF_INET:
          return server_socket.getsockname()
      else:  # pragma: no cover
          raise ValueError("expected an IPv6, IPv4, or Unix socket")
  
  
  def get_server_uri(server, secure=False, resource_name="/", user_info=None):
      """
      Return a WebSocket URI for connecting to the given server.
  
      """
      proto = "wss" if secure else "ws"
      user_info = ":".join(user_info) + "@" if user_info else ""
      host, port = get_server_address(server)
      if ":" in host:  # IPv6 address
          host = f"[{host}]"
      return f"{proto}://{user_info}{host}:{port}{resource_name}"
  
  
  class UnauthorizedServerProtocol(WebSocketServerProtocol):
      async def process_request(self, path, request_headers):
          # Test returning headers as a Headers instance (1/3)
          return http.HTTPStatus.UNAUTHORIZED, Headers([("X-Access", "denied")]), b""
  
  
  class ForbiddenServerProtocol(WebSocketServerProtocol):
      async def process_request(self, path, request_headers):
          # Test returning headers as a dict (2/3)
          return http.HTTPStatus.FORBIDDEN, {"X-Access": "denied"}, b""
  
  
  class HealthCheckServerProtocol(WebSocketServerProtocol):
      async def process_request(self, path, request_headers):
          # Test returning headers as a list of pairs (3/3)
          if path == "/__health__/":
              return http.HTTPStatus.OK, [("X-Access", "OK")], b"status = green\n"
  
  
  class ProcessRequestReturningIntProtocol(WebSocketServerProtocol):
      async def process_request(self, path, request_headers):
          assert path == "/__health__/"
          return 200, [], b"OK\n"
  
  
  class SlowOpeningHandshakeProtocol(WebSocketServerProtocol):
      async def process_request(self, path, request_headers):
          await asyncio.sleep(10 * MS)
  
  
  class FooClientProtocol(WebSocketClientProtocol):
      pass
  
  
  class BarClientProtocol(WebSocketClientProtocol):
      pass
  
  
  class ClientServerTestsMixin:
      secure = False
  
      def setUp(self):
          super().setUp()
          self.server = None
  
      def start_server(self, deprecation_warnings=None, **kwargs):
          handler = kwargs.pop("handler", default_handler)
          # Disable compression by default in tests.
          kwargs.setdefault("compression", None)
          # Disable pings by default in tests.
          kwargs.setdefault("ping_interval", None)
  
          # 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 start_server():
              return await serve(handler, "localhost", 0, **kwargs)
  
          with warnings.catch_warnings(record=True) as recorded_warnings:
              warnings.simplefilter("always")
              self.server = self.loop.run_until_complete(start_server())
  
          expected_warnings = [] if deprecation_warnings is None else deprecation_warnings
          self.assertDeprecationWarnings(recorded_warnings, expected_warnings)
  
      def start_client(
          self, resource_name="/", user_info=None, deprecation_warnings=None, **kwargs
      ):
          # Disable compression by default in tests.
          kwargs.setdefault("compression", None)
          # Disable pings by default in tests.
          kwargs.setdefault("ping_interval", None)
  
          secure = kwargs.get("ssl") is not None
          try:
              server_uri = kwargs.pop("uri")
          except KeyError:
              server_uri = get_server_uri(self.server, secure, resource_name, user_info)
  
          # 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 start_client():
              return await connect(server_uri, **kwargs)
  
          with warnings.catch_warnings(record=True) as recorded_warnings:
              warnings.simplefilter("always")
              self.client = self.loop.run_until_complete(start_client())
  
          expected_warnings = [] if deprecation_warnings is None else deprecation_warnings
          self.assertDeprecationWarnings(recorded_warnings, expected_warnings)
  
      def stop_client(self):
          self.loop.run_until_complete(
              asyncio.wait_for(self.client.close_connection_task, timeout=1)
          )
  
      def stop_server(self):
          self.server.close()
          self.loop.run_until_complete(
              asyncio.wait_for(self.server.wait_closed(), timeout=1)
          )
  
      @contextlib.contextmanager
      def temp_server(self, **kwargs):
          with temp_test_server(self, **kwargs):
              yield
  
      @contextlib.contextmanager
      def temp_client(self, *args, **kwargs):
          with temp_test_client(self, *args, **kwargs):
              yield
  
      def make_http_request(self, path="/", headers=None):
          if headers is None:
              headers = {}
  
          # Set url to 'https?://<host>:<port><path>'.
          url = get_server_uri(
              self.server, resource_name=path, secure=self.secure
          ).replace("ws", "http")
  
          request = urllib.request.Request(url=url, headers=headers)
  
          if self.secure:
              open_health_check = functools.partial(
                  urllib.request.urlopen, request, context=self.client_context
              )
          else:
              open_health_check = functools.partial(urllib.request.urlopen, request)
  
          return self.loop.run_in_executor(None, open_health_check)
  
  
  class SecureClientServerTestsMixin(ClientServerTestsMixin):
      secure = True
  
      @property
      def server_context(self):
          ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
          ssl_context.load_cert_chain(CERTIFICATE)
          return ssl_context
  
      @property
      def client_context(self):
          ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
          ssl_context.load_verify_locations(CERTIFICATE)
          return ssl_context
  
      def start_server(self, **kwargs):
          kwargs.setdefault("ssl", self.server_context)
          super().start_server(**kwargs)
  
      def start_client(self, path="/", **kwargs):
          kwargs.setdefault("ssl", self.client_context)
          super().start_client(path, **kwargs)
  
  
  class CommonClientServerTests:
      """
      Mixin that defines most tests but doesn't inherit unittest.TestCase.
  
      Tests are run by the ClientServerTests and SecureClientServerTests subclasses.
  
      """
  
      @with_server()
      @with_client()
      def test_basic(self):
          self.loop.run_until_complete(self.client.send("Hello!"))
          reply = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(reply, "Hello!")
  
      def test_redirect(self):
          redirect_statuses = [
              http.HTTPStatus.MOVED_PERMANENTLY,
              http.HTTPStatus.FOUND,
              http.HTTPStatus.SEE_OTHER,
              http.HTTPStatus.TEMPORARY_REDIRECT,
              http.HTTPStatus.PERMANENT_REDIRECT,
          ]
          for status in redirect_statuses:
              with temp_test_redirecting_server(self, status):
                  with self.temp_client("/absolute_redirect"):
                      self.loop.run_until_complete(self.client.send("Hello!"))
                      reply = self.loop.run_until_complete(self.client.recv())
                      self.assertEqual(reply, "Hello!")
  
      def test_redirect_relative_location(self):
          with temp_test_redirecting_server(self):
              with self.temp_client("/relative_redirect"):
                  self.loop.run_until_complete(self.client.send("Hello!"))
                  reply = self.loop.run_until_complete(self.client.recv())
                  self.assertEqual(reply, "Hello!")
  
      def test_infinite_redirect(self):
          with temp_test_redirecting_server(self):
              with self.assertRaises(InvalidHandshake):
                  with self.temp_client("/infinite"):
                      self.fail("did not raise")
  
      def test_redirect_missing_location(self):
          with temp_test_redirecting_server(self):
              with self.assertRaises(InvalidHeader):
                  with self.temp_client("/missing_location"):
                      self.fail("did not raise")
  
      def test_loop_backwards_compatibility(self):
          with self.temp_server(
              loop=self.loop,
              deprecation_warnings=["remove loop argument"],
          ):
              with self.temp_client(
                  loop=self.loop,
                  deprecation_warnings=["remove loop argument"],
              ):
                  self.loop.run_until_complete(self.client.send("Hello!"))
                  reply = self.loop.run_until_complete(self.client.recv())
                  self.assertEqual(reply, "Hello!")
  
      @with_server()
      def test_explicit_host_port(self):
          uri = get_server_uri(self.server, self.secure)
          wsuri = parse_uri(uri)
  
          # Change host and port to invalid values.
          scheme = "wss" if wsuri.secure else "ws"
          port = 65535 - wsuri.port
          changed_uri = f"{scheme}://example.com:{port}/"
  
          with self.temp_client(uri=changed_uri, host=wsuri.host, port=wsuri.port):
              self.loop.run_until_complete(self.client.send("Hello!"))
              reply = self.loop.run_until_complete(self.client.recv())
              self.assertEqual(reply, "Hello!")
  
      @with_server()
      def test_explicit_socket(self):
          class TrackedSocket(socket.socket):
              def __init__(self, *args, **kwargs):
                  self.used_for_read = False
                  self.used_for_write = False
                  super().__init__(*args, **kwargs)
  
              def recv(self, *args, **kwargs):
                  self.used_for_read = True
                  return super().recv(*args, **kwargs)
  
              def recv_into(self, *args, **kwargs):
                  self.used_for_read = True
                  return super().recv_into(*args, **kwargs)
  
              def send(self, *args, **kwargs):
                  self.used_for_write = True
                  return super().send(*args, **kwargs)
  
          server_socket = [
              sock for sock in self.server.sockets if sock.family == socket.AF_INET
          ][0]
          client_socket = TrackedSocket(socket.AF_INET, socket.SOCK_STREAM)
          client_socket.connect(server_socket.getsockname())
  
          try:
              self.assertFalse(client_socket.used_for_read)
              self.assertFalse(client_socket.used_for_write)
  
              with self.temp_client(sock=client_socket):
                  self.loop.run_until_complete(self.client.send("Hello!"))
                  reply = self.loop.run_until_complete(self.client.recv())
                  self.assertEqual(reply, "Hello!")
  
              self.assertTrue(client_socket.used_for_read)
              self.assertTrue(client_socket.used_for_write)
  
          finally:
              client_socket.close()
  
      @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "this test requires Unix sockets")
      def test_unix_socket(self):
          with temp_unix_socket_path() as path:
              # Like self.start_server() but with unix_serve().
              async def start_server():
                  return await unix_serve(default_handler, path)
  
              self.server = self.loop.run_until_complete(start_server())
  
              try:
                  # Like self.start_client() but with unix_connect()
                  async def start_client():
                      return await unix_connect(path)
  
                  self.client = self.loop.run_until_complete(start_client())
  
                  try:
                      self.loop.run_until_complete(self.client.send("Hello!"))
                      reply = self.loop.run_until_complete(self.client.recv())
                      self.assertEqual(reply, "Hello!")
  
                  finally:
                      self.stop_client()
  
              finally:
                  self.stop_server()
  
      def test_ws_handler_argument_backwards_compatibility(self):
          async def handler_with_path(ws, path):
              await ws.send(path)
  
          with self.temp_server(
              handler=handler_with_path,
              deprecation_warnings=["remove second argument of ws_handler"],
          ):
              with self.temp_client("/path"):
                  self.assertEqual(
                      self.loop.run_until_complete(self.client.recv()),
                      "/path",
                  )
  
      def test_ws_handler_argument_backwards_compatibility_partial(self):
          async def handler_with_path(ws, path, extra):
              await ws.send(path)
  
          bound_handler_with_path = functools.partial(handler_with_path, extra=None)
  
          with self.temp_server(
              handler=bound_handler_with_path,
              deprecation_warnings=["remove second argument of ws_handler"],
          ):
              with self.temp_client("/path"):
                  self.assertEqual(
                      self.loop.run_until_complete(self.client.recv()),
                      "/path",
                  )
  
      async def process_request_OK(path, request_headers):
          return http.HTTPStatus.OK, [], b"OK\n"
  
      @with_server(process_request=process_request_OK)
      def test_process_request_argument(self):
          response = self.loop.run_until_complete(self.make_http_request("/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.code, 200)
  
      def legacy_process_request_OK(path, request_headers):
          return http.HTTPStatus.OK, [], b"OK\n"
  
      @with_server(process_request=legacy_process_request_OK)
      def test_process_request_argument_backwards_compatibility(self):
          with warnings.catch_warnings(record=True) as recorded_warnings:
              warnings.simplefilter("always")
              response = self.loop.run_until_complete(self.make_http_request("/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.code, 200)
  
          self.assertDeprecationWarnings(
              recorded_warnings, ["declare process_request as a coroutine"]
          )
  
      class ProcessRequestOKServerProtocol(WebSocketServerProtocol):
          async def process_request(self, path, request_headers):
              return http.HTTPStatus.OK, [], b"OK\n"
  
      @with_server(create_protocol=ProcessRequestOKServerProtocol)
      def test_process_request_override(self):
          response = self.loop.run_until_complete(self.make_http_request("/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.code, 200)
  
      class LegacyProcessRequestOKServerProtocol(WebSocketServerProtocol):
          def process_request(self, path, request_headers):
              return http.HTTPStatus.OK, [], b"OK\n"
  
      @with_server(create_protocol=LegacyProcessRequestOKServerProtocol)
      def test_process_request_override_backwards_compatibility(self):
          with warnings.catch_warnings(record=True) as recorded_warnings:
              warnings.simplefilter("always")
              response = self.loop.run_until_complete(self.make_http_request("/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.code, 200)
  
          self.assertDeprecationWarnings(
              recorded_warnings, ["declare process_request as a coroutine"]
          )
  
      def select_subprotocol_chat(client_subprotocols, server_subprotocols):
          return "chat"
  
      @with_server(
          subprotocols=["superchat", "chat"], select_subprotocol=select_subprotocol_chat
      )
      @with_client("/subprotocol", subprotocols=["superchat", "chat"])
      def test_select_subprotocol_argument(self):
          server_subprotocol = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(server_subprotocol, repr("chat"))
          self.assertEqual(self.client.subprotocol, "chat")
  
      class SelectSubprotocolChatServerProtocol(WebSocketServerProtocol):
          def select_subprotocol(self, client_subprotocols, server_subprotocols):
              return "chat"
  
      @with_server(
          subprotocols=["superchat", "chat"],
          create_protocol=SelectSubprotocolChatServerProtocol,
      )
      @with_client("/subprotocol", subprotocols=["superchat", "chat"])
      def test_select_subprotocol_override(self):
          server_subprotocol = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(server_subprotocol, repr("chat"))
          self.assertEqual(self.client.subprotocol, "chat")
  
      @with_server()
      @with_client("/deprecated_attributes")
      def test_protocol_deprecated_attributes(self):
          # The test could be connecting with IPv6 or IPv4.
          expected_client_attrs = [
              server_socket.getsockname()[:2] + (self.secure,)
              for server_socket in self.server.sockets
          ]
          with warnings.catch_warnings(record=True) as recorded_warnings:
              warnings.simplefilter("always")
              client_attrs = (self.client.host, self.client.port, self.client.secure)
          self.assertDeprecationWarnings(
              recorded_warnings,
              [
                  "use remote_address[0] instead of host",
                  "use remote_address[1] instead of port",
                  "don't use secure",
              ],
          )
          self.assertIn(client_attrs, expected_client_attrs)
  
          expected_server_attrs = ("localhost", 0, self.secure)
          with warnings.catch_warnings(record=True) as recorded_warnings:
              warnings.simplefilter("always")
              self.loop.run_until_complete(self.client.send(""))
              server_attrs = self.loop.run_until_complete(self.client.recv())
          self.assertDeprecationWarnings(
              recorded_warnings,
              [
                  "use local_address[0] instead of host",
                  "use local_address[1] instead of port",
                  "don't use secure",
              ],
          )
          self.assertEqual(server_attrs, repr(expected_server_attrs))
  
      @with_server()
      @with_client("/path")
      def test_protocol_path(self):
          client_path = self.client.path
          self.assertEqual(client_path, "/path")
          server_path = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(server_path, "/path")
  
      @with_server()
      @with_client("/headers")
      def test_protocol_headers(self):
          client_req = self.client.request_headers
          client_resp = self.client.response_headers
          self.assertEqual(client_req["User-Agent"], USER_AGENT)
          self.assertEqual(client_resp["Server"], USER_AGENT)
          server_req = self.loop.run_until_complete(self.client.recv())
          server_resp = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(server_req, repr(client_req))
          self.assertEqual(server_resp, repr(client_resp))
  
      @with_server()
      @with_client("/headers", extra_headers={"X-Spam": "Eggs"})
      def test_protocol_custom_request_headers(self):
          req_headers = self.loop.run_until_complete(self.client.recv())
          self.loop.run_until_complete(self.client.recv())
          self.assertIn("('X-Spam', 'Eggs')", req_headers)
  
      @with_server()
      @with_client("/headers", extra_headers={"User-Agent": "websockets"})
      def test_protocol_custom_user_agent_header_legacy(self):
          req_headers = self.loop.run_until_complete(self.client.recv())
          self.loop.run_until_complete(self.client.recv())
          self.assertEqual(req_headers.count("User-Agent"), 1)
          self.assertIn("('User-Agent', 'websockets')", req_headers)
  
      @with_server()
      @with_client("/headers", user_agent_header=None)
      def test_protocol_no_user_agent_header(self):
          req_headers = self.loop.run_until_complete(self.client.recv())
          self.loop.run_until_complete(self.client.recv())
          self.assertNotIn("User-Agent", req_headers)
  
      @with_server()
      @with_client("/headers", user_agent_header="websockets")
      def test_protocol_custom_user_agent_header(self):
          req_headers = self.loop.run_until_complete(self.client.recv())
          self.loop.run_until_complete(self.client.recv())
          self.assertEqual(req_headers.count("User-Agent"), 1)
          self.assertIn("('User-Agent', 'websockets')", req_headers)
  
      @with_server(extra_headers=lambda p, r: {"X-Spam": "Eggs"})
      @with_client("/headers")
      def test_protocol_custom_response_headers_callable(self):
          self.loop.run_until_complete(self.client.recv())
          resp_headers = self.loop.run_until_complete(self.client.recv())
          self.assertIn("('X-Spam', 'Eggs')", resp_headers)
  
      @with_server(extra_headers=lambda p, r: None)
      @with_client("/headers")
      def test_protocol_custom_response_headers_callable_none(self):
          self.loop.run_until_complete(self.client.recv())  # doesn't crash
          self.loop.run_until_complete(self.client.recv())  # nothing to check
  
      @with_server(extra_headers={"X-Spam": "Eggs"})
      @with_client("/headers")
      def test_protocol_custom_response_headers(self):
          self.loop.run_until_complete(self.client.recv())
          resp_headers = self.loop.run_until_complete(self.client.recv())
          self.assertIn("('X-Spam', 'Eggs')", resp_headers)
  
      @with_server(extra_headers={"Server": "websockets"})
      @with_client("/headers")
      def test_protocol_custom_server_header_legacy(self):
          self.loop.run_until_complete(self.client.recv())
          resp_headers = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(resp_headers.count("Server"), 1)
          self.assertIn("('Server', 'websockets')", resp_headers)
  
      @with_server(server_header=None)
      @with_client("/headers")
      def test_protocol_no_server_header(self):
          self.loop.run_until_complete(self.client.recv())
          resp_headers = self.loop.run_until_complete(self.client.recv())
          self.assertNotIn("Server", resp_headers)
  
      @with_server(server_header="websockets")
      @with_client("/headers")
      def test_protocol_custom_server_header(self):
          self.loop.run_until_complete(self.client.recv())
          resp_headers = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(resp_headers.count("Server"), 1)
          self.assertIn("('Server', 'websockets')", resp_headers)
  
      @with_server(create_protocol=HealthCheckServerProtocol)
      def test_http_request_http_endpoint(self):
          # Making an HTTP request to an HTTP endpoint succeeds.
          response = self.loop.run_until_complete(self.make_http_request("/__health__/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.code, 200)
              self.assertEqual(response.read(), b"status = green\n")
  
      @with_server(create_protocol=HealthCheckServerProtocol)
      def test_http_request_ws_endpoint(self):
          # Making an HTTP request to a WS endpoint fails.
          with self.assertRaises(urllib.error.HTTPError) as raised:
              self.loop.run_until_complete(self.make_http_request())
  
          with contextlib.closing(raised.exception):
              self.assertEqual(raised.exception.code, 426)
              self.assertEqual(raised.exception.headers["Upgrade"], "websocket")
  
      @with_server(create_protocol=HealthCheckServerProtocol)
      def test_ws_connection_http_endpoint(self):
          # Making a WS connection to an HTTP endpoint fails.
          with self.assertRaises(InvalidStatusCode) as raised:
              self.start_client("/__health__/")
  
          self.assertEqual(raised.exception.status_code, 200)
  
      @with_server(create_protocol=HealthCheckServerProtocol)
      def test_ws_connection_ws_endpoint(self):
          # Making a WS connection to a WS endpoint succeeds.
          self.start_client()
          self.loop.run_until_complete(self.client.send("Hello!"))
          self.loop.run_until_complete(self.client.recv())
          self.stop_client()
  
      @with_server(create_protocol=HealthCheckServerProtocol, server_header=None)
      def test_http_request_no_server_header(self):
          response = self.loop.run_until_complete(self.make_http_request("/__health__/"))
  
          with contextlib.closing(response):
              self.assertNotIn("Server", response.headers)
  
      @with_server(create_protocol=HealthCheckServerProtocol, server_header="websockets")
      def test_http_request_custom_server_header(self):
          response = self.loop.run_until_complete(self.make_http_request("/__health__/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.headers["Server"], "websockets")
  
      @with_server(create_protocol=ProcessRequestReturningIntProtocol)
      def test_process_request_returns_int_status(self):
          response = self.loop.run_until_complete(self.make_http_request("/__health__/"))
  
          with contextlib.closing(response):
              self.assertEqual(response.code, 200)
              self.assertEqual(response.read(), b"OK\n")
  
      def assert_client_raises_code(self, status_code):
          with self.assertRaises(InvalidStatusCode) as raised:
              self.start_client()
          self.assertEqual(raised.exception.status_code, status_code)
  
      @with_server(create_protocol=UnauthorizedServerProtocol)
      def test_server_create_protocol(self):
          self.assert_client_raises_code(401)
  
      def create_unauthorized_server_protocol(*args, **kwargs):
          return UnauthorizedServerProtocol(*args, **kwargs)
  
      @with_server(create_protocol=create_unauthorized_server_protocol)
      def test_server_create_protocol_function(self):
          self.assert_client_raises_code(401)
  
      @with_server(
          klass=UnauthorizedServerProtocol,
          deprecation_warnings=["rename klass to create_protocol"],
      )
      def test_server_klass_backwards_compatibility(self):
          self.assert_client_raises_code(401)
  
      @with_server(
          create_protocol=ForbiddenServerProtocol,
          klass=UnauthorizedServerProtocol,
          deprecation_warnings=["rename klass to create_protocol"],
      )
      def test_server_create_protocol_over_klass(self):
          self.assert_client_raises_code(403)
  
      @with_server()
      @with_client("/path", create_protocol=FooClientProtocol)
      def test_client_create_protocol(self):
          self.assertIsInstance(self.client, FooClientProtocol)
  
      @with_server()
      @with_client(
          "/path",
          create_protocol=(lambda *args, **kwargs: FooClientProtocol(*args, **kwargs)),
      )
      def test_client_create_protocol_function(self):
          self.assertIsInstance(self.client, FooClientProtocol)
  
      @with_server()
      @with_client(
          "/path",
          klass=FooClientProtocol,
          deprecation_warnings=["rename klass to create_protocol"],
      )
      def test_client_klass(self):
          self.assertIsInstance(self.client, FooClientProtocol)
  
      @with_server()
      @with_client(
          "/path",
          create_protocol=BarClientProtocol,
          klass=FooClientProtocol,
          deprecation_warnings=["rename klass to create_protocol"],
      )
      def test_client_create_protocol_over_klass(self):
          self.assertIsInstance(self.client, BarClientProtocol)
  
      @with_server(close_timeout=7)
      @with_client("/close_timeout")
      def test_server_close_timeout(self):
          close_timeout = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(eval(close_timeout), 7)
  
      @with_server(timeout=6, deprecation_warnings=["rename timeout to close_timeout"])
      @with_client("/close_timeout")
      def test_server_timeout_backwards_compatibility(self):
          close_timeout = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(eval(close_timeout), 6)
  
      @with_server(
          close_timeout=7,
          timeout=6,
          deprecation_warnings=["rename timeout to close_timeout"],
      )
      @with_client("/close_timeout")
      def test_server_close_timeout_over_timeout(self):
          close_timeout = self.loop.run_until_complete(self.client.recv())
          self.assertEqual(eval(close_timeout), 7)
  
      @with_server()
      @with_client("/close_timeout", close_timeout=7)
      def test_client_close_timeout(self):
          self.assertEqual(self.client.close_timeout, 7)
  
      @with_server()
      @with_client(
          "/close_timeout",
          timeout=6,
          deprecation_warnings=["rename timeout to close_timeout"],
      )
      def test_client_timeout_backwards_compatibility(self):
          self.assertEqual(self.client.close_timeout, 6)
  
  