diff --git a/dbt/adapters/sqlserver/sqlserver_connections.py b/dbt/adapters/sqlserver/sqlserver_connections.py index a91baeb1..aa4891ca 100644 --- a/dbt/adapters/sqlserver/sqlserver_connections.py +++ b/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) ) diff --git a/dbt/adapters/sqlserver/sqlserver_credentials.py b/dbt/adapters/sqlserver/sqlserver_credentials.py index bf1f5075..43ceba22 100644 --- a/dbt/adapters/sqlserver/sqlserver_credentials.py +++ b/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): diff --git a/tests/conftest.py b/tests/conftest.py index 540ee302..10999e38 100644 --- a/tests/conftest.py +++ b/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") diff --git a/tests/unit/adapters/mssql/test_sqlserver_connection_manager.py b/tests/unit/adapters/mssql/test_sqlserver_connection_manager.py index 2acb2520..d3e1ab5b 100644 --- a/tests/unit/adapters/mssql/test_sqlserver_connection_manager.py +++ b/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]