add unittest for common/ssh_client package

.github/workflows/test_ssh_client.yml

@@ -0,0 +1,30 @@
+name: Test Ssh Client
+
+on:
+  push:
+    branches: "*"
+  pull_request:
+    branches: "*"
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        fetch-depth: 0 # Fetch all history for proper version detection
+
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v3
+      with:
+        python-version: 3.8
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements3.txt
+
+    - name: Run tests
+      run: python -m unittest discover -s test/common/ssh_client -p 'test_*.py'

common/ssh_client/kubernetes_client.py

@@ -42,11 +42,14 @@ def __init__(self, context=None, node=None):
     def exec_cmd(self, cmd):
         exec_command = ['/bin/sh', '-c', cmd]
         self.stdio.verbose("KubernetesClient exec_cmd: {0}".format(cmd))
-        resp = stream(self.client.connect_get_namespaced_pod_exec, self.pod_name, self.namespace, command=exec_command, stderr=True, stdin=False, stdout=True, tty=False, container=self.container_name)
-        self.stdio.verbose("KubernetesClient exec_cmd.resp: {0}".format(resp))
-        if "init system (PID 1). Can't operate." in resp:
-            return "KubernetesClient can't get the resp by {0}".format(cmd)
-        return resp
+        try:
+            resp = stream(self.client.connect_get_namespaced_pod_exec, self.pod_name, self.namespace, command=exec_command, stderr=True, stdin=False, stdout=True, tty=False, container=self.container_name)
+            self.stdio.verbose("KubernetesClient exec_cmd.resp: {0}".format(resp))
+            if "init system (PID 1). Can't operate." in resp:
+                return "KubernetesClient can't get the resp by {0}".format(cmd)
+            return resp
+        except Exception as e:
+            return f"KubernetesClient can't get the resp by {cmd}: {str(e)}"
 
     def download(self, remote_path, local_path):
         return self.__download_file_from_pod(self.namespace, self.pod_name, self.container_name, remote_path, local_path)