Implement a RebootJob

Closes #21

Author: thatbudakguy

mqterm/jobs.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from binascii import hexlify
 from hashlib import sha256
@@ -247,6 +248,38 @@ def output(self):
         return BytesIO(str(self.bytes_written).encode("utf-8"))
 
 
+class RebootJob(Job):
+    """A job to perform a hard or soft reboot of the device."""
+
+    argc = 1
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.mode = self.args[0]
+
+    def output(self):
+        """Reboot the device after a given delay."""
+        import machine
+
+        # Schedule reboot in one second
+        async def reboot_callback():
+            await asyncio.sleep(1)
+            try:
+                machine.reset() if self.mode == "hard" else machine.soft_reset()
+            except AttributeError:
+                logging.error("Operation not supported on this platform")
+
+        asyncio.create_task(reboot_callback())
+
+        # Log the reboot action and return as output
+        msg = f"Performing {self.mode} reboot"
+        if self.mode == "hard":
+            logging.critical(msg)
+        else:
+            logging.warning(msg)
+        return BytesIO(msg.encode("utf-8"))
+
+
 # Map commands to associated job names
 COMMANDS = {
     "whoami": WhoAmIJob,
@@ -255,4 +288,5 @@ def output(self):
     "ls": ListDirJob,
     "cp": PutFileJob,
     "ota": FirmwareUpdateJob,
+    "reboot": RebootJob,
 }

tests/test_jobs.py

@@ -12,6 +12,7 @@
     Job,
     PlatformInfoJob,
     PutFileJob,
+    RebootJob,
     WhoAmIJob,
 )
 
@@ -169,7 +170,7 @@ def test_wait_partial_block(self):
     def test_last_block_fill(self):
         """Should fill space in last block with empty data on final update"""
         initial_data = b"\xde\xad\xbe\xef"
-        checksum = hexlify(sha256(initial_data).digest()).decode('utf-8')
+        checksum = hexlify(sha256(initial_data).digest()).decode("utf-8")
         job = FirmwareUpdateJob("ota", [checksum])
 
         # Send and complete the update
@@ -195,7 +196,7 @@ def test_last_block_fill(self):
     def test_output(self):
         """Should return the total bytes written as output"""
         initial_data = b"\xde\xad\xbe\xef"
-        checksum = hexlify(sha256(initial_data).digest()).decode('utf-8')
+        checksum = hexlify(sha256(initial_data).digest()).decode("utf-8")
         job = FirmwareUpdateJob("ota", [checksum])
 
         # Send and complete the update
@@ -205,3 +206,17 @@ def test_output(self):
         # Output should be the total bytes written
         output = job.output().read().decode("utf-8").strip()
         self.assertEqual(output, "4")
+
+
+class TestRebootJob(TestCase):
+    def test_run_hard(self):
+        """Reboot job should signal and perform a hard reboot"""
+        job = RebootJob("reboot", ["hard"])
+        output = job.output().read().decode("utf-8").strip()
+        self.assertEqual(output, "Performing hard reboot")
+
+    def test_run_soft(self):
+        """Reboot job should signal a soft reboot"""
+        job = RebootJob("reboot", ["soft"])
+        output = job.output().read().decode("utf-8").strip()
+        self.assertEqual(output, "Performing soft reboot")