123456789101112131415161718192021222324 |
- def test_conn_method_aad_app_cert(self, mock_init):
- mock_init.return_value = None
- db.merge_conn(
- Connection(
- conn_id=ADX_TEST_CONN_ID,
- conn_type='azure_data_explorer',
- login='client_id',
- host='https://help.kusto.windows.net',
- extra=json.dumps(
- {
- 'extra__azure_data_explorer__tenant': 'tenant',
- 'extra__azure_data_explorer__auth_method': 'AAD_APP_CERT',
- 'extra__azure_data_explorer__certificate': 'PEM',
- 'extra__azure_data_explorer__thumbprint': 'thumbprint',
- }
- ),
- )
- )
- AzureDataExplorerHook(azure_data_explorer_conn_id=ADX_TEST_CONN_ID)
- assert mock_init.called_with(
- KustoConnectionStringBuilder.with_aad_application_certificate_authentication(
- 'https://help.kusto.windows.net', 'client_id', 'PEM', 'thumbprint', 'tenant'
- )
- )
|