mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
zeekay
75dd1132bd
The wire protocol was the last holdout and it was not cosmetic: the Hanzo Datastore server reads X-Datastore-User, and answers a request bearing X-ClickHouse-User with 401. Verified against the live server, both directions. So a client still sending the old headers is not merely off-brand, it cannot authenticate over HTTP. Everything else follows: types, dialects, error modules, docstrings and doc links now name Datastore, and doc URLs point at docs.hanzo.ai/datastore rather than the upstream vendor's site. Zero occurrences of the old name remain. Still cut from the pinned upstream version, so behaviour is unchanged.
52 lines
2.1 KiB
Python
52 lines
2.1 KiB
Python
import os
|
|
|
|
import pytest
|
|
from urllib3.exceptions import SSLError
|
|
|
|
from datastore_connect import get_client
|
|
from datastore_connect.driver.common import coerce_bool
|
|
from datastore_connect.driver.exceptions import OperationalError
|
|
from tests.helpers import PROJECT_ROOT_DIR
|
|
|
|
# See .docker/datastore/single_node_tls for the server configuration
|
|
cert_dir = f'{PROJECT_ROOT_DIR}/.docker/datastore/single_node_tls/certificates/'
|
|
host = 'server1.datastore.test'
|
|
|
|
|
|
def test_basic_tls():
|
|
if not coerce_bool(os.environ.get('DATASTORE_CONNECT_TEST_TLS', 'False')):
|
|
pytest.skip('TLS tests not enabled')
|
|
client = get_client(interface='https', host=host, port=10843, verify=False)
|
|
assert client.command("SELECT 'insecure'") == 'insecure'
|
|
client.close_connections()
|
|
|
|
client = get_client(interface='https', host=host, port=10843, ca_cert=f'{cert_dir}ca.crt')
|
|
assert client.command("SELECT 'verify_server'") == 'verify_server'
|
|
client.close_connections()
|
|
|
|
try:
|
|
get_client(interface='https', host='localhost', port=10843, ca_cert=f'{cert_dir}ca.crt')
|
|
pytest.fail('Expected TLS exception with a different hostname')
|
|
except OperationalError as ex:
|
|
assert isinstance(ex.__cause__.reason, SSLError) # pylint: disable=no-member
|
|
client.close_connections()
|
|
|
|
try:
|
|
get_client(interface='https', host='localhost', port=10843)
|
|
pytest.fail('Expected TLS exception with a self-signed cert')
|
|
except OperationalError as ex:
|
|
assert isinstance(ex.__cause__.reason, SSLError) # pylint: disable=no-member
|
|
|
|
|
|
def test_mutual_tls():
|
|
if not coerce_bool(os.environ.get('DATASTORE_CONNECT_TEST_TLS', 'False')):
|
|
pytest.skip('TLS tests not enabled')
|
|
client = get_client(interface='https',
|
|
username='cert_user',
|
|
host=host,
|
|
port=10843,
|
|
ca_cert=f'{cert_dir}ca.crt',
|
|
client_cert=f'{cert_dir}client.crt',
|
|
client_cert_key=f'{cert_dir}client.key')
|
|
assert client.command('SELECT user()') == 'cert_user'
|