{{ message }}
Repository navigation
Expand file tree
/
Copy pathtest_008_auth.py
More file actions
3345 lines (2720 loc) · 142 KB
/
Copy pathtest_008_auth.py
File metadata and controls
3345 lines (2720 loc) · 142 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Copyright (c) Microsoft Corporation.
Licensed under the MIT license.
Tests for the auth module.
"""
import pytest
import collections
import platform
import sys
import threading
import warnings
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import patch, MagicMock
from mssql_python.auth import (
AADAuth,
ServicePrincipalAuth,
TokenInfo,
_parse_tenant_id,
process_auth_parameters,
remove_sensitive_params,
get_auth_token_info,
compute_identity_key,
extract_auth_type,
_credential_cache,
acquire_token_from_credential,
acquire_raw_token_from_credential,
_SQL_SCOPE,
_credential_cache_lock,
_account_id_cache,
)
try:
from azure.core.credentials import TokenCredential
except ImportError: # pragma: no cover - azure-core is an optional test dependency
TokenCredential = None
from mssql_python.constants import AuthType, ConstantsDDBC
from mssql_python.exceptions import InterfaceError, OperationalError
import secrets
SAMPLE_TOKEN = secrets.token_hex(44)
@pytest.fixture(autouse=True)
def setup_azure_identity():
"""Setup mock azure.identity module"""
class MockToken:
token = SAMPLE_TOKEN
class MockDefaultAzureCredential:
def get_token(self, scope):
return MockToken()
class MockAuthenticationRecord:
# Minimal stand-in for azure.identity.AuthenticationRecord: only the
# public home_account_id is consumed by the pooling key logic.
home_account_id = "mock-home-account-id"
class MockDeviceCodeCredential:
def __init__(self, **kwargs):
# Accept disable_automatic_authentication (and any future kwargs)
# the same way real azure-identity credentials do.
self._init_kwargs = kwargs
def authenticate(self, **kwargs):
return MockAuthenticationRecord()
def get_token(self, scope):
return MockToken()
class MockInteractiveBrowserCredential:
def __init__(self, **kwargs):
self._init_kwargs = kwargs
def authenticate(self, **kwargs):
return MockAuthenticationRecord()
def get_token(self, scope):
return MockToken()
class MockClientSecretCredential:
# Captures construction kwargs and get_token args so ServicePrincipal
# tests can assert the right tenant/client_id/secret/scope flowed
# through from the connection string + STS URL.
last_init_kwargs = None
last_scope = None
def __init__(self, **kwargs):
MockClientSecretCredential.last_init_kwargs = kwargs
def get_token(self, scope):
MockClientSecretCredential.last_scope = scope
return MockToken()
class MockManagedIdentityCredential:
# Captures construction kwargs so user-assigned MSI tests can assert
# client_id was forwarded correctly.
last_init_kwargs = None
def __init__(self, **kwargs):
MockManagedIdentityCredential.last_init_kwargs = kwargs
def get_token(self, scope):
return MockToken()
class MockRequestsTransport:
# Captures construction kwargs so the SP factory's timeout config
# can be asserted.
last_init_kwargs = None
def __init__(self, **kwargs):
MockRequestsTransport.last_init_kwargs = kwargs
# Mock ClientAuthenticationError
class MockClientAuthenticationError(Exception):
pass
# Prefer the real AuthenticationRequiredError so production code exercises
# the exact exception type it will see in production; fall back to a local
# subclass only when azure-identity is not installed (the scenario this mock
# exists to cover). Either way the mocked module exposes the same class
# object that production imports, so silent-first's except-clause matches.
try:
from azure.identity import AuthenticationRequiredError as MockAuthenticationRequiredError
except ImportError:
class MockAuthenticationRequiredError(MockClientAuthenticationError):
"""Local stand-in when azure-identity is unavailable."""
class MockIdentity:
DefaultAzureCredential = MockDefaultAzureCredential
DeviceCodeCredential = MockDeviceCodeCredential
InteractiveBrowserCredential = MockInteractiveBrowserCredential
ClientSecretCredential = MockClientSecretCredential
ManagedIdentityCredential = MockManagedIdentityCredential
AuthenticationRequiredError = MockAuthenticationRequiredError
class MockCore:
class exceptions:
ClientAuthenticationError = MockClientAuthenticationError
class pipeline:
class transport:
RequestsTransport = MockRequestsTransport
# Create mock azure module if it doesn't exist
if "azure" not in sys.modules:
sys.modules["azure"] = type("MockAzure", (), {})()
# Add identity and core modules to azure
sys.modules["azure.identity"] = MockIdentity()
sys.modules["azure.core"] = MockCore()
sys.modules["azure.core.exceptions"] = MockCore.exceptions()
sys.modules["azure.core.pipeline"] = MockCore.pipeline()
sys.modules["azure.core.pipeline.transport"] = MockCore.pipeline.transport()
yield
# Cleanup
for module in [
"azure.identity",
"azure.core",
"azure.core.exceptions",
"azure.core.pipeline",
"azure.core.pipeline.transport",
]:
if module in sys.modules:
del sys.modules[module]
@pytest.fixture(autouse=True)
def clear_credential_cache():
"""Clear the module-level credential and account-id caches between tests."""
_credential_cache.clear()
_account_id_cache.clear()
yield
_credential_cache.clear()
_account_id_cache.clear()
class TestAuthType:
def test_auth_type_constants(self):
assert AuthType.INTERACTIVE.value == "activedirectoryinteractive"
assert AuthType.DEVICE_CODE.value == "activedirectorydevicecode"
assert AuthType.DEFAULT.value == "activedirectorydefault"
assert AuthType.MSI.value == "activedirectorymsi"
assert AuthType.SERVICE_PRINCIPAL.value == "activedirectoryserviceprincipal"
class TestAADAuth:
def test_get_token_struct(self):
token_struct = AADAuth.get_token_struct(SAMPLE_TOKEN)
assert isinstance(token_struct, bytes)
assert len(token_struct) > 4
def test_get_raw_token_default(self):
raw_token = AADAuth.get_raw_token("default")
assert isinstance(raw_token, str)
assert raw_token == SAMPLE_TOKEN
def test_get_token_default(self):
token_struct = AADAuth.get_token("default")
assert isinstance(token_struct, bytes)
def test_get_token_device_code(self):
token_struct = AADAuth.get_token("devicecode")
assert isinstance(token_struct, bytes)
def test_get_token_interactive(self):
token_struct = AADAuth.get_token("interactive")
assert isinstance(token_struct, bytes)
def test_get_token_credential_mapping(self):
# Test that all supported auth types work
supported_types = ["default", "devicecode", "interactive"]
for auth_type in supported_types:
token_struct = AADAuth.get_token(auth_type)
assert isinstance(token_struct, bytes)
assert len(token_struct) > 4
def test_get_token_client_authentication_error(self):
"""Test that ClientAuthenticationError is properly handled"""
from azure.core.exceptions import ClientAuthenticationError
# Create a mock credential that raises ClientAuthenticationError
class MockFailingCredential:
def get_token(self, scope):
raise ClientAuthenticationError("Mock authentication failed")
# Use monkeypatch to mock the credential creation
def mock_get_token_failing(auth_type):
from azure.core.exceptions import ClientAuthenticationError
if auth_type == "default":
try:
credential = MockFailingCredential()
token = credential.get_token("https://database.windows.net/.default").token
return AADAuth.get_token_struct(token)
except ClientAuthenticationError as e:
raise RuntimeError(
f"Azure AD authentication failed for MockFailingCredential: {e}. "
f"This could be due to invalid credentials, missing environment variables, "
f"user cancellation, network issues, or unsupported configuration."
) from e
else:
return AADAuth.get_token(auth_type)
with pytest.raises(RuntimeError, match="Azure AD authentication failed"):
mock_get_token_failing("default")
def test_get_token_general_exception_handling_init_error(self):
"""Test general Exception handling during credential initialization (Lines 52-56)."""
# Test by modifying the mock credential classes to raise exceptions
import sys
# Get the current azure.identity module (which is mocked)
azure_identity = sys.modules["azure.identity"]
# Store original credentials
original_default = azure_identity.DefaultAzureCredential
original_device = azure_identity.DeviceCodeCredential
original_interactive = azure_identity.InteractiveBrowserCredential
# Create a mock credential that raises exceptions during initialization
class MockCredentialWithInitError:
def __init__(self):
raise ValueError("Mock credential initialization failed")
def get_token(self, scope):
pass # Won't be reached
try:
# Test DefaultAzureCredential initialization error
azure_identity.DefaultAzureCredential = MockCredentialWithInitError
with pytest.raises(RuntimeError) as exc_info:
AADAuth.get_token("default")
# Verify the error message format (lines 54-56)
error_message = str(exc_info.value)
assert "Failed to create MockCredentialWithInitError" in error_message
assert "Mock credential initialization failed" in error_message
# Verify exception chaining is preserved (from e)
assert exc_info.value.__cause__ is not None
assert isinstance(exc_info.value.__cause__, ValueError)
# Test different exception types
class MockCredentialWithTypeError:
def __init__(self, **kwargs):
raise TypeError("Invalid argument type passed")
azure_identity.DeviceCodeCredential = MockCredentialWithTypeError
with pytest.raises(RuntimeError) as exc_info:
AADAuth.get_token("devicecode")
assert "Failed to create MockCredentialWithTypeError" in str(exc_info.value)
assert "Invalid argument type passed" in str(exc_info.value)
assert isinstance(exc_info.value.__cause__, TypeError)
finally:
# Restore original credentials
azure_identity.DefaultAzureCredential = original_default
azure_identity.DeviceCodeCredential = original_device
azure_identity.InteractiveBrowserCredential = original_interactive
def test_get_token_general_exception_handling_token_error(self):
"""Test general Exception handling during token retrieval (Lines 52-56)."""
import sys
azure_identity = sys.modules["azure.identity"]
# Store original credentials
original_interactive = azure_identity.InteractiveBrowserCredential
# Create a credential that fails during get_token call
class MockCredentialWithTokenError:
def __init__(self, **kwargs):
pass # Successful initialization
def get_token(self, scope):
raise OSError("Network connection failed during token retrieval")
try:
azure_identity.InteractiveBrowserCredential = MockCredentialWithTokenError
with pytest.raises(RuntimeError) as exc_info:
AADAuth.get_token("interactive")
# Verify the error message format (lines 54-56)
error_message = str(exc_info.value)
assert "Failed to create MockCredentialWithTokenError" in error_message
assert "Network connection failed during token retrieval" in error_message
# Verify exception chaining
assert exc_info.value.__cause__ is not None
assert isinstance(exc_info.value.__cause__, OSError)
finally:
# Restore original credential
azure_identity.InteractiveBrowserCredential = original_interactive
def test_get_token_various_exception_types_coverage(self):
"""Test coverage of different exception types (Lines 52-56)."""
import sys
azure_identity = sys.modules["azure.identity"]
# Store original credential
original_default = azure_identity.DefaultAzureCredential
# Test different exception types that could occur
exception_test_cases = [
(ImportError, "Required dependency missing"),
(AttributeError, "Missing required attribute"),
(RuntimeError, "Custom runtime error"),
]
for exception_type, exception_message in exception_test_cases:
class MockCredentialWithCustomError:
def __init__(self):
raise exception_type(exception_message)
try:
azure_identity.DefaultAzureCredential = MockCredentialWithCustomError
with pytest.raises(RuntimeError) as exc_info:
AADAuth.get_token("default")
# Verify the error message format (lines 54-56)
error_message = str(exc_info.value)
assert "Failed to create MockCredentialWithCustomError" in error_message
assert exception_message in error_message
# Verify exception chaining is preserved
assert exc_info.value.__cause__ is not None
assert isinstance(exc_info.value.__cause__, exception_type)
finally:
# Restore for next iteration
azure_identity.DefaultAzureCredential = original_default
class TestSilentFirstInteractive:
"""Regression tests for the silent-first pattern (interactive / device-code).
Contract: ``_acquire_token`` must try ``get_token()`` silently first and
invoke ``authenticate()`` ONLY when the SDK raises
``AuthenticationRequiredError`` — never on every acquisition. This guards
against reverting to the old "authenticate() on every checkout" behavior,
which would prompt the user (browser / device code) on each pooled reconnect.
"""
def test_silent_success_never_calls_authenticate(self):
"""When get_token() succeeds silently, authenticate() is never invoked."""
az = sys.modules["azure.identity"]
original = az.InteractiveBrowserCredential
class SilentCred:
def __init__(self, **kwargs):
self.authenticate_calls = 0
def authenticate(self, **kwargs):
self.authenticate_calls += 1
class R:
home_account_id = "acct"
return R()
def get_token(self, scope):
class T:
token = SAMPLE_TOKEN
return T()
try:
az.InteractiveBrowserCredential = SilentCred
AADAuth._acquire_token("interactive")
cred = _credential_cache["interactive"]
assert cred.authenticate_calls == 0, "authenticate() must not run on a silent success"
finally:
az.InteractiveBrowserCredential = original
def test_authentication_required_triggers_authenticate_once(self):
"""AuthenticationRequiredError from the silent get_token() falls back to
authenticate() exactly once, then retries get_token() successfully."""
az = sys.modules["azure.identity"]
original = az.DeviceCodeCredential
auth_required = az.AuthenticationRequiredError
class ChallengeCred:
def __init__(self, **kwargs):
self.authenticate_calls = 0
self.get_token_calls = 0
def authenticate(self, **kwargs):
self.authenticate_calls += 1
class R:
home_account_id = "device-acct"
return R()
def get_token(self, scope):
self.get_token_calls += 1
if self.get_token_calls == 1:
raise auth_required("interactive authentication required")
class T:
token = SAMPLE_TOKEN
return T()
try:
az.DeviceCodeCredential = ChallengeCred
_, _, _, home_account_id = AADAuth._acquire_token("devicecode")
cred = _credential_cache["devicecode"]
# Exactly one interactive step, and get_token tried twice
# (silent attempt that raised, then the post-authenticate retry).
assert cred.authenticate_calls == 1
assert cred.get_token_calls == 2
assert home_account_id == "device-acct"
finally:
az.DeviceCodeCredential = original
def test_subsequent_acquisitions_are_silent(self):
"""After the first interactive authenticate(), later acquisitions for
the same cached credential acquire silently and never re-prompt."""
az = sys.modules["azure.identity"]
original = az.InteractiveBrowserCredential
auth_required = az.AuthenticationRequiredError
class OnceChallengeCred:
def __init__(self, **kwargs):
self.authenticate_calls = 0
self.challenged = False
def authenticate(self, **kwargs):
self.authenticate_calls += 1
class R:
home_account_id = "acct"
return R()
def get_token(self, scope):
if not self.challenged:
self.challenged = True
raise auth_required("need interactive")
class T:
token = SAMPLE_TOKEN
return T()
try:
az.InteractiveBrowserCredential = OnceChallengeCred
AADAuth._acquire_token("interactive") # challenge -> authenticate()
AADAuth._acquire_token("interactive") # silent
AADAuth._acquire_token("interactive") # silent
cred = _credential_cache["interactive"]
assert cred.authenticate_calls == 1, "authenticate() must run only on the first prompt"
finally:
az.InteractiveBrowserCredential = original
def test_home_account_id_is_cached_and_reused(self):
"""The home_account_id resolved by the interactive step is cached in
_account_id_cache and surfaced on subsequent silent acquisitions."""
az = sys.modules["azure.identity"]
original = az.InteractiveBrowserCredential
auth_required = az.AuthenticationRequiredError
class OnceChallengeCred:
def __init__(self, **kwargs):
self.challenged = False
def authenticate(self, **kwargs):
class R:
home_account_id = "stable-acct"
return R()
def get_token(self, scope):
if not self.challenged:
self.challenged = True
raise auth_required("need interactive")
class T:
token = SAMPLE_TOKEN
return T()
try:
az.InteractiveBrowserCredential = OnceChallengeCred
_, _, _, hid1 = AADAuth._acquire_token("interactive")
# The cache now holds the account id for the interactive cache key.
assert "stable-acct" in _account_id_cache.values()
_, _, _, hid2 = AADAuth._acquire_token("interactive")
assert hid1 == "stable-acct"
assert hid2 == "stable-acct"
finally:
az.InteractiveBrowserCredential = original
class TestAuthenticateInteractive:
"""Direct branch coverage for the _authenticate_interactive fallback helper."""
def test_returns_none_when_credential_has_no_authenticate(self):
"""A credential/test double lacking authenticate() yields None so the
caller falls back to the token-hash pool key."""
class NoAuth:
pass
assert AADAuth._authenticate_interactive(NoAuth()) is None
def test_uses_scopes_keyword_when_supported(self):
captured = {}
class Cred:
def authenticate(self, **kwargs):
captured.update(kwargs)
class R:
home_account_id = "acct-1"
return R()
assert AADAuth._authenticate_interactive(Cred()) == "acct-1"
assert "scopes" in captured
def test_falls_back_to_no_args_when_scopes_rejected(self):
"""Older/mocked authenticate() signatures that reject the scopes
keyword are retried with no arguments."""
calls = {"with_scopes": 0, "no_args": 0}
class Cred:
def authenticate(self, *args, **kwargs):
if kwargs:
calls["with_scopes"] += 1
raise TypeError("unexpected keyword argument 'scopes'")
calls["no_args"] += 1
class R:
home_account_id = "acct-2"
return R()
assert AADAuth._authenticate_interactive(Cred()) == "acct-2"
assert calls["with_scopes"] == 1
assert calls["no_args"] == 1
def test_returns_none_when_record_lacks_home_account_id(self):
class Cred:
def authenticate(self, **kwargs):
return object() # no home_account_id attribute
assert AADAuth._authenticate_interactive(Cred()) is None
class TestProcessAuthParameters:
def test_empty_parameters(self):
auth_type = process_auth_parameters({})
assert auth_type is None
def test_interactive_auth_windows(self, monkeypatch):
monkeypatch.setattr(platform, "system", lambda: "Windows")
params = {"Authentication": "ActiveDirectoryInteractive", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type is None
def test_interactive_auth_non_windows(self, monkeypatch):
monkeypatch.setattr(platform, "system", lambda: "Darwin")
params = {"Authentication": "ActiveDirectoryInteractive", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type == "interactive"
def test_device_code_auth(self):
params = {"Authentication": "ActiveDirectoryDeviceCode", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type == "devicecode"
def test_default_auth(self):
params = {"Authentication": "ActiveDirectoryDefault", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type == "default"
def test_service_principal_auth_leaves_odbc_path_alone(self):
"""ServicePrincipal is handled natively by ODBC. process_auth_parameters
must return None so the ODBC path doesn't pre-acquire a token (which
would require tenant_id we don't have client-side). Bulkcopy still
gets "serviceprincipal" from extract_auth_type."""
params = {"Authentication": "ActiveDirectoryServicePrincipal", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type is None
def test_service_principal_auth_case_insensitive(self):
params = {"Authentication": "activedirectoryserviceprincipal", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type is None
def test_msi_auth(self):
params = {"Authentication": "ActiveDirectoryMSI", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type == "msi"
def test_msi_auth_case_insensitive(self):
params = {"Authentication": "activedirectorymsi", "Server": "test"}
auth_type = process_auth_parameters(params)
assert auth_type == "msi"
class TestRemoveSensitiveParams:
def test_remove_sensitive_parameters(self):
params = {
"Server": "test",
"UID": "user",
"PWD": "password",
"Encrypt": "yes",
"TrustServerCertificate": "yes",
"Authentication": "ActiveDirectoryDefault",
"Trusted_Connection": "yes",
"Database": "testdb",
}
filtered_params = remove_sensitive_params(params)
assert "Server" in filtered_params
assert "Database" in filtered_params
assert "UID" not in filtered_params
assert "PWD" not in filtered_params
assert "Encrypt" in filtered_params
assert "TrustServerCertificate" in filtered_params
assert "Trusted_Connection" not in filtered_params
assert "Authentication" not in filtered_params
class TestExtractAuthType:
def test_interactive(self):
assert (
extract_auth_type({"Server": "test", "Authentication": "ActiveDirectoryInteractive"})
== "interactive"
)
def test_default(self):
assert (
extract_auth_type({"Server": "test", "Authentication": "ActiveDirectoryDefault"})
== "default"
)
def test_devicecode(self):
assert (
extract_auth_type({"Server": "test", "Authentication": "ActiveDirectoryDeviceCode"})
== "devicecode"
)
def test_serviceprincipal(self):
assert (
extract_auth_type(
{"Server": "test", "Authentication": "ActiveDirectoryServicePrincipal"}
)
== "serviceprincipal"
)
def test_msi(self):
assert (
extract_auth_type({"Server": "test", "Authentication": "ActiveDirectoryMSI"}) == "msi"
)
def test_no_auth(self):
assert extract_auth_type({"Server": "test", "Database": "db"}) is None
def test_unsupported_auth(self):
assert extract_auth_type({"Server": "test", "Authentication": "SqlPassword"}) is None
class TestManagedIdentity:
"""Tests for ActiveDirectoryMSI support (system- and user-assigned)."""
def test_get_token_system_assigned_msi(self):
"""System-assigned MSI: ManagedIdentityCredential() constructed with no kwargs."""
az = sys.modules["azure.identity"]
az.ManagedIdentityCredential.last_init_kwargs = None
token_struct = AADAuth.get_token("msi")
assert isinstance(token_struct, bytes)
assert az.ManagedIdentityCredential.last_init_kwargs == {}
def test_get_raw_token_system_assigned_msi(self):
raw_token = AADAuth.get_raw_token("msi")
assert raw_token == SAMPLE_TOKEN
def test_get_token_user_assigned_msi(self):
"""User-assigned MSI: client_id is forwarded to the credential constructor."""
az = sys.modules["azure.identity"]
az.ManagedIdentityCredential.last_init_kwargs = None
client_id = "11111111-2222-3333-4444-555555555555"
token_struct = AADAuth.get_token("msi", {"client_id": client_id})
assert isinstance(token_struct, bytes)
assert az.ManagedIdentityCredential.last_init_kwargs == {"client_id": client_id}
def test_msi_separate_cache_entries_per_client_id(self):
"""System-assigned and user-assigned MSI must not share a cached credential."""
AADAuth.get_token("msi") # system-assigned
AADAuth.get_token("msi", {"client_id": "abc"})
AADAuth.get_token("msi", {"client_id": "def"})
# System-assigned uses the bare string key; user-assigned uses tuples.
assert "msi" in _credential_cache
assert ("msi", (("client_id", "abc"),)) in _credential_cache
assert ("msi", (("client_id", "def"),)) in _credential_cache
assert _credential_cache["msi"] is not _credential_cache[("msi", (("client_id", "abc"),))]
@patch("mssql_python.connection.ddbc_bindings.Connection")
def test_msi_auth_type_stored_on_connection(self, mock_ddbc_conn):
"""MSI with UID: Connection stores auth_type and credential_kwargs."""
mock_ddbc_conn.return_value = MagicMock()
from mssql_python import connect
az = sys.modules["azure.identity"]
az.ManagedIdentityCredential.last_init_kwargs = None
conn = connect(
"Server=test;Database=testdb;Authentication=ActiveDirectoryMSI;"
"UID=11111111-2222-3333-4444-555555555555"
)
assert conn._auth_type == "msi"
assert conn._credential_kwargs == {"client_id": "11111111-2222-3333-4444-555555555555"}
# UID must be stripped from the sanitized connection string
assert "UID=" not in conn.connection_str
conn.close()
@patch("mssql_python.connection.ddbc_bindings.Connection")
def test_msi_system_assigned_no_credential_kwargs(self, mock_ddbc_conn):
"""System-assigned MSI: no UID -> credential_kwargs is None."""
mock_ddbc_conn.return_value = MagicMock()
from mssql_python import connect
conn = connect("Server=test;Database=testdb;Authentication=ActiveDirectoryMSI")
assert conn._auth_type == "msi"
assert conn._credential_kwargs is None
conn.close()
@patch("mssql_python.connection.ddbc_bindings.Connection")
def test_msi_braced_uid_value_is_unwrapped(self, mock_ddbc_conn):
"""A braced UID value (UID={hello=world}) must be unwrapped by the
canonical _ConnectionStringParser; the inner '=' must NOT split."""
mock_ddbc_conn.return_value = MagicMock()
from mssql_python import connect
conn = connect(
"Server=test;Authentication=ActiveDirectoryMSI;" "UID={hello=world};Database=testdb"
)
assert conn._auth_type == "msi"
assert conn._credential_kwargs == {"client_id": "hello=world"}
conn.close()
@patch("mssql_python.connection.ddbc_bindings.Connection")
def test_msi_braced_uid_with_semicolon_is_preserved(self, mock_ddbc_conn):
"""A braced UID value containing a semicolon (legal under ODBC) must
be returned intact, not truncated at the inner ';'."""
mock_ddbc_conn.return_value = MagicMock()
from mssql_python import connect
conn = connect(
"Server=test;Authentication=ActiveDirectoryMSI;" "UID={abc;def;ghi};Database=testdb"
)
assert conn._auth_type == "msi"
assert conn._credential_kwargs == {"client_id": "abc;def;ghi"}
conn.close()
@patch("mssql_python.connection.ddbc_bindings.Connection")
def test_msi_without_uid_is_system_assigned(self, mock_ddbc_conn):
"""MSI without UID at all should be treated as system-assigned."""
mock_ddbc_conn.return_value = MagicMock()
from mssql_python import connect
conn = connect("Server=test;Authentication=ActiveDirectoryMSI;Database=testdb")
assert conn._auth_type == "msi"
assert conn._credential_kwargs is None
# UID should not appear in the connection string
assert "UID=" not in conn.connection_str
conn.close()
def test_bulkcopy_path_preserves_user_assigned_msi_client_id(self):
"""Regression test (cursor.bulkcopy() end-to-end) for the silent
system-assigned fallback: the bulkcopy fresh-token code path must
forward Connection._credential_kwargs to AADAuth.get_raw_token,
not re-parse the (now UID-stripped) connection_str.
Fails if cursor.py is reverted to call extract_credential_kwargs on
self.connection.connection_str, OR if Connection stops persisting
_credential_kwargs."""
from mssql_python.cursor import Cursor
client_id = "11111111-2222-3333-4444-555555555555"
# Mock Connection holding what Connection.__init__ would store after
# process_connection_string strips UID from the user-supplied string.
mock_conn = MagicMock()
# Post-sanitization string: NO UID. If cursor re-parses this, the
# forwarded kwargs will be {} and the assert below will fail.
mock_conn.connection_str = "Server=tcp:test.database.windows.net;Database=testdb;"
mock_conn._auth_type = "msi"
mock_conn._credential_kwargs = {"client_id": client_id}
mock_conn._token_provider = None
mock_conn._is_connected = True
cursor = Cursor.__new__(Cursor)
cursor._connection = mock_conn
cursor._timeout = 0
cursor.closed = False
cursor.hstmt = None
captured = {}
def fake_get_raw_token(auth_type, credential_kwargs=None):
captured["auth_type"] = auth_type
captured["credential_kwargs"] = credential_kwargs
return SAMPLE_TOKEN
mock_pycore_cursor = MagicMock()
mock_pycore_cursor.bulkcopy.return_value = {
"rows_copied": 1,
"batch_count": 1,
"elapsed_time": 0.1,
}
mock_pycore_conn = MagicMock()
mock_pycore_conn.cursor.return_value = mock_pycore_cursor
mock_pycore_module = MagicMock()
mock_pycore_module.PyCoreConnection = lambda ctx, **kwargs: mock_pycore_conn
with (
patch.dict("sys.modules", {"mssql_py_core": mock_pycore_module}),
patch("mssql_python.auth.AADAuth.get_raw_token", side_effect=fake_get_raw_token),
):
cursor.bulkcopy("dbo.test_table", [(1, "row")], timeout=10)
assert captured["auth_type"] == "msi"
assert captured["credential_kwargs"] == {"client_id": client_id}, (
f"bulkcopy must forward Connection._credential_kwargs verbatim; "
f"got {captured['credential_kwargs']!r}. If this is {{}} or None, "
f"the cursor likely re-parses the (UID-stripped) connection_str."
)
class TestCredentialInstanceCache:
"""Tests for the credential instance caching behavior."""
def test_credential_reused_across_calls(self):
"""The same credential instance should be returned for repeated calls."""
AADAuth.get_token("default")
assert "default" in _credential_cache
first_instance = _credential_cache["default"]
AADAuth.get_token("default")
assert _credential_cache["default"] is first_instance
def test_different_auth_types_get_separate_instances(self):
"""Each auth type should have its own cached credential."""
AADAuth.get_token("default")
AADAuth.get_token("devicecode")
assert "default" in _credential_cache
assert "devicecode" in _credential_cache
assert _credential_cache["default"] is not _credential_cache["devicecode"]
def test_get_raw_token_uses_cached_credential(self):
"""get_raw_token should also use the cached credential instance."""
AADAuth.get_token("default")
cached = _credential_cache["default"]
AADAuth.get_raw_token("default")
assert _credential_cache["default"] is cached
def test_cache_starts_empty(self):
"""Cache should be empty at the start due to the clear_credential_cache fixture."""
assert len(_credential_cache) == 0
def test_cached_credential_refreshes_token_after_expiry(self):
"""Verify that the cached credential instance returns fresh tokens on each call.
This simulates what happens when Azure Identity SDK refreshes an expired
token internally: because we cache the credential (not the token), each
_acquire_token() call invokes get_token() on the same instance, giving
the SDK the opportunity to return a refreshed token when the old one has
expired.
"""
import sys
azure_identity = sys.modules["azure.identity"]
original = azure_identity.DefaultAzureCredential
call_count = 0
tokens = ["initial_token_abc123", "refreshed_token_xyz789"]
class MockCredentialWithRefresh:
def get_token(self, scope):
nonlocal call_count
idx = min(call_count, len(tokens) - 1)
call_count += 1
class Token:
token = tokens[idx]
return Token()
try:
azure_identity.DefaultAzureCredential = MockCredentialWithRefresh
# First call — gets initial token
_, raw_token_1, _, _ = AADAuth._acquire_token("default")
assert raw_token_1 == "initial_token_abc123"
assert call_count == 1
# Same credential instance is cached
cached = _credential_cache["default"]
assert isinstance(cached, MockCredentialWithRefresh)
# Second call — same credential instance, but SDK returns refreshed token
# (simulating post-expiry refresh)
_, raw_token_2, _, _ = AADAuth._acquire_token("default")
assert raw_token_2 == "refreshed_token_xyz789"
assert call_count == 2
# Credential instance is still the same (not recreated)
assert _credential_cache["default"] is cached
finally:
azure_identity.DefaultAzureCredential = original
class TestAcquireTokenImportError:
"""Test the ImportError path when azure-identity is not installed."""
def test_import_error_raises_runtime_error(self):
"""_acquire_token raises RuntimeError when azure.identity is missing."""
import sys
# Temporarily remove the mocked azure modules
saved = {}
for mod_name in list(sys.modules):
if mod_name == "azure" or mod_name.startswith("azure."):
saved[mod_name] = sys.modules.pop(mod_name)
# Make the import fail
import builtins
real_import = builtins.__import__
def blocked_import(name, *args, **kwargs):
if name.startswith("azure"):
raise ImportError("No module named 'azure'")
return real_import(name, *args, **kwargs)
You can’t perform that action at this time.