(
self,
url: str,
subprotocols: Sequence[str] | None = None,
**kwargs: Any,
)
| 653 | ) |
| 654 | |
| 655 | def websocket_connect( |
| 656 | self, |
| 657 | url: str, |
| 658 | subprotocols: Sequence[str] | None = None, |
| 659 | **kwargs: Any, |
| 660 | ) -> WebSocketTestSession: |
| 661 | url = urljoin("ws://testserver", url) |
| 662 | headers = kwargs.get("headers", {}) |
| 663 | headers.setdefault("connection", "upgrade") |
| 664 | headers.setdefault("sec-websocket-key", "testserver==") |
| 665 | headers.setdefault("sec-websocket-version", "13") |
| 666 | if subprotocols is not None: |
| 667 | headers.setdefault("sec-websocket-protocol", ", ".join(subprotocols)) |
| 668 | kwargs["headers"] = headers |
| 669 | try: |
| 670 | super().request("GET", url, **kwargs) |
| 671 | except _Upgrade as exc: |
| 672 | session = exc.session |
| 673 | else: |
| 674 | raise RuntimeError("Expected WebSocket upgrade") # pragma: no cover |
| 675 | |
| 676 | return session |
| 677 | |
| 678 | def __enter__(self) -> Self: |
| 679 | with contextlib.ExitStack() as stack: |
no test coverage detected