ci: migration 09_http_handler to pytest

.github/actions/test_compat_client_cluster/action.yml

@@ -54,6 +54,6 @@ runs:
 
     - name: Run test
       shell: bash
-      run: nox -f tests/nox/noxfile.py
+      run: nox -f tests/nox/noxfile.py -s python_client java_client
       env:
         JDBC_MAIN_VER: ${{env.JDBC_MAIN_VER}}

.github/actions/test_stateful_cluster_linux/action.yml

@@ -13,3 +13,8 @@ runs:
       shell: bash
       run: |
         ./scripts/ci/ci-run-stateful-tests-cluster-minio.sh
+
+    - uses: wntrblm/[email protected]
+    - name: Run nox test
+      shell: bash
+      run: nox -f tests/nox/noxfile.py -s test_suites -- suites/1_stateful

.github/actions/test_stateful_standalone_linux/action.yml

@@ -13,3 +13,8 @@ runs:
       shell: bash
       run: |
         ./scripts/ci/ci-run-stateful-tests-standalone-minio.sh
+
+    - uses: wntrblm/[email protected]
+    - name: Run nox test
+      shell: bash
+      run: nox -f tests/nox/noxfile.py -s test_suites -- suites/1_stateful

tests/nox/noxfile.py

@@ -55,3 +55,10 @@ def run_jdbc_test(session, driver_version, main_version):
         external=True,
         env=env,
     )
+
+
+@nox.session
+def test_suites(session):
+    session.install("pytest", "requests", "pytest-asyncio")
+    # Usage: nox -s test_suites -- suites/1_stateful/09_http_handler/test_09_0007_session.py::test_session
+    session.run("pytest", *session.posargs)

tests/nox/pyproject.toml

@@ -1,2 +1,3 @@
 [tool.pytest.ini_options]
 addopts = "-s -vv"
+testpaths = ["suites"]

tests/nox/suites/1_stateful/09_http_handler/test_09_0000_invalid_utf8.py

@@ -0,0 +1,27 @@
+import requests
+
+# Define the URLs and credentials
+query_url = "http://localhost:8000/v1/query"
+auth = ("root", "")
+
+
+def execute_sql(sql):
+    """Execute SQL query via HTTP API"""
+    payload = {"sql": sql, "pagination": {"wait_time_secs": 3}}
+    response = requests.post(
+        query_url, auth=auth, headers={"Content-Type": "application/json"}, json=payload
+    )
+    return response.json()
+
+
+def test_invalid_utf8():
+    # Setup: create table and insert invalid UTF-8 data
+    assert execute_sql("drop table if exists t1")["error"] == None
+    assert execute_sql("create table t1(a varchar)")["error"] == None
+    assert execute_sql("insert INTO t1 VALUES(FROM_BASE64('0Aw='))")["error"] == {
+        "code": 1006,
+        "message": "invalid utf8 sequence while evaluating function `to_string(D00C)` in expr `CAST(from_hex('D00C')::string AS String NULL)`",
+    }
+
+    # Query the table
+    assert execute_sql("select * from t1")["data"] == []

tests/nox/suites/1_stateful/09_http_handler/test_09_0001_json_response.py

@@ -0,0 +1,46 @@
+import requests
+import json
+
+
+# Define the URLs and credentials
+query_url = "http://localhost:8000/v1/query"
+auth = ("root", "")
+
+
+def execute_query(payload_data):
+    """Execute query with custom payload data"""
+    response = requests.post(
+        query_url,
+        auth=auth,
+        headers={"Content-Type": "application/json"},
+        data=payload_data,
+    )
+    return response.json()
+
+
+def test_json_response_errors():
+    # Test 1: Invalid SQL query - select non-existent column
+    result1 = execute_query('{"sql": "select a", "pagination": { "wait_time_secs": 5}}')
+    assert result1.get("state", "Unknown") == "Failed"
+    assert result1["error"] == json.loads(
+        """{"code": 1065, "message": "error: \\n  --> SQL:1:8\\n  |\\n1 | select a\\n  |        ^ column a doesn't exist\\n\\n"}"""
+    )
+
+    # Test 2: Malformed JSON - missing quote before sql key
+    result2 = execute_query(
+        '{sql": "select * from tx", "pagination": { "wait_time_secs": 2}}'
+    )
+    assert result2 == json.loads(
+        '{"error": {"code": 400, "message": "parse error: key must be a string at line 1 column 2"}}'
+    )
+
+    # Test 3: Invalid endpoint URL
+    response3 = requests.post(
+        "http://localhost:8000/v1/querq/",  # Note: wrong endpoint
+        auth=auth,
+        headers={"Content-Type": "application/json"},
+        data='{"sql": "select * from tx", "pagination": { "wait_time_secs": 2}}',
+    )
+    assert response3.json() == json.loads(
+        '{"error": {"code": 404, "message": "not found"}}'
+    )

tests/nox/suites/1_stateful/09_http_handler/test_09_0002_sql_ends_semicolon.py

@@ -0,0 +1,48 @@
+import requests
+import json
+from suites.utils import comparison_output
+
+# Define the URLs and credentials
+query_url = "http://localhost:8000/v1/query"
+auth = ("root", "")
+
+
+def execute_query(sql):
+    """Execute SQL query via HTTP API"""
+    payload = {"sql": sql, "pagination": {"wait_time_secs": 5}}
+    response = requests.post(
+        query_url, auth=auth, headers={"Content-Type": "application/json"}, json=payload
+    )
+    return response.json()
+
+
+@comparison_output(
+    """>>>> create or replace table t_09_0002 (a int)
+<<<<
+"Succeeded"
+null
+"Succeeded"
+null
+"Succeeded"
+null
+"Succeeded"
+null"""
+)
+def test_sql_ends_with_semicolon():
+    # Create table
+    assert execute_query("create or replace table t_09_0002 (a int)")["error"] == None
+    print(">>>> create or replace table t_09_0002 (a int)")
+    print("<<<<")
+
+    # Test INSERT statements with semicolons
+    queries = [
+        "insert into t_09_0002 from @~/not_exist file_format=(type=csv);",
+        "insert into t_09_0002 select 1;",
+        "insert into t_09_0002 values (1);",
+        "select 1;",
+    ]
+
+    for query in queries:
+        result = execute_query(query)
+        print(f'"{result.get("state", "Unknown")}"')
+        print(json.dumps(result["error"]))

tests/nox/suites/1_stateful/09_http_handler/test_09_0003_error_detail.py

@@ -0,0 +1,68 @@
+import requests
+import os
+
+
+# Define the URLs and credentials
+query_url = "http://localhost:8000/v1/query"
+upload_url = "http://localhost:8000/v1/upload_to_stage"
+auth = ("root", "")
+
+
+def execute_sql(sql):
+    """Execute SQL query via HTTP API"""
+    payload = {"sql": sql, "pagination": {"wait_time_secs": 6}}
+    response = requests.post(
+        query_url, auth=auth, headers={"Content-Type": "application/json"}, json=payload
+    )
+    return response.json()
+
+
+def upload_to_stage(stage_name, file_path):
+    """Upload file to stage"""
+    with open(file_path, "rb") as f:
+        files = {"upload": f}
+        response = requests.put(
+            upload_url,
+            auth=auth,
+            headers={"x-databend-stage-name": stage_name},
+            files=files,
+        )
+    return response.json()
+
+
+def test_error_detail():
+    # Setup: drop existing table and stage
+    assert execute_sql("drop table if exists products")["error"] == None
+    assert execute_sql("drop stage if exists s1")["error"] == None
+
+    # Create stage and table
+    assert execute_sql("CREATE STAGE s1 FILE_FORMAT = (TYPE = CSV)")["error"] == None
+    assert execute_sql("remove @s1")["error"] == None
+    assert (
+        execute_sql("create table products (id int, name string, description string)")[
+            "error"
+        ]
+        == None
+    )
+
+    # Upload CSV file to stage (select.csv has 2 columns, table has 3 columns)
+    csv_file_path = os.path.join(
+        os.path.dirname(__file__), "../../../../data/csv/select.csv"
+    )
+    upload_result = upload_to_stage("s1", csv_file_path)
+    assert not "data" in upload_result
+
+    # Try to copy from stage to table - should fail with column mismatch error
+    copy_result = execute_sql(
+        "copy /*+ set_var(enable_distributed_copy_into = 0) */ into products (id, name, description) from @s1/"
+    )
+
+    assert copy_result["error"] == {
+        "code": 1046,
+        "message": "Number of columns in file (2) does not match that of the corresponding table (3)",
+        "detail": "at file 'select.csv', line 1",
+    }
+
+    # Cleanup
+    assert execute_sql("drop table if exists products")["error"] == None
+    assert execute_sql("drop stage if exists s1")["error"] == None

tests/nox/suites/1_stateful/09_http_handler/test_09_0004_large_row.py

@@ -0,0 +1,57 @@
+import requests
+from suites.utils import comparison_output
+
+# Define the URLs and credentials
+query_url = "http://localhost:8000/v1/query"
+auth = ("root", "")
+
+
+def execute_query(sql, query_id=None):
+    """Execute SQL query via HTTP API with optional query ID"""
+    headers = {"Content-Type": "application/json"}
+    if query_id:
+        headers["X-DATABEND-QUERY-ID"] = query_id
+
+    payload = {"sql": sql, "pagination": {"wait_time_secs": 5}}
+    response = requests.post(query_url, auth=auth, headers=headers, json=payload)
+    return response.json()
+
+
+def get_query_page(query_id, page):
+    """Get a specific page of query results"""
+    page_url = f"http://localhost:8000/v1/query/{query_id}/{page}"
+    response = requests.get(
+        page_url, auth=auth, headers={"Content-Type": "application/json"}
+    )
+    return response.json()
+
+
+def test_large_row():
+    qid = "09_004"
+
+    # Execute query with large JSON aggregation
+    result = execute_query(
+        "select json_array_agg(json_object('num',number)), (number % 2) as s from numbers(2000000) group by s;",
+        qid,
+    )
+    assert len(result.get("data", [])) == 1
+
+    # Get page 1 and page 2 of the results
+    page1_result = get_query_page(qid, "page/1")
+    assert len(page1_result.get("data", [])) == 1
+
+    page2_result = get_query_page(qid, "page/2")
+    assert len(page2_result.get("data", [])) == 0
+
+    result = execute_query("SELECT repeat(number::string, 2000) from numbers(100000)")
+    rows = len(result["data"])
+
+    qid = result["id"]
+
+    for i in range(1, 1000):
+        result = get_query_page(qid, f"page/{i}")
+        rows += len(result["data"])
+        if result["next_uri"] == result["final_uri"]:
+            assert get_query_page(qid, "final")["error"] == None
+            assert rows == 100000
+            return

tests/nox/suites/1_stateful/09_http_handler/test_09_0005_kill.py

@@ -0,0 +1,78 @@
+import requests
+import random
+from suites.utils import comparison_output
+
+# Define the URLs and credentials
+query_url = "http://localhost:8000/v1/query"
+auth = ("root", "")
+
+
+def execute_query(sql, query_id):
+    """Execute SQL query via HTTP API with specific query ID"""
+    headers = {"Content-Type": "application/json", "x-databend-query-id": query_id}
+    payload = {"sql": sql, "pagination": {"wait_time_secs": 6}}
+    response = requests.post(query_url, auth=auth, headers=headers, json=payload)
+    return response.json()
+
+
+def kill_query(query_id):
+    """Kill a running query"""
+    kill_url = f"http://localhost:8000/v1/query/{query_id}/kill"
+    response = requests.get(kill_url, auth=auth)
+    return response.status_code, response.text
+
+
+def get_query_page(query_id, page_number):
+    """Get a specific page of query results"""
+    page_url = f"http://localhost:8000/v1/query/{query_id}/page/{page_number}"
+    response = requests.get(page_url, auth=auth)
+    return response.status_code, response.text
+
+
+def get_query_final(query_id):
+    """Get final result of a query"""
+    final_url = f"http://localhost:8000/v1/query/{query_id}/final"
+    response = requests.get(final_url, auth=auth)
+    return response.json()
+
+
+@comparison_output(
+    """## query
+"Running"
+## kill
+200
+## page
+{"error":{"code":400,"message":"[HTTP-QUERY] Query ID QID canceled"}}
+400
+## final
+{'code': 1043, 'message': 'canceled by client'}
+"""
+)
+def test_kill_query():
+    # Generate random query ID
+    qid = f"my_query_for_kill_{random.randint(1000, 9999)}"
+
+    print("## query")
+    # Start a long-running query
+    query_result = execute_query(
+        "select sleep(0.5), number from numbers(15000000000);", qid
+    )
+    print(f'"{query_result.get("state", "Unknown")}"')
+
+    print("## kill")
+    # Kill the query
+    kill_status, kill_text = kill_query(qid)
+    print(kill_status)
+
+    print("## page")
+    # Try to get page 0 - should fail with cancellation error
+    page_status, page_text = get_query_page(qid, 0)
+    # Replace actual query ID with "QID" for consistent output
+    page_text = page_text.replace(qid, "QID")
+    print(page_text)
+    print(page_status)
+
+    print("## final")
+    # Get final result - should show cancellation error
+    final_result = get_query_final(qid)
+    print(final_result["error"])