fix

Author: FumingZhang

src/aks-preview/azext_aks_preview/bastion/bastion.py

@@ -9,7 +9,6 @@
 import socket
 import subprocess
 import sys
-import time
 from urllib.parse import urlparse
 
 import psutil

src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py

@@ -23,7 +23,6 @@
     use_shared_identity,
 )
 from azure.cli.core.azclierror import ClientRequestError
-from azure.core.exceptions import (HttpResponseError)
 from azure.cli.testsdk import CliTestError, ScenarioTest, live_only
 from azure.cli.testsdk.scenario_tests import AllowLargeResponse
 from knack.util import CLIError
@@ -179,6 +178,21 @@ def _get_asm_upgrade_version(self, resource_group, name):
         sorted_upgrades = self._sort_revisions(res["upgrades"])
         return sorted_upgrades[0]
 
+    def _verify_kubectl_installation(self) -> bool:
+        """Verify if kubectl is installed and accessible."""
+        try:
+            subprocess.run(
+                ["kubectl", "version", "--client"],
+                check=True,
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                shell=False,
+            )
+            return True
+        except subprocess.CalledProcessError:
+            return False
+
     @classmethod
     def generate_ssh_keys(cls):
         # If the `--ssh-key-value` option is not specified, the validator will try to read the ssh-key from the "~/.ssh" directory,
@@ -10948,11 +10962,12 @@ def test_aks_kollect_with_managed_aad(
         sp_oid = self._get_test_identity_object_id()
         print(f"objectid of service principal is {sp_oid}")
 
-        # Install kubectl (for setting up service principal permissions, and required by the 'kollect' command).
-        try:
-            subprocess.call(["az", "aks", "install-cli"])
-        except subprocess.CalledProcessError as err:
-            raise CliTestError(f"Failed to install kubectl with error: '{err}'")
+        if not self._verify_kubectl_installation():
+            # Install kubectl (for setting up service principal permissions, and required by the 'kollect' command).
+            try:
+                subprocess.call(["az", "aks", "install-cli"])
+            except subprocess.CalledProcessError as err:
+                raise CliTestError(f"Failed to install kubectl with error: '{err}'")
 
         # Grant the service principal cluster-admin access using the admin account
         # (it'd be nice if `az aks command invoke` had an --admin option, but it appears not to, so we have to download admin credentials)
@@ -12679,7 +12694,7 @@ def test_aks_azure_service_mesh_with_egress_gateway(
         try:
             subprocess.call(["az", "aks", "install-cli"])
         except subprocess.CalledProcessError as err:
-            raise CLITestError("Failed to install kubectl with error: '{}'!".format(err))
+            raise CliTestError("Failed to install kubectl with error: '{}'!".format(err))
 
         try:
             # get credential
@@ -12717,7 +12732,7 @@ def test_aks_azure_service_mesh_with_egress_gateway(
                     stderr=subprocess.STDOUT,
                 )
                 if not f"namespace/{istio_egress_namespace} created" in k_create_sgc_namespace_output:
-                    raise CLITestError(f"failed to create istio egress gateway namespace: {istio_egress_namespace}")
+                    raise CliTestError(f"failed to create istio egress gateway namespace: {istio_egress_namespace}")
 
                 k_create_sgc_command = ["kubectl", "apply", "-f", sgc_browse_path, "--kubeconfig", browse_path]
                 k_create_sgc_output = subprocess.check_output(
@@ -12726,7 +12741,7 @@ def test_aks_azure_service_mesh_with_egress_gateway(
                     stderr=subprocess.STDOUT,
                 )
                 if not f"staticgatewayconfiguration.egressgateway.kubernetes.azure.com/{istio_sgc_name} created" in k_create_sgc_output:
-                    raise CLITestError("failed to create StaticGatewayConfiguration")
+                    raise CliTestError("failed to create StaticGatewayConfiguration")
             finally:
                 # Delete files
                 if os.path.exists(browse_path):
@@ -16381,16 +16396,18 @@ def test_aks_bastion(self, resource_group, resource_group_location):
                               f"--vnet-name {vnet_name} --enable-tunneling"
         self.cmd(create_bastion_cmd, checks=[self.check("provisioningState", "Succeeded")])
 
-        # install kubectl
-        _, ctl_temp_file = tempfile.mkstemp()
-        _, login_temp_file = tempfile.mkstemp()
-        version = "latest"
-        install_cmd = 'aks install-cli --client-version={} --install-location={} --base-src-url={} ' \
-                    '--kubelogin-version={} --kubelogin-install-location={} --kubelogin-base-src-url={}'.format(version, ctl_temp_file, "", version, login_temp_file, "")
-        self.cmd(install_cmd, checks=[self.is_empty()])
+        kubectl_path = "kubectl"
+        if not self._verify_kubectl_installation():
+            # install kubectl
+            _, kubectl_path = tempfile.mkstemp()
+            _, login_temp_file = tempfile.mkstemp()
+            version = "latest"
+            install_cmd = 'aks install-cli --client-version={} --install-location={} --base-src-url={} ' \
+                        '--kubelogin-version={} --kubelogin-install-location={} --kubelogin-base-src-url={}'.format(version, kubectl_path, "", version, login_temp_file, "")
+            self.cmd(install_cmd, checks=[self.is_empty()])
 
         # test bastion connectivity
-        os.environ["AKS_BASTION_TEST_HOOK"] = ctl_temp_file
+        os.environ["AKS_BASTION_TEST_HOOK"] = kubectl_path
         bastion_cmd = f"aks bastion -g {resource_group} -n {aks_name}"
         self.cmd(bastion_cmd, checks=[self.is_empty()])