Add test_connection method to AzureContainerInstanceHook

airflow/providers/microsoft/azure/hooks/container_instance.py

@@ -36,7 +36,7 @@ class AzureContainerInstanceHook(AzureBaseHook):
     client_id (Application ID) as login, the generated password as password,
     and tenantId and subscriptionId in the extra's field as a json.
 
-    :param conn_id: :ref:`Azure connection id<howto/connection:azure>` of
+    :param azure_conn_id: :ref:`Azure connection id<howto/connection:azure>` of
         a service principal which will be used to start the container instance.
     """
 
@@ -45,8 +45,8 @@ class AzureContainerInstanceHook(AzureBaseHook):
     conn_type = 'azure_container_instance'
     hook_name = 'Azure Container Instance'
 
-    def __init__(self, conn_id: str = default_conn_name) -> None:
-        super().__init__(sdk_client=ContainerInstanceManagementClient, conn_id=conn_id)
+    def __init__(self, azure_conn_id: str = default_conn_name) -> None:
+        super().__init__(sdk_client=ContainerInstanceManagementClient, conn_id=azure_conn_id)
         self.connection = self.get_conn()
 
     def create_or_update(self, resource_group: str, name: str, container_group: ContainerGroup) -> None:
@@ -138,3 +138,15 @@ def exists(self, resource_group: str, name: str) -> bool:
             if container.name == name:
                 return True
         return False
+
+    def test_connection(self):
+        """Test a configured Azure Container Instance connection."""
+        try:
+            # Attempt to list existing container groups under the configured subscription and retrieve the
+            # first in the returned iterator. We need to _actually_ try to retrieve an object to properly
+            # test the connection.
+            next(self.connection.container_groups.list(), None)
+        except Exception as e:
+            return False, str(e)
+
+        return True, "Successfully connected to Azure Container Instance."

airflow/providers/microsoft/azure/operators/container_instances.py

@@ -184,7 +184,7 @@ def execute(self, context: "Context") -> int:
         # Check name again in case it was templated.
         self._check_name(self.name)
 
-        self._ci_hook = AzureContainerInstanceHook(conn_id=self.ci_conn_id)
+        self._ci_hook = AzureContainerInstanceHook(azure_conn_id=self.ci_conn_id)
 
         if self.fail_if_exists:
             self.log.info("Testing if container group already exists")

tests/providers/microsoft/azure/hooks/test_azure_container_instance.py

@@ -50,7 +50,7 @@ def setUp(self):
             'azure.common.credentials.ServicePrincipalCredentials.__init__', autospec=True, return_value=None
         ):
             with patch('azure.mgmt.containerinstance.ContainerInstanceManagementClient'):
-                self.hook = AzureContainerInstanceHook(conn_id='azure_container_instance_test')
+                self.hook = AzureContainerInstanceHook(azure_conn_id='azure_container_instance_test')
 
     @patch('azure.mgmt.containerinstance.models.ContainerGroup')
     @patch('azure.mgmt.containerinstance.operations.ContainerGroupsOperations.create_or_update')
@@ -97,3 +97,17 @@ def test_exists_with_not_existing(self, list_mock):
             )
         ]
         assert not self.hook.exists('test', 'not found')
+
+    @patch('azure.mgmt.containerinstance.operations.ContainerGroupsOperations.list')
+    def test_connection_success(self, mock_container_groups_list):
+        mock_container_groups_list.return_value = iter([])
+        status, msg = self.hook.test_connection()
+        assert status is True
+        assert msg == "Successfully connected to Azure Container Instance."
+
+    @patch('azure.mgmt.containerinstance.operations.ContainerGroupsOperations.list')
+    def test_connection_failure(self, mock_container_groups_list):
+        mock_container_groups_list.side_effect = Exception("Authentication failed.")
+        status, msg = self.hook.test_connection()
+        assert status is False
+        assert msg == "Authentication failed."