Add SQL_COPT_SS_ACCESS_TOKEN constant, replace hardcoded 1256

Author: bewithgaurav

mssql_python/auth.py

@@ -9,7 +9,7 @@
 from typing import Tuple, Dict, Optional, List
 
 from mssql_python.logging import logger
-from mssql_python.constants import AuthType
+from mssql_python.constants import AuthType, ConstantsDDBC
 
 
 class AADAuth:
@@ -293,7 +293,11 @@ def process_connection_string(
                 "process_connection_string: Token authentication configured successfully - auth_type=%s",
                 auth_type,
             )
-            return ";".join(modified_parameters) + ";", {1256: token_struct}, auth_type
+            return (
+                ";".join(modified_parameters) + ";",
+                {ConstantsDDBC.SQL_COPT_SS_ACCESS_TOKEN.value: token_struct},
+                auth_type,
+            )
         else:
             logger.warning(
                 "process_connection_string: Token acquisition failed, proceeding without token"

mssql_python/constants.py

@@ -158,6 +158,9 @@ class ConstantsDDBC(Enum):
     SQL_ATTR_SERVER_NAME = 13
     SQL_ATTR_RESET_CONNECTION = 116
 
+    # SQL Server-specific connection option constants
+    SQL_COPT_SS_ACCESS_TOKEN = 1256
+
     # Transaction Isolation Level Constants
     SQL_TXN_READ_UNCOMMITTED = 1
     SQL_TXN_READ_COMMITTED = 2

tests/test_008_auth.py

@@ -16,7 +16,7 @@
     process_connection_string,
     extract_auth_type,
 )
-from mssql_python.constants import AuthType
+from mssql_python.constants import AuthType, ConstantsDDBC
 import secrets
 
 SAMPLE_TOKEN = secrets.token_hex(44)
@@ -338,8 +338,8 @@ def test_process_connection_string_with_default_auth(self):
         assert "Server=test" in result_str
         assert "Database=testdb" in result_str
         assert attrs is not None
-        assert 1256 in attrs
-        assert isinstance(attrs[1256], bytes)
+        assert ConstantsDDBC.SQL_COPT_SS_ACCESS_TOKEN.value in attrs
+        assert isinstance(attrs[ConstantsDDBC.SQL_COPT_SS_ACCESS_TOKEN.value], bytes)
         assert auth_type == "default"
 
     def test_process_connection_string_no_auth(self):
@@ -361,8 +361,8 @@ def test_process_connection_string_interactive_non_windows(self, monkeypatch):
         assert "Server=test" in result_str
         assert "Database=testdb" in result_str
         assert attrs is not None
-        assert 1256 in attrs
-        assert isinstance(attrs[1256], bytes)
+        assert ConstantsDDBC.SQL_COPT_SS_ACCESS_TOKEN.value in attrs
+        assert isinstance(attrs[ConstantsDDBC.SQL_COPT_SS_ACCESS_TOKEN.value], bytes)
         assert auth_type == "interactive"