test(e2e): live kernel-backend (use_sea=True) suite

Moves the previously-ad-hoc /tmp/connector_smoke.py into the repo
as a real pytest module under tests/e2e/ — same convention as the
rest of the e2e suite. Uses the existing session-scoped
`connection_details` fixture from the top-level conftest so it
shares the credential surface with every other live test.
11 tests cover:
- connect() with use_sea=True opens a session.
- SELECT 1: rows + description shape (column name + dbapi type slug).
- SELECT * FROM range(10000): multi-batch drain.
- fetchmany() pacing across the buffer boundary.
- fetchall_arrow() returns a pyarrow Table.
- All four metadata methods (catalogs / schemas / tables / columns).
- session_configuration={'ANSI_MODE': 'false'} round-trips.
- Bad SQL surfaces as DatabaseError with `code='SqlError'` and
`sql_state='42P01'` attached as exception attributes.
Module-level skips:
- `databricks_sql_kernel` not importable → whole module skipped via
pytest.importorskip (the wheel hasn't been installed).
- Live creds missing → fixture-level skip with a pointed message.
Run: `pytest tests/e2e/test_kernel_backend.py -v`. All 11 pass
against dogfood in ~20s.
Co-authored-by: Isaac
Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>

Author: vikrantpuppala

tests/e2e/test_kernel_backend.py

@@ -0,0 +1,186 @@
+"""E2E tests for ``use_sea=True`` (routes through the Rust kernel
+via the PyO3 ``databricks_sql_kernel`` module).
+
+PAT auth only. Anything else surfaces as ``NotSupportedError``
+from the auth bridge — covered as a unit test, not exercised here.
+
+Skipped automatically when:
+ - The standard ``DATABRICKS_SERVER_HOSTNAME`` / ``HTTP_PATH`` /
+ ``TOKEN`` creds aren't set (existing connector convention).
+ - ``databricks_sql_kernel`` isn't importable (the wheel hasn't
+ been installed; run ``pip install
+ 'databricks-sql-connector[kernel]'`` or, for local dev,
+ ``cd databricks-sql-kernel/pyo3 && maturin develop --release``
+ into this venv).
+
+Run from the connector repo root:
+
+ set -a && source ~/.databricks/pecotesting-creds && set +a
+ .venv/bin/pytest tests/e2e/test_kernel_backend.py -v
+"""
+
+from __future__ importannotations
+
+importpytest
+
+importdatabricks.sqlassql
+fromdatabricks.sql.excimportDatabaseError
+
+
+# Skip the whole module unless the kernel wheel is importable.
+pytest.importorskip(
+"databricks_sql_kernel",
+reason="use_sea=True requires the databricks-sql-kernel package",
+)
+
+
+@pytest.fixture(scope="module")
+defkernel_conn_params(connection_details):
+"""Live-cred check + connection params for use_sea=True.
+
+ Skips the module if any cred is missing rather than letting
+ every test fail with a confusing connect-time error.
+ """
+host=connection_details.get("host")
+http_path=connection_details.get("http_path")
+token=connection_details.get("access_token")
+ifnot (hostandhttp_pathandtoken):
+pytest.skip(
+"DATABRICKS_SERVER_HOSTNAME / DATABRICKS_HTTP_PATH / "
+"DATABRICKS_TOKEN not set"
+ )
+return {
+"server_hostname": host,
+"http_path": http_path,
+"access_token": token,
+"use_sea": True,
+ }
+
+
+@pytest.fixture
+defconn(kernel_conn_params):
+"""One-shot connection per test (the simple_test pattern the
+ existing e2e suite uses for cursor-level tests)."""
+c=sql.connect(**kernel_conn_params)
+try:
+yieldc
+finally:
+c.close()
+
+
+deftest_connect_with_use_sea_opens_a_session(conn):
+assertconn.open, "connection should report open after connect()"
+
+
+deftest_select_one(conn):
+withconn.cursor() ascur:
+cur.execute("SELECT 1 AS n")
+assertcur.description[0][0] =="n"
+# description type slug matches what Thrift produces
+assertcur.description[0][1] =="int"
+rows=cur.fetchall()
+assertlen(rows) ==1
+assertrows[0][0] ==1
+
+
+deftest_drain_large_range_to_arrow(conn):
+"""SELECT * FROM range(10000) drains as a pyarrow Table with
+ 10000 rows. Exercises the CloudFetch / multi-batch path on the
+ kernel side."""
+withconn.cursor() ascur:
+cur.execute("SELECT * FROM range(10000)")
+rows=cur.fetchall()
+assertlen(rows) ==10000
+
+
+deftest_fetchmany_pacing(conn):
+"""fetchmany honours the requested size and stops cleanly at
+ end-of-stream — covers the buffer-slicing logic in
+ KernelResultSet."""
+withconn.cursor() ascur:
+cur.execute("SELECT * FROM range(50)")
+r1=cur.fetchmany(10)
+r2=cur.fetchmany(20)
+r3=cur.fetchmany(100) # capped at remaining
+assert (len(r1), len(r2), len(r3)) == (10, 20, 20)
+
+
+deftest_fetchall_arrow(conn):
+withconn.cursor() ascur:
+cur.execute("SELECT 1 AS a, 'hi' AS b")
+table=cur.fetchall_arrow()
+asserttable.num_rows==1
+asserttable.column_names== ["a", "b"]
+
+
+# ── Metadata ──────────────────────────────────────────────────────
+
+
+deftest_metadata_catalogs(conn):
+withconn.cursor() ascur:
+cur.catalogs()
+rows=cur.fetchall()
+assertlen(rows) >0
+
+
+deftest_metadata_schemas(conn):
+withconn.cursor() ascur:
+cur.schemas(catalog_name="main")
+rows=cur.fetchall()
+assertlen(rows) >0
+
+
+deftest_metadata_tables(conn):
+withconn.cursor() ascur:
+cur.tables(catalog_name="system", schema_name="information_schema")
+rows=cur.fetchall()
+assertlen(rows) >0
+
+
+deftest_metadata_columns(conn):
+withconn.cursor() ascur:
+cur.columns(
+catalog_name="system",
+schema_name="information_schema",
+table_name="tables",
+ )
+rows=cur.fetchall()
+assertlen(rows) >0
+
+
+# ── Session configuration ─────────────────────────────────────────
+
+
+deftest_session_configuration_round_trips(kernel_conn_params):
+"""`session_configuration` flows through to the kernel's
+ `session_conf` and is honoured by the server.
+
+ `ANSI_MODE` is the safe choice — it's on the SEA allow-list and
+ isn't workspace-policy-clamped (unlike `STATEMENT_TIMEOUT`) or
+ rejected by the warehouse (unlike `TIMEZONE` on dogfood)."""
+params=dict(kernel_conn_params)
+params["session_configuration"] = {"ANSI_MODE": "false"}
+withsql.connect(**params) asc:
+withc.cursor() ascur:
+cur.execute("SET ANSI_MODE")
+rows=cur.fetchall()
+kv= {r[0]: r[1] forrinrows}
+assertkv.get("ANSI_MODE") =="false", f"got {rows!r}"
+
+
+# ── Error mapping ─────────────────────────────────────────────────
+
+
+deftest_bad_sql_surfaces_as_databaseerror(conn):
+"""Bad SQL should surface as a PEP 249 ``DatabaseError`` with
+ the kernel's structured fields (`code`, `sql_state`, `query_id`)
+ attached as attributes — the connector backend re-raises the
+ kernel's ``SqlError`` to ``DatabaseError`` while preserving the
+ server-reported state."""
+withconn.cursor() ascur:
+withpytest.raises(DatabaseError) asexc_info:
+cur.execute("SELECT * FROM definitely_not_a_table_xyz_kernel_e2e")
+err=exc_info.value
+# Structured fields copied off the kernel exception:
+assertgetattr(err, "code", None) =="SqlError"
+assertgetattr(err, "sql_state", None) =="42P01"