feat: Add Core connection validation and refactor _determine_auth

- Refactor _determine_auth to accept URL object directly
- Add validation for Core connections to reject engine_name, account_name, and credentials
- Update existing Core connection tests to use valid format
- Add comprehensive tests for Core connection validation errors
- Addresses GitHub PR comments for stricter Core connection handling

Co-Authored-By: [email protected] <[email protected]>

Author: devin-ai-integration[bot]

src/firebolt_db/firebolt_dialect.py

@@ -162,10 +162,22 @@ def create_connect_args(self, url: URL) -> Tuple[List, Dict]:
         # bool flag to boolean for SDK compatibility
         token_cache_flag = bool(strtobool(parameters.pop("use_token_cache", "True")))
 
+        # Validate Core connection parameters
         if is_core_connection:
-            auth = _determine_auth("", "", token_cache_flag, is_core=True)
-        else:
-            auth = _determine_auth(url.username, url.password, token_cache_flag)
+            if url.username or url.password:
+                raise ArgumentError(
+                    "Core connections do not support username/password authentication"
+                )
+            if url.database:
+                raise ArgumentError(
+                    "Core connections do not support engine_name parameter"
+                )
+            if "account_name" in parameters:
+                raise ArgumentError(
+                    "Core connections do not support account_name parameter"
+                )
+
+        auth = _determine_auth(url, token_cache_flag)
 
         kwargs: Dict[str, Union[str, Auth, Dict[str, Any], None]] = {
             "database": url.host or None,
@@ -386,12 +398,13 @@ def get_is_nullable(column_is_nullable: int) -> bool:
     return column_is_nullable == 1
 
 
-def _determine_auth(
-    key: str, secret: str, token_cache_flag: bool = True, is_core: bool = False
-) -> Auth:
-    if is_core:
+def _determine_auth(url: URL, token_cache_flag: bool = True) -> Auth:
+    parameters = dict(url.query)
+    is_core_connection = "url" in parameters
+
+    if is_core_connection:
         return FireboltCore()
-    elif "@" in key:
-        return UsernamePassword(key, secret, token_cache_flag)
+    elif "@" in (url.username or ""):
+        return UsernamePassword(url.username, url.password, token_cache_flag)
     else:
-        return ClientCredentials(key, secret, token_cache_flag)
+        return ClientCredentials(url.username, url.password, token_cache_flag)

tests/unit/test_firebolt_dialect.py

@@ -304,13 +304,11 @@ def test_unicode_description(
         assert dialect._check_unicode_description(connection)
 
     def test_create_connect_args_core_connection(self, dialect: FireboltDialect):
-        connection_url = (
-            "test_engine://test_db_name/test_engine_name?" "url=http://localhost:8080"
-        )
+        connection_url = "test_engine://test_db_name?url=http://localhost:8080"
         u = url.make_url(connection_url)
         result_list, result_dict = dialect.create_connect_args(u)
 
-        assert result_dict["engine_name"] == "test_engine_name"
+        assert result_dict["engine_name"] is None
         assert result_dict["database"] == "test_db_name"
         assert result_dict["url"] == "http://localhost:8080"
         assert isinstance(result_dict["auth"], FireboltCore)
@@ -340,6 +338,45 @@ def test_create_connect_args_core_no_credentials_required(
         assert isinstance(result_dict["auth"], FireboltCore)
         assert "account_name" not in result_dict
 
+    def test_create_connect_args_core_with_credentials_error(
+        self, dialect: FireboltDialect
+    ):
+        connection_url = (
+            "test_engine://user:pass@test_db_name?url=http://localhost:8080"
+        )
+        u = url.make_url(connection_url)
+
+        with raises(
+            ArgumentError,
+            match="Core connections do not support username/password authentication",
+        ):
+            dialect.create_connect_args(u)
+
+    def test_create_connect_args_core_with_engine_error(self, dialect: FireboltDialect):
+        connection_url = (
+            "test_engine://test_db_name/test_engine?url=http://localhost:8080"
+        )
+        u = url.make_url(connection_url)
+
+        with raises(
+            ArgumentError, match="Core connections do not support engine_name parameter"
+        ):
+            dialect.create_connect_args(u)
+
+    def test_create_connect_args_core_with_account_error(
+        self, dialect: FireboltDialect
+    ):
+        connection_url = (
+            "test_engine://test_db_name?url=http://localhost:8080&account_name=test"
+        )
+        u = url.make_url(connection_url)
+
+        with raises(
+            ArgumentError,
+            match="Core connections do not support account_name parameter",
+        ):
+            dialect.create_connect_args(u)
+
 
 def test_get_is_nullable():
     assert firebolt_db.firebolt_dialect.get_is_nullable(1)