mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
Joe SpadolaandGitHub
19ffd53f86
* implement native async client * add pylint exceptions * linting * improve test config setup * fix some tests * enable log tables on startup * improve error handling * linting * fix deprecation warning in aiohttp * error handling * separate sync/async stream failure tests * more error handling issues * linting * test updates and improvements * test adjustments * increase async stream buffer size * don't parallelize cloud tests * apply consistency settings to client factory fixture * up cloud par to 4 * enforce sequential consistency in helper * update enable_cleanup_closed vers * add conn abort to wrong port assertion * small test refactor * more consistency * add deadlock check in async queue * enforce consistent use of async iter stream consumption * linting * DO materialize non-streaming queries * accept either concurrent session error * update async client to always reset context * add pipelining to arrow methods * feature parity with recent sync client changes * fix jwt test after bad merge conflict resolution * 0.12.0rc1 release prep * fix assertion in racy test * clean up streaming source on exception * linting * more linting * deprecate legacy async client and replace with async native * update changelog
63 lines
2.6 KiB
Python
63 lines
2.6 KiB
Python
import pytest
|
|
|
|
from clickhouse_connect.driver import Client
|
|
from clickhouse_connect.driver.exceptions import StreamFailureError
|
|
|
|
|
|
def test_mid_stream_exception(test_client: Client):
|
|
"""Test that mid-stream exceptions are properly detected and raised."""
|
|
query = "SELECT sleepEachRow(0.01), throwIf(number=100) FROM numbers(200)"
|
|
|
|
with pytest.raises(StreamFailureError) as exc_info:
|
|
result = test_client.query(query, settings={"max_block_size": 1, "wait_end_of_query": 0})
|
|
_ = result.result_set
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "Value passed to 'throwIf' function is non-zero" in error_msg
|
|
assert test_client.command("SELECT 1") == 1
|
|
|
|
|
|
def test_mid_stream_exception_streaming(test_client: Client):
|
|
"""Test that mid-stream exceptions are properly detected in streaming mode."""
|
|
query = "SELECT sleepEachRow(0.01), throwIf(number=100) FROM numbers(200)"
|
|
|
|
with pytest.raises(StreamFailureError) as exc_info:
|
|
with test_client.query_rows_stream(query, settings={"max_block_size": 1, "wait_end_of_query": 0}) as stream:
|
|
for _ in stream:
|
|
pass
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "Value passed to 'throwIf' function is non-zero" in error_msg
|
|
assert test_client.command("SELECT 1") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mid_stream_exception_async(test_native_async_client):
|
|
"""Test that mid-stream exceptions are properly detected and raised (async)."""
|
|
query = "SELECT sleepEachRow(0.01), throwIf(number=100) FROM numbers(200)"
|
|
|
|
with pytest.raises(StreamFailureError) as exc_info:
|
|
result = await test_native_async_client.query(query, settings={"max_block_size": 1, "wait_end_of_query": 0})
|
|
_ = result.result_set
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "Value passed to 'throwIf' function is non-zero" in error_msg
|
|
assert await test_native_async_client.command("SELECT 1") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mid_stream_exception_streaming_async(test_native_async_client):
|
|
"""Test that mid-stream exceptions are properly detected in streaming mode (async)."""
|
|
query = "SELECT sleepEachRow(0.01), throwIf(number=100) FROM numbers(200)"
|
|
|
|
with pytest.raises(StreamFailureError) as exc_info:
|
|
async with await test_native_async_client.query_rows_stream(
|
|
query, settings={"max_block_size": 1, "wait_end_of_query": 0}
|
|
) as stream:
|
|
async for _ in stream:
|
|
pass
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "Value passed to 'throwIf' function is non-zero" in error_msg
|
|
assert await test_native_async_client.command("SELECT 1") == 1
|