Add firmware_update phase

agent/testflinger_agent/agent.py

@@ -123,7 +123,14 @@ def mark_device_offline(self):
 
     def process_jobs(self):
         """Coordinate checking for new jobs and handling them if they exists"""
-        TEST_PHASES = ["setup", "provision", "test", "allocate", "reserve"]
+        TEST_PHASES = [
+            "setup",
+            "provision",
+            "firmware_update",
+            "test",
+            "allocate",
+            "reserve",
+        ]
 
         # First, see if we have any old results that we couldn't send last time
         self.retry_old_results()

agent/testflinger_agent/job.py

@@ -59,6 +59,13 @@ def run_test_phase(self, phase, rundir):
         if phase == "provision" and not self.job_data.get("provision_data"):
             logger.info("No provision_data defined in job data, skipping...")
             return 0
+        if phase == "firmware_update" and not self.job_data.get(
+            "firmware_update_data"
+        ):
+            logger.info(
+                "No firmware_update_data defined in job data, skipping..."
+            )
+            return 0
         if phase == "test" and not self.job_data.get("test_data"):
             logger.info("No test_data defined in job data, skipping...")
             return 0

agent/testflinger_agent/schema.py

@@ -33,6 +33,7 @@
     voluptuous.Required("job_queues"): list,
     voluptuous.Required("setup_command", default=""): str,
     voluptuous.Required("provision_command", default=""): str,
+    voluptuous.Required("firmware_update_command", default=""): str,
     voluptuous.Required("test_command", default=""): str,
     voluptuous.Required("allocate_command", default=""): str,
     voluptuous.Required("reserve_command", default=""): str,

device-connectors/src/testflinger_device_connectors/cmd.py

@@ -39,6 +39,7 @@ def main():
         cmd_subparser = dev_subparser.add_subparsers()
         for cmd, func in (
             ("provision", dev_module.provision),
+            ("firmware_update", dev_module.firmware_update),
             ("runtest", dev_module.runtest),
             ("allocate", dev_module.allocate),
             ("reserve", dev_module.reserve),

device-connectors/src/testflinger_device_connectors/devices/__init__.py

@@ -26,6 +26,9 @@
 import yaml
 
 import testflinger_device_connectors
+from testflinger_device_connectors.fw_devices.firmware_update import (
+    detect_device,
+)
 
 
 class ProvisioningError(Exception):
@@ -115,6 +118,59 @@ def stop(self):
 
 
 class DefaultDevice:
+    def firmware_update(self, args):
+        """Default method for processing firmware update commands"""
+        with open(args.config) as configfile:
+            config = yaml.safe_load(configfile)
+        testflinger_device_connectors.configure_logging(config)
+        testflinger_device_connectors.logmsg(
+            logging.INFO, "BEGIN firmware_update"
+        )
+
+        test_opportunity = testflinger_device_connectors.get_test_opportunity(
+            args.job_data
+        )
+        fw_config = test_opportunity.get("firmware_update_data")
+        ignore_failure = fw_config.get("ignore_failure", False)
+        version = fw_config.get("version")
+        device_ip = config["device_ip"]
+        target_device_username = "ubuntu"
+        exitcode = 0
+        supported_version = ["latest"]
+
+        if version not in supported_version:
+            testflinger_device_connectors.logmsg(
+                logging.INFO,
+                "Fail to provide version in firmware_update_data. "
+                + "Current supported version: latest",
+            )
+            exitcode = 1
+        else:
+            try:
+                target_device = detect_device(
+                    device_ip, target_device_username
+                )
+                target_device.get_fw_info()
+                if version == "latest":
+                    reboot_required = target_device.upgrade()
+                if reboot_required:
+                    target_device.reboot()
+                    update_succeeded = target_device.check_results()
+                    if not update_succeeded:
+                        exitcode = 1
+            except Exception as e:
+                testflinger_device_connectors.logmsg(
+                    logging.ERROR, f"Firmware Update failed: {str(e)}"
+                )
+                exitcode = 1
+            finally:
+                testflinger_device_connectors.logmsg(
+                    logging.INFO, "END firmware_update"
+                )
+        if ignore_failure:
+            exitcode = 0
+        return exitcode
+
     def runtest(self, args):
         """Default method for processing test commands"""
         with open(args.config) as configfile: