Add: copy SSH keys at the end of zapper_iot provisioning (#446)

canonical · Jan 15, 2025 · 167ed0f · 167ed0f
1 parent 5f9fc23
commit 167ed0f
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 29 deletions.
diff --git a/device-connectors/src/testflinger_device_connectors/devices/zapper/__init__.py b/device-connectors/src/testflinger_device_connectors/devices/zapper/__init__.py
@@ -23,6 +23,7 @@
 
 import json
 import logging
+import subprocess
 from abc import ABC, abstractmethod
 from typing import Any, Dict, Tuple
 
@@ -103,5 +104,35 @@ def _run(self, zapper_ip, *args, **kwargs):
             **kwargs,
         )
 
+    def _copy_ssh_id(self):
+        """Copy the ssh id to the device"""
+
+        logger.info("Copying the agent's SSH public key to the DUT.")
+
+        try:
+            test_username = self.job_data.get("test_data", {}).get(
+                "test_username", "ubuntu"
+            )
+            test_password = self.job_data.get("test_data", {}).get(
+                "test_password", "ubuntu"
+            )
+        except AttributeError:
+            test_username = "ubuntu"
+            test_password = "ubuntu"
+
+        cmd = [
+            "sshpass",
+            "-p",
+            test_password,
+            "ssh-copy-id",
+            "-o",
+            "StrictHostKeyChecking=no",
+            "-o",
+            "UserKnownHostsFile=/dev/null",
+            f"{test_username}@{self.config['device_ip']}",
+        ]
+        subprocess.check_call(cmd, timeout=60)
+
+    @abstractmethod
     def _post_run_actions(self, args):
         """Run further actions after Zapper API returns successfully."""
diff --git a/device-connectors/src/testflinger_device_connectors/devices/zapper/tests/test_zapper.py b/device-connectors/src/testflinger_device_connectors/devices/zapper/tests/test_zapper.py
@@ -28,6 +28,9 @@ class MockConnector(ZapperConnector):
     def _validate_configuration(self):
         return (), {}
 
+    def _post_run_actions(self, args):
+        pass
+
 
 class ZapperConnectorTests(unittest.TestCase):
     """Unit tests for ZapperConnector class."""

diff --git a/device-connectors/src/testflinger_device_connectors/devices/zapper_iot/__init__.py b/device-connectors/src/testflinger_device_connectors/devices/zapper_iot/__init__.py
@@ -75,3 +75,8 @@ def _validate_configuration(
         provisioning_data["urls"] = urls
 
         return ((), provisioning_data)
+
+    def _post_run_actions(self, args):
+        """Run further actions after Zapper API returns successfully."""
+
+        self._copy_ssh_id()
diff --git a/device-connectors/src/testflinger_device_connectors/devices/zapper_kvm/__init__.py b/device-connectors/src/testflinger_device_connectors/devices/zapper_kvm/__init__.py
@@ -185,35 +185,6 @@ def _run_oem_script(self, args):
 
         oemscript.provision()
 
-    def _copy_ssh_id(self):
-        """Copy the ssh id to the device"""
-
-        logger.info("Copying the agent's SSH public key to the DUT.")
-
-        try:
-            test_username = self.job_data.get("test_data", {}).get(
-                "test_username", "ubuntu"
-            )
-            test_password = self.job_data.get("test_data", {}).get(
-                "test_password", "ubuntu"
-            )
-        except AttributeError:
-            test_username = "ubuntu"
-            test_password = "ubuntu"
-
-        cmd = [
-            "sshpass",
-            "-p",
-            test_password,
-            "ssh-copy-id",
-            "-o",
-            "StrictHostKeyChecking=no",
-            "-o",
-            "UserKnownHostsFile=/dev/null",
-            f"{test_username}@{self.config['device_ip']}",
-        ]
-        subprocess.check_output(cmd, stderr=subprocess.STDOUT, timeout=60)
-
     def _change_password(self, username, orig_password):
         """Change password via SSH to the one specified in the job data."""