(self)
| 75 | ) |
| 76 | |
| 77 | async def test_read_response(self): |
| 78 | # Example from the protocol overview in RFC 6455 |
| 79 | self.stream.feed_data( |
| 80 | b"HTTP/1.1 101 Switching Protocols\r\n" |
| 81 | b"Upgrade: websocket\r\n" |
| 82 | b"Connection: Upgrade\r\n" |
| 83 | b"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n" |
| 84 | b"Sec-WebSocket-Protocol: chat\r\n" |
| 85 | b"\r\n" |
| 86 | ) |
| 87 | status_code, reason, headers = await read_response(self.stream) |
| 88 | self.assertEqual(status_code, 101) |
| 89 | self.assertEqual(reason, "Switching Protocols") |
| 90 | self.assertEqual(headers["Upgrade"], "websocket") |
| 91 | |
| 92 | async def test_read_response_empty(self): |
| 93 | self.stream.feed_eof() |
nothing calls this directly
no test coverage detected