Aligned U2M detection method for shared_connection with _static_connection_kwargs.

Signed-off-by: davem-bis <68955845+davem-bis@users.noreply.github.com>

Author: davem-bis

sqlmesh/core/config/connection.py

@@ -837,7 +837,7 @@ class DatabricksConnectionConfig(ConnectionConfig):
     @property
     def shared_connection(self) -> bool:
         """The connection should only be shared if U2M OAuth is being used"""
-        return self.auth_type is not None and self.oauth_client_id is None
+        return self.auth_type is not None and self.oauth_client_secret is None
 
     @model_validator(mode="before")
     def _databricks_connect_validator(cls, data: t.Any) -> t.Any:

tests/core/test_connection_config.py

@@ -1456,7 +1456,8 @@ def test_databricks__u2m_oauth__shared_connection_pool(make_config):
     assert isinstance(adapter._connection_pool, ThreadLocalSharedConnectionPool)
 
 
-def test_databricks__m2m_oauth__connection_pool(make_config):
+@patch.object(DatabricksConnectionConfig, "_connection_factory_with_kwargs")
+def test_databricks__m2m_oauth__connection_pool(mock_connection_factory_with_kwargs, make_config):
     from sqlmesh.utils.connection_pool import ThreadLocalConnectionPool
 
     config = make_config(
@@ -1465,6 +1466,7 @@ def test_databricks__m2m_oauth__connection_pool(make_config):
         http_path="sql/test/foo",
         auth_type="databricks-oauth",
         oauth_client_id="oauth_client_id",
+        oauth_client_secret="oauth_client_secret",
         concurrent_tasks=4,
     )
     assert isinstance(config, DatabricksConnectionConfig)