2222from __future__ import annotations
2323
2424import hashlib
25+ from http .server import BaseHTTPRequestHandler , ThreadingHTTPServer
26+ import select
27+ import socket
2528import sys
29+ import threading
2630from uuid import uuid4
2731
2832import pytest
@@ -137,15 +141,61 @@ def conn(kernel_conn_params):
137141 c .close ()
138142
139143
140- @pytest .mark .parametrize ("use_cloud_fetch" , [False , True ])
144+ @pytest .fixture
145+ def workspace_only_proxy (kernel_conn_params , monkeypatch ):
146+ connections = {"workspace" : 0 , "blocked" : 0 }
147+ workspace = kernel_conn_params ["server_hostname" ] + ":443"
148+
149+ class Handler (BaseHTTPRequestHandler ):
150+ def do_CONNECT (self ):
151+ self .close_connection = True
152+ if self .path != workspace :
153+ connections ["blocked" ] += 1
154+ self .send_error (403 )
155+ return
156+ with socket .create_connection (
157+ (kernel_conn_params ["server_hostname" ], 443 ), timeout = 30
158+ ) as upstream :
159+ connections ["workspace" ] += 1
160+ self .send_response (200 )
161+ self .end_headers ()
162+ while True :
163+ ready , _ , _ = select .select ([self .connection , upstream ], [], [], 30 )
164+ if not ready :
165+ return
166+ for source in ready :
167+ data = source .recv (65536 )
168+ if not data :
169+ return
170+ target = (
171+ upstream if source is self .connection else self .connection
172+ )
173+ target .sendall (data )
174+
175+ def log_message (self , * _args ):
176+ pass
177+
178+ proxy = ThreadingHTTPServer (("127.0.0.1" , 0 ), Handler )
179+ thread = threading .Thread (target = proxy .serve_forever , daemon = True )
180+ thread .start ()
181+ for name in ("HTTPS_PROXY" , "https_proxy" , "HTTP_PROXY" , "http_proxy" ):
182+ monkeypatch .setenv (name , f"http://127.0.0.1:{ proxy .server_port } " )
183+ for name in ("NO_PROXY" , "no_proxy" ):
184+ monkeypatch .setenv (name , "" )
185+ try :
186+ yield connections
187+ finally :
188+ proxy .shutdown ()
189+ proxy .server_close ()
190+ thread .join ()
191+
192+
141193def test_cloudfetch_disabled_returns_complete_large_result (
142- kernel_conn_params , caplog , use_cloud_fetch
194+ kernel_conn_params , workspace_only_proxy
143195):
144196 from databricks .sql .backend .kernel .client import KernelDatabricksClient
145197
146- with caplog .at_level (logging .INFO , logger = "databricks.sql.kernel" ), sql .connect (
147- ** kernel_conn_params , use_cloud_fetch = use_cloud_fetch
148- ) as connection :
198+ with sql .connect (** kernel_conn_params , use_cloud_fetch = False ) as connection :
149199 assert isinstance (connection .session .backend , KernelDatabricksClient )
150200 with connection .cursor () as cursor :
151201 cursor .execute (
@@ -168,19 +218,9 @@ def test_cloudfetch_disabled_returns_complete_large_result(
168218 "a25ada4c6c476b53e97eb6e8a945e9ce4c14cc432c7ad672dc6fb607d660c86a"
169219 )
170220
171- messages = [
172- record .getMessage ()
173- for record in caplog .records
174- if record .name .startswith ("databricks.sql.kernel" )
175- ]
176- assert (
177- any ("Using CloudFetch reader" in message for message in messages )
178- == use_cloud_fetch
179- )
180- assert any (
181- "inline" in message .lower () and "reader" in message .lower ()
182- for message in messages
183- ) == (not use_cloud_fetch )
221+ assert workspace_only_proxy ["workspace" ] > 0
222+ assert workspace_only_proxy ["blocked" ] == 0
223+
184224
185225def test_connect_with_use_kernel_opens_a_session (conn ):
186226 assert conn .open , "connection should report open after connect()"
0 commit comments