(self)
| 161 | self.logger.warning(msg) |
| 162 | |
| 163 | def _should_upgrade(self) -> bool: |
| 164 | upgrade = self._get_upgrade() |
| 165 | if upgrade == b"websocket" and self._should_upgrade_to_ws(): |
| 166 | return True |
| 167 | if upgrade is not None: |
| 168 | self._unsupported_upgrade_warning() |
| 169 | return False |
| 170 | |
| 171 | def data_received(self, data: bytes) -> None: |
| 172 | self._unset_keepalive_if_required() |
no test coverage detected