content: import asyncio
  
  from websockets.exceptions import SecurityError
  from websockets.legacy.http import *
  from websockets.legacy.http import read_headers
  
  from .utils import AsyncioTestCase
  
  
  class HTTPAsyncTests(AsyncioTestCase):
      def setUp(self):
          super().setUp()
          self.stream = asyncio.StreamReader(loop=self.loop)
  
      async def test_read_request(self):
          # Example from the protocol overview in RFC 6455
          self.stream.feed_data(
              b"GET /chat HTTP/1.1\r\n"
              b"Host: server.example.com\r\n"
              b"Upgrade: websocket\r\n"
              b"Connection: Upgrade\r\n"
              b"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
              b"Origin: http://example.com\r\n"
              b"Sec-WebSocket-Protocol: chat, superchat\r\n"
              b"Sec-WebSocket-Version: 13\r\n"
              b"\r\n"
          )
          path, headers = await read_request(self.stream)
          self.assertEqual(path, "/chat")
          self.assertEqual(headers["Upgrade"], "websocket")
  
      async def test_read_request_empty(self):
          self.stream.feed_eof()
          with self.assertRaises(EOFError) as raised:
              await read_request(self.stream)
          self.assertEqual(
              str(raised.exception),
              "connection closed while reading HTTP request line",
          )
  
      async def test_read_request_invalid_request_line(self):
          self.stream.feed_data(b"GET /\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_request(self.stream)
          self.assertEqual(
              str(raised.exception),
              "invalid HTTP request line: GET /",
          )
  
      async def test_read_request_unsupported_method(self):
          self.stream.feed_data(b"OPTIONS * HTTP/1.1\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_request(self.stream)
          self.assertEqual(
              str(raised.exception),
              "unsupported HTTP method: OPTIONS",
          )
  
      async def test_read_request_unsupported_version(self):
          self.stream.feed_data(b"GET /chat HTTP/1.0\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_request(self.stream)
          self.assertEqual(
              str(raised.exception),
              "unsupported HTTP version: HTTP/1.0",
          )
  
      async def test_read_request_invalid_header(self):
          self.stream.feed_data(b"GET /chat HTTP/1.1\r\nOops\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_request(self.stream)
          self.assertEqual(
              str(raised.exception),
              "invalid HTTP header line: Oops",
          )
  
      async def test_read_response(self):
          # Example from the protocol overview in RFC 6455
          self.stream.feed_data(
              b"HTTP/1.1 101 Switching Protocols\r\n"
              b"Upgrade: websocket\r\n"
              b"Connection: Upgrade\r\n"
              b"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n"
              b"Sec-WebSocket-Protocol: chat\r\n"
              b"\r\n"
          )
          status_code, reason, headers = await read_response(self.stream)
          self.assertEqual(status_code, 101)
          self.assertEqual(reason, "Switching Protocols")
          self.assertEqual(headers["Upgrade"], "websocket")
  
      async def test_read_response_empty(self):
          self.stream.feed_eof()
          with self.assertRaises(EOFError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "connection closed while reading HTTP status line",
          )
  
      async def test_read_request_invalid_status_line(self):
          self.stream.feed_data(b"Hello!\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "invalid HTTP status line: Hello!",
          )
  
      async def test_read_response_unsupported_version(self):
          self.stream.feed_data(b"HTTP/1.0 400 Bad Request\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "unsupported HTTP version: HTTP/1.0",
          )
  
      async def test_read_response_invalid_status(self):
          self.stream.feed_data(b"HTTP/1.1 OMG WTF\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "invalid HTTP status code: OMG",
          )
  
      async def test_read_response_unsupported_status(self):
          self.stream.feed_data(b"HTTP/1.1 007 My name is Bond\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "unsupported HTTP status code: 007",
          )
  
      async def test_read_response_invalid_reason(self):
          self.stream.feed_data(b"HTTP/1.1 200 \x7f\r\n\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "invalid HTTP reason phrase: \x7f",
          )
  
      async def test_read_response_invalid_header(self):
          self.stream.feed_data(b"HTTP/1.1 500 Internal Server Error\r\nOops\r\n")
          with self.assertRaises(ValueError) as raised:
              await read_response(self.stream)
          self.assertEqual(
              str(raised.exception),
              "invalid HTTP header line: Oops",
          )
  
      async def test_header_name(self):
          self.stream.feed_data(b"foo bar: baz qux\r\n\r\n")
          with self.assertRaises(ValueError):
              await read_headers(self.stream)
  
      async def test_header_value(self):
          self.stream.feed_data(b"foo: \x00\x00\x0f\r\n\r\n")
          with self.assertRaises(ValueError):
              await read_headers(self.stream)
  
      async def test_headers_limit(self):
          self.stream.feed_data(b"foo: bar\r\n" * 129 + b"\r\n")
          with self.assertRaises(SecurityError):
              await read_headers(self.stream)
  
      async def test_line_limit(self):
          # Header line contains 5 + 8186 + 2 = 8193 bytes.
          self.stream.feed_data(b"foo: " + b"a" * 8186 + b"\r\n\r\n")
          with self.assertRaises(SecurityError):
              await read_headers(self.stream)
  
      async def test_line_ending(self):
          self.stream.feed_data(b"foo: bar\n\n")
          with self.assertRaises(EOFError):
              await read_headers(self.stream)
  