(configuration, url, headers=None)
| 507 | |
| 508 | |
| 509 | def create_websocket(configuration, url, headers=None): |
| 510 | enableTrace(False) |
| 511 | |
| 512 | # We just need to pass the Authorization, ignore all the other |
| 513 | # http headers we get from the generated code |
| 514 | header = [] |
| 515 | if headers and 'authorization' in headers: |
| 516 | header.append("authorization: %s" % headers['authorization']) |
| 517 | if headers and 'sec-websocket-protocol' in headers: |
| 518 | header.append("sec-websocket-protocol: %s" % |
| 519 | headers['sec-websocket-protocol']) |
| 520 | else: |
| 521 | header.append("sec-websocket-protocol: %s,%s" % (V5_CHANNEL_PROTOCOL, V4_CHANNEL_PROTOCOL)) |
| 522 | |
| 523 | if url.startswith('wss://') and configuration.verify_ssl: |
| 524 | ssl_opts = { |
| 525 | 'cert_reqs': ssl.CERT_REQUIRED, |
| 526 | 'ca_certs': configuration.ssl_ca_cert or certifi.where(), |
| 527 | } |
| 528 | if configuration.assert_hostname is not None: |
| 529 | ssl_opts['check_hostname'] = configuration.assert_hostname |
| 530 | else: |
| 531 | ssl_opts = {'cert_reqs': ssl.CERT_NONE} |
| 532 | |
| 533 | if configuration.cert_file: |
| 534 | ssl_opts['certfile'] = configuration.cert_file |
| 535 | if configuration.key_file: |
| 536 | ssl_opts['keyfile'] = configuration.key_file |
| 537 | if configuration.tls_server_name: |
| 538 | ssl_opts['server_hostname'] = configuration.tls_server_name |
| 539 | |
| 540 | websocket = WebSocket(sslopt=ssl_opts, skip_utf8_validation=False) |
| 541 | connect_opt = { |
| 542 | 'header': header |
| 543 | } |
| 544 | |
| 545 | if configuration.proxy or configuration.proxy_headers: |
| 546 | connect_opt = websocket_proxycare(connect_opt, configuration, url, headers) |
| 547 | |
| 548 | websocket.connect(url, **connect_opt) |
| 549 | return websocket |
| 550 | |
| 551 | def websocket_proxycare(connect_opt, configuration, url, headers): |
| 552 | """ An internal function to be called in api-client when a websocket |
no test coverage detected