()
| 479 | return True |
| 480 | |
| 481 | def check() -> bool: |
| 482 | if self.ping_timeout: |
| 483 | has_timeout_expired = ( |
| 484 | time.time() - self.last_ping_tm > self.ping_timeout |
| 485 | ) |
| 486 | has_pong_not_arrived_after_last_ping = ( |
| 487 | self.last_pong_tm - self.last_ping_tm < 0 |
| 488 | ) |
| 489 | has_pong_arrived_too_late = ( |
| 490 | self.last_pong_tm - self.last_ping_tm > self.ping_timeout |
| 491 | ) |
| 492 | |
| 493 | if ( |
| 494 | self.last_ping_tm |
| 495 | and has_timeout_expired |
| 496 | and ( |
| 497 | has_pong_not_arrived_after_last_ping |
| 498 | or has_pong_arrived_too_late |
| 499 | ) |
| 500 | ): |
| 501 | raise WebSocketTimeoutException("ping/pong timed out") |
| 502 | return True |
| 503 | |
| 504 | def closed( |
| 505 | e: Union[ |
nothing calls this directly
no test coverage detected