Support new encryption options

dbt/adapters/sqlserver/sqlserver_connections.py

@@ -103,7 +103,10 @@ def open(cls, connection: Connection) -> Connection:
         assert credentials.encrypt is not None
         assert credentials.trust_cert is not None
 
-        con_str.append(bool_to_connection_string_arg("encrypt", credentials.encrypt))
+        if isinstance(credentials.encrypt, bool):
+            con_str.append(bool_to_connection_string_arg("encrypt", credentials.encrypt))
+        else:
+            con_str.append(f"encrypt={credentials.encrypt}")
         con_str.append(
             bool_to_connection_string_arg("TrustServerCertificate", credentials.trust_cert)
         )

dbt/adapters/sqlserver/sqlserver_credentials.py

@@ -1,8 +1,21 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import Literal, Optional
 
 from dbt.adapters.fabric import FabricCredentials
 
+# Source: https://learn.microsoft.com/en-us/sql/relational-databases/security/networking/tds-8?view=sql-server-ver17#additional-changes-to-connection-string-encryption-properties  # noqa: E501
+EncryptType = Literal[
+    "true",
+    "yes",
+    "mandatory",
+    "strict",
+    "optional",
+    "false",
+    "no",
+    True,
+    False,
+]
+
 
 @dataclass
 class SQLServerCredentials(FabricCredentials):
@@ -13,6 +26,7 @@ class SQLServerCredentials(FabricCredentials):
 
     port: Optional[int] = 1433
     authentication: Optional[str] = "sql"
+    encrypt: Optional[EncryptType] = True
 
     @property
     def type(self):

tests/conftest.py

@@ -30,6 +30,8 @@ def dbt_profile_target(request: FixtureRequest, dbt_profile_target_update):
         target = _profile_user()
     elif profile == "user_azure":
         target = _profile_user_azure()
+    elif profile == "user_encrypt_strict":
+        target = _profile_user_encrypt_strict()
     else:
         raise ValueError(f"Unknown profile: {profile}")
 
@@ -152,6 +154,15 @@ def _profile_user_azure():
     return profile
 
 
+def _profile_user_encrypt_strict():
+    return {
+        **_profile_user(),
+        **{
+            "encrypt": "strict",
+        },
+    }
+
+
 @pytest.fixture(autouse=True)
 def skip_by_profile_type(request: FixtureRequest):
     profile_type = request.config.getoption("--profile")

tests/unit/adapters/mssql/test_sqlserver_connection_manager.py

@@ -1,7 +1,11 @@
+from unittest import mock
+
 import pytest
 from azure.identity import AzureCliCredential
+from dbt.adapters.contracts.connection import Connection
 
 from dbt.adapters.sqlserver.sqlserver_connections import (  # byte_array_to_datetime,
+    SQLServerConnectionManager,
     bool_to_connection_string_arg,
     get_pyodbc_attrs_before_credentials,
 )
@@ -39,3 +43,49 @@ def test_get_pyodbc_attrs_before_empty_dict_when_service_principal(
 )
 def test_bool_to_connection_string_arg(key: str, value: bool, expected: str) -> None:
     assert bool_to_connection_string_arg(key, value) == expected
+
+
+@mock.patch("pyodbc.connect")
+def test_encrypt_strict(pyodbc_connect_mock):
+    """encrypt set to strict is supported."""
+    # Given a connection with encrypt set to strict
+    connection = Connection(
+        "sqlserver",
+        "test",
+        SQLServerCredentials(
+            driver="ODBC Driver 17 for SQL Server",
+            host="fake.sql.sqlserver.net",
+            database="dbt",
+            schema="sqlserver",
+            encrypt="strict",
+        ),
+    )
+
+    # When the connection is open
+    SQLServerConnectionManager.open(connection)
+
+    # Then the connection string contains encrypt=strict
+    assert "encrypt=strict" in pyodbc_connect_mock.call_args[0][0]
+
+
+@mock.patch("pyodbc.connect")
+def test_encrypt_true(pyodbc_connect_mock):
+    """encrypt set to True is supported."""
+    # Given a connection with encrypt set to True
+    connection = Connection(
+        "sqlserver",
+        "test",
+        SQLServerCredentials(
+            driver="ODBC Driver 17 for SQL Server",
+            host="fake.sql.sqlserver.net",
+            database="dbt",
+            schema="sqlserver",
+            encrypt=True,
+        ),
+    )
+
+    # When the connection is open
+    SQLServerConnectionManager.open(connection)
+
+    # Then the connection string contains encrypt=strict
+    assert "encrypt=Yes" in pyodbc_connect_mock.call_args[0][0]