(self, container, params=None)
| 311 | |
| 312 | @check_resource('container') |
| 313 | def _attach_websocket(self, container, params=None): |
| 314 | url = self._url("/containers/{0}/attach/ws", container) |
| 315 | req = requests.Request("POST", url, params=self._attach_params(params)) |
| 316 | full_url = req.prepare().url |
| 317 | full_url = full_url.replace("http://", "ws://", 1) |
| 318 | full_url = full_url.replace("https://", "wss://", 1) |
| 319 | return self._create_websocket_connection(full_url) |
| 320 | |
| 321 | def _create_websocket_connection(self, url): |
| 322 | try: |
no test coverage detected