(
self,
url: str,
subprotocols: Sequence[str] | None = None,
**kwargs: Any,
)
| 477 | ) |
| 478 | |
| 479 | def websocket_connect( |
| 480 | self, |
| 481 | url: str, |
| 482 | subprotocols: Sequence[str] | None = None, |
| 483 | **kwargs: Any, |
| 484 | ) -> WebSocketTestSession: |
| 485 | url = urljoin("ws://testserver", url) |
| 486 | headers = kwargs.get("headers", {}) |
| 487 | headers.setdefault("connection", "upgrade") |
| 488 | headers.setdefault("sec-websocket-key", "testserver==") |
| 489 | headers.setdefault("sec-websocket-version", "13") |
| 490 | if subprotocols is not None: |
| 491 | headers.setdefault("sec-websocket-protocol", ", ".join(subprotocols)) |
| 492 | kwargs["headers"] = headers |
| 493 | try: |
| 494 | super().request("GET", url, **kwargs) |
| 495 | except _Upgrade as exc: |
| 496 | session = exc.session |
| 497 | else: |
| 498 | raise RuntimeError("Expected WebSocket upgrade") # pragma: no cover |
| 499 | |
| 500 | return session |
| 501 | |
| 502 | def __enter__(self) -> Self: |
| 503 | with contextlib.ExitStack() as stack: |
no test coverage detected