(self)
| 13 | self.stream = asyncio.StreamReader(loop=self.loop) |
| 14 | |
| 15 | async def test_read_request(self): |
| 16 | # Example from the protocol overview in RFC 6455 |
| 17 | self.stream.feed_data( |
| 18 | b"GET /chat HTTP/1.1\r\n" |
| 19 | b"Host: server.example.com\r\n" |
| 20 | b"Upgrade: websocket\r\n" |
| 21 | b"Connection: Upgrade\r\n" |
| 22 | b"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" |
| 23 | b"Origin: http://example.com\r\n" |
| 24 | b"Sec-WebSocket-Protocol: chat, superchat\r\n" |
| 25 | b"Sec-WebSocket-Version: 13\r\n" |
| 26 | b"\r\n" |
| 27 | ) |
| 28 | path, headers = await read_request(self.stream) |
| 29 | self.assertEqual(path, "/chat") |
| 30 | self.assertEqual(headers["Upgrade"], "websocket") |
| 31 | |
| 32 | async def test_read_request_empty(self): |
| 33 | self.stream.feed_eof() |
nothing calls this directly
no test coverage detected