mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
zeekay
999d38466d
Imports read `datastore_connect`, distribution `datastore-connect`, so nothing has to import another product's name to talk to Hanzo Datastore. Cut from upstream 0.8.17 — the version this stack already ran — rather than from upstream HEAD, so it is a PURE RENAME with no behaviour change. Forking from HEAD would have dragged versions along with the name (connect alone jumped 0.8.17 -> 1.6.0, a major), which is a different and much riskier change wearing a rename's clothes. X-ClickHouse-* headers are untouched: they are wire protocol the server sends and expects, and a different token shape from the module name.
67 lines
3.2 KiB
Python
67 lines
3.2 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from urllib3 import ProxyManager
|
|
|
|
import datastore_connect
|
|
from tests.integration_tests.conftest import TestConfig
|
|
|
|
|
|
def test_proxies(test_config: TestConfig):
|
|
if not test_config.proxy_address:
|
|
pytest.skip('Proxy address not configured')
|
|
if test_config.port in (8123, 10723):
|
|
client = datastore_connect.get_client(host=test_config.host,
|
|
port=test_config.port,
|
|
username=test_config.username,
|
|
password=test_config.password,
|
|
http_proxy=test_config.proxy_address)
|
|
assert '2' in client.command('SELECT version()')
|
|
client.close()
|
|
|
|
try:
|
|
os.environ['HTTP_PROXY'] = f'http://{test_config.proxy_address}'
|
|
client = datastore_connect.get_client(host=test_config.host,
|
|
port=test_config.port,
|
|
username=test_config.username,
|
|
password=test_config.password)
|
|
assert isinstance(client.http, ProxyManager)
|
|
assert '2' in client.command('SELECT version()')
|
|
client.close()
|
|
|
|
os.environ['no_proxy'] = f'{test_config.host}:{test_config.port}'
|
|
client = datastore_connect.get_client(host=test_config.host,
|
|
port=test_config.port,
|
|
username=test_config.username,
|
|
password=test_config.password)
|
|
assert not isinstance(client.http, ProxyManager)
|
|
assert '2' in client.command('SELECT version()')
|
|
client.close()
|
|
finally:
|
|
os.environ.pop('HTTP_PROXY', None)
|
|
os.environ.pop('no_proxy', None)
|
|
else:
|
|
cert_file = f'{Path(__file__).parent}/proxy_ca_cert.crt'
|
|
client = datastore_connect.get_client(host=test_config.host,
|
|
port=test_config.port,
|
|
username=test_config.username,
|
|
password=test_config.password,
|
|
ca_cert=cert_file,
|
|
https_proxy=test_config.proxy_address)
|
|
assert '2' in client.command('SELECT version()')
|
|
client.close()
|
|
|
|
try:
|
|
os.environ['HTTPS_PROXY'] = f'{test_config.proxy_address}'
|
|
client = datastore_connect.get_client(host=test_config.host,
|
|
port=test_config.port,
|
|
username=test_config.username,
|
|
password=test_config.password,
|
|
ca_cert=cert_file)
|
|
assert isinstance(client.http, ProxyManager)
|
|
assert '2' in client.command('SELECT version()')
|
|
client.close()
|
|
finally:
|
|
os.environ.pop('HTTPS_PROXY', None)
|