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.
79 lines
3.1 KiB
Python
79 lines
3.1 KiB
Python
import uuid
|
|
|
|
from typing import Callable
|
|
|
|
import pytest
|
|
|
|
from datastore_connect.driver import create_async_client
|
|
from tests.integration_tests.conftest import TestConfig
|
|
|
|
SESSION_KEY = 'session_id'
|
|
|
|
|
|
def test_client_default_session_id(test_create_client: Callable):
|
|
# by default, the sync client will autogenerate the session id
|
|
client = test_create_client()
|
|
session_id = client.get_client_setting(SESSION_KEY)
|
|
try:
|
|
uuid.UUID(session_id)
|
|
except ValueError:
|
|
pytest.fail(f"Invalid session_id: {session_id}")
|
|
client.close()
|
|
|
|
|
|
def test_client_autogenerate_session_id(test_create_client: Callable):
|
|
client = test_create_client()
|
|
session_id = client.get_client_setting(SESSION_KEY)
|
|
try:
|
|
uuid.UUID(session_id)
|
|
except ValueError:
|
|
pytest.fail(f"Invalid session_id: {session_id}")
|
|
|
|
|
|
def test_client_custom_session_id(test_create_client: Callable):
|
|
session_id = 'custom_session_id'
|
|
client = test_create_client(session_id=session_id)
|
|
assert client.get_client_setting(SESSION_KEY) == session_id
|
|
client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_client_default_session_id(test_config: TestConfig):
|
|
# by default, the async client will NOT autogenerate the session id
|
|
async_client = await create_async_client(database=test_config.test_database,
|
|
host=test_config.host,
|
|
port=test_config.port,
|
|
user=test_config.username,
|
|
password=test_config.password)
|
|
assert async_client.get_client_setting(SESSION_KEY) is None
|
|
await async_client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_client_autogenerate_session_id(test_config: TestConfig):
|
|
async_client = await create_async_client(database=test_config.test_database,
|
|
host=test_config.host,
|
|
port=test_config.port,
|
|
user=test_config.username,
|
|
password=test_config.password,
|
|
autogenerate_session_id=True)
|
|
session_id = async_client.get_client_setting(SESSION_KEY)
|
|
try:
|
|
uuid.UUID(session_id)
|
|
except ValueError:
|
|
pytest.fail(f"Invalid session_id: {session_id}")
|
|
await async_client.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_client_custom_session_id(test_config: TestConfig):
|
|
session_id = 'custom_session_id'
|
|
async_client = await create_async_client(database=test_config.test_database,
|
|
host=test_config.host,
|
|
port=test_config.port,
|
|
user=test_config.username,
|
|
password=test_config.password,
|
|
session_id=session_id)
|
|
assert async_client.get_client_setting(SESSION_KEY) == session_id
|
|
await async_client.close()
|