(self, url, user_agent=None)
| 164 | return self._exception |
| 165 | |
| 166 | def connect(self, url, user_agent=None): |
| 167 | parsed_url = urlparse(url) |
| 168 | path = parsed_url.path + "?" + parsed_url.query |
| 169 | request = websocket.create_handshake_request( |
| 170 | host=parsed_url.hostname, path=path |
| 171 | ) |
| 172 | if user_agent: |
| 173 | request.headers.set("User-Agent", user_agent) |
| 174 | |
| 175 | environment = os.environ.copy() |
| 176 | proxy_options = None |
| 177 | proxy_url = environment.get("HTTP_PROXY") or environment.get( |
| 178 | "HTTPS_PROXY" |
| 179 | ) |
| 180 | no_proxy = environment.get("NO_PROXY", "") |
| 181 | if proxy_url and parsed_url.hostname not in no_proxy: |
| 182 | parsed_proxy_url = urlparse(proxy_url) |
| 183 | logger.debug( |
| 184 | f"Using the following proxy: {parsed_proxy_url.hostname}" |
| 185 | ) |
| 186 | proxy_options = HttpProxyOptions( |
| 187 | host_name=parsed_proxy_url.hostname, |
| 188 | port=parsed_proxy_url.port, |
| 189 | auth_type=HttpProxyAuthenticationType.Basic |
| 190 | if proxy_url |
| 191 | else HttpProxyAuthenticationType.Nothing, |
| 192 | auth_username=parsed_proxy_url.username or None, |
| 193 | auth_password=parsed_proxy_url.password or None, |
| 194 | ) |
| 195 | |
| 196 | self._tls_connection_options.set_server_name(parsed_url.hostname) |
| 197 | websocket.connect( |
| 198 | host=parsed_url.hostname, |
| 199 | handshake_request=request, |
| 200 | port=443, |
| 201 | tls_connection_options=self._tls_connection_options, |
| 202 | proxy_options=proxy_options, |
| 203 | on_connection_setup=self._on_connection, |
| 204 | on_connection_shutdown=self._on_connection_shutdown, |
| 205 | on_incoming_frame_payload=self._on_incoming_frame_payload_data, |
| 206 | on_incoming_frame_complete=self._on_incoming_frame_complete, |
| 207 | ) |
| 208 | |
| 209 | # Wait for the on_connection callback to be called. |
| 210 | self._on_connection_event.wait() |
| 211 | |
| 212 | if not self._websocket: |
| 213 | self.close() |
| 214 | raise self._exception |
| 215 | |
| 216 | def write_data_from_input(self): |
| 217 | try: |
no test coverage detected