(self)
| 163 | self.logger.warning(msg) |
| 164 | |
| 165 | def _should_upgrade(self) -> bool: |
| 166 | upgrade = self._get_upgrade() |
| 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() |
no test coverage detected