Revert e2e OAuth M2M; use SP PAT instead (keep DBR LTS check on M2M)

The azure-prod DATABRICKS_TOKEN is now a personal access token owned by the
PECO service principal, so its identity matches DATABRICKS_USER. That fixes the
Personal Staging Location tests (stage://tmp/<SP>/...) with a plain PAT, without
the OAuth M2M machinery -- which also broke the retry/HTTP tests, since M2M
makes a live token-endpoint call that those tests' urllib3 mocking intercepts.

Reverts the e2e auth changes (conftest.auth_connect_kwargs + the consumer call
sites + the databricks-sdk dev dep + the code-coverage.yml SP env) back to the
plain access_token path. The DBR LTS install check keeps OAuth M2M: it hits the
workspace Jobs/SCIM API (which rejects a warehouse-scoped PAT), is proven 15/15
green on M2M, and installs databricks-sdk itself in its own workflow.

Co-authored-by: Isaac
Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>

Author: vikrantpuppala

.github/workflows/code-coverage.yml

@@ -45,14 +45,6 @@ jobs:
       DATABRICKS_SERVER_HOSTNAME: ${{ secrets.DATABRICKS_HOST }}
       DATABRICKS_HTTP_PATH: ${{ secrets.TEST_PECO_WAREHOUSE_HTTP_PATH }}
       DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
-      # OAuth M2M as the PECO service principal. The e2e suite prefers this
-      # over the PAT (see conftest.auth_connect_kwargs) so the connecting
-      # identity IS the service principal -- required for the Personal Staging
-      # Location tests, whose stage path is stage://tmp/<DATABRICKS_USER>/ and
-      # DATABRICKS_USER is this same SP. A PAT for a different identity fails
-      # those with PERMISSION_DENIED on the personal stage.
-      DATABRICKS_CLIENT_ID: ${{ secrets.TEST_PECO_SP_ID }}
-      DATABRICKS_CLIENT_SECRET: ${{ secrets.TEST_PECO_SP_OAUTH_SECRET }}
       DATABRICKS_CATALOG: peco
       DATABRICKS_USER: ${{ secrets.TEST_PECO_SP_ID }}
     steps:

conftest.py

@@ -17,17 +17,6 @@ def access_token():
     return os.getenv("DATABRICKS_TOKEN")
 
 
-@pytest.fixture(scope="session")
-def client_id():
-    # OAuth M2M service-principal client id (application id).
-    return os.getenv("DATABRICKS_CLIENT_ID")
-
-
-@pytest.fixture(scope="session")
-def client_secret():
-    return os.getenv("DATABRICKS_CLIENT_SECRET")
-
-
 @pytest.fixture(scope="session")
 def ingestion_user():
     return os.getenv("DATABRICKS_USER")
@@ -44,58 +33,12 @@ def schema():
 
 
 @pytest.fixture(scope="session", autouse=True)
-def connection_details(
-    host,
-    http_path,
-    access_token,
-    client_id,
-    client_secret,
-    ingestion_user,
-    catalog,
-    schema,
-):
+def connection_details(host, http_path, access_token, ingestion_user, catalog, schema):
     return {
         "host": host,
         "http_path": http_path,
         "access_token": access_token,
-        "client_id": client_id,
-        "client_secret": client_secret,
         "ingestion_user": ingestion_user,
         "catalog": catalog,
         "schema": schema,
     }
-
-
-def auth_connect_kwargs(details):
-    """Return the sql.connect auth kwargs from connection_details.
-
-    Prefers OAuth M2M (service principal) when DATABRICKS_CLIENT_ID /
-    DATABRICKS_CLIENT_SECRET are set, otherwise falls back to a PAT
-    (DATABRICKS_TOKEN). M2M is required for identity-scoped operations such as
-    Personal Staging Location tests (stage://tmp/<DATABRICKS_USER>/...), where
-    the connecting identity must equal DATABRICKS_USER -- the service principal.
-    A PAT belonging to a different identity cannot access that stage.
-    """
-    client_id = details.get("client_id")
-    client_secret = details.get("client_secret")
-    if client_id and client_secret:
-        host = details["host"]
-        host_url = host if host.startswith("http") else f"https://{host}"
-
-        def credential_provider():
-            # Imported lazily so a PAT-only environment doesn't require the SDK.
-            from databricks.sdk.core import Config, oauth_service_principal
-
-            return oauth_service_principal(
-                Config(
-                    host=host_url,
-                    client_id=client_id,
-                    client_secret=client_secret,
-                    # Explicit so an ambient DATABRICKS_TOKEN doesn't collide
-                    # ("more than one authorization method configured").
-                    auth_type="oauth-m2m",
-                )
-            )
-
-        return {"credentials_provider": credential_provider}
-    return {"access_token": details.get("access_token")}

pyproject.toml

@@ -93,13 +93,6 @@ black = "^22.3.0"
 pytest-dotenv = "^0.5.2"
 pytest-cov = "^4.0.0"
 pytest-xdist = "^3.0.0"
-# Test-only: the e2e suite authenticates to the warehouse via OAuth M2M
-# (service principal) using databricks.sdk's oauth_service_principal — see
-# conftest.auth_connect_kwargs. The connector itself has no native
-# Databricks-OAuth M2M path (its azure-sp-m2m targets Azure AD; its
-# databricks-oauth path is interactive U2M), so the SDK provides the
-# client-credentials flow against /oidc/v1/token. Not a runtime dependency.
-databricks-sdk = ">=0.20"
 numpy = [
     { version = ">=1.16.6", python = ">=3.8,<3.11" },
     { version = ">=1.23.4", python = ">=3.11" },

tests/e2e/test_circuit_breaker.py

@@ -21,7 +21,6 @@
 
 import databricks.sql as sql
 from databricks.sql.telemetry.circuit_breaker_manager import CircuitBreakerManager
-from conftest import auth_connect_kwargs
 
 
 def wait_for_circuit_state(circuit_breaker, expected_states, timeout=5):
@@ -121,7 +120,7 @@ def mock_request(*args, **kwargs):
             with sql.connect(
                 server_hostname=self.arguments["host"],
                 http_path=self.arguments["http_path"],
-                **auth_connect_kwargs(self.arguments),
+                access_token=self.arguments.get("access_token"),
                 force_enable_telemetry=True,
                 telemetry_batch_size=1,
                 _telemetry_circuit_breaker_enabled=True,
@@ -182,7 +181,7 @@ def mock_rate_limited_request(*args, **kwargs):
             with sql.connect(
                 server_hostname=self.arguments["host"],
                 http_path=self.arguments["http_path"],
-                **auth_connect_kwargs(self.arguments),
+                access_token=self.arguments.get("access_token"),
                 force_enable_telemetry=True,
                 telemetry_batch_size=1,
                 _telemetry_circuit_breaker_enabled=False,  # Disabled
@@ -216,7 +215,7 @@ def mock_conditional_request(*args, **kwargs):
             with sql.connect(
                 server_hostname=self.arguments["host"],
                 http_path=self.arguments["http_path"],
-                **auth_connect_kwargs(self.arguments),
+                access_token=self.arguments.get("access_token"),
                 force_enable_telemetry=True,
                 telemetry_batch_size=1,
                 _telemetry_circuit_breaker_enabled=True,

tests/e2e/test_driver.py

@@ -99,9 +99,9 @@ def connection_params(self):
         return params
 
     def auth_params(self):
-        from conftest import auth_connect_kwargs
-
-        return auth_connect_kwargs(self.arguments)
+        return {
+            "access_token": self.arguments.get("access_token"),
+        }
 
     @contextmanager
     def connection(self, extra_params=()):

tests/e2e/test_telemetry_e2e.py

@@ -24,12 +24,10 @@ def get_details(self, connection_details):
         self.arguments = connection_details.copy()
 
     def connection_params(self):
-        from conftest import auth_connect_kwargs
-
         return {
             "server_hostname": self.arguments["host"],
             "http_path": self.arguments["http_path"],
-            **auth_connect_kwargs(self.arguments),
+            "access_token": self.arguments.get("access_token"),
         }
 
     @contextmanager

tests/e2e/test_transactions.py

@@ -56,12 +56,10 @@ def _unique_table_name_raw(suffix):
 @pytest.fixture
 def mst_conn_params(connection_details):
     """Connection parameters with MST enabled."""
-    from conftest import auth_connect_kwargs
-
     return {
         "server_hostname": connection_details["host"],
         "http_path": connection_details["http_path"],
-        **auth_connect_kwargs(connection_details),
+        "access_token": connection_details.get("access_token"),
         "ignore_transactions": False,
     }