(self, data: bytes)
| 167 | return upgrade == b"websocket" and self._should_upgrade_to_ws() |
| 168 | |
| 169 | def data_received(self, data: bytes) -> None: |
| 170 | self._unset_keepalive_if_required() |
| 171 | |
| 172 | try: |
| 173 | self.parser.feed_data(data) |
| 174 | except httptools.HttpParserError: |
| 175 | msg = "Invalid HTTP request received." |
| 176 | self.logger.warning(msg) |
| 177 | self.send_400_response(msg) |
| 178 | return |
| 179 | except httptools.HttpParserUpgrade: |
| 180 | if self._should_upgrade(): |
| 181 | self.handle_websocket_upgrade() |
| 182 | else: |
| 183 | self._unsupported_upgrade_warning() |
| 184 | |
| 185 | def handle_websocket_upgrade(self) -> None: |
| 186 | if self.logger.level <= TRACE_LOG_LEVEL: |
no test coverage detected