Implement the RunPyJob for eval/exec

Closes #23

Author: thatbudakguy

mqterm/jobs.py

@@ -4,6 +4,14 @@
 from hashlib import sha256
 from io import BytesIO
 
+try:
+    from os import dupterm
+except ImportError:
+    # unix mpy doesn't have dupterm; define a no-op version
+    def dupterm(stream_object, index=0):
+        return stream_object
+
+
 from micropython import const
 
 from mqterm import VERSION
@@ -22,6 +30,7 @@ def __init__(self, cmd, args=[], client_id="localhost", **kwargs):
                 f"Wrong number of arguments for {cmd}; expected {self.argc}"
             )
 
+        self.globals = kwargs.get("globals", {})
         self.cmd = cmd
         self.args = args
         self.client_id = client_id
@@ -45,13 +54,31 @@ def ready(self):
         return True
 
     @classmethod
-    def from_cmd(cls, cmd_str, client_id=None):
-        """Create a job from a command string, e.g. 'get_file file1.txt'."""
-        cmd, *args = cmd_str.split()
+    def from_cmd(cls, cmd_str, client_id=None, globals={}):
+        """Create a job from a command string, e.g. 'cat file1.txt'."""
+        # Split command string into command and following arguments
+        try:
+            cmd, remainder = cmd_str.split(" ", 1)
+        except ValueError:
+            cmd, remainder = cmd_str, None
+
+        # Lookup command in command table
         if cmd not in COMMANDS:
             raise ValueError(f"Unknown command: '{cmd}'")
         job_cls = COMMANDS[cmd]
-        return job_cls(cmd, args, client_id)
+
+        # For eval, preserve the remainder as a single string argument and de-quote it
+        # Otherwise, split remainder into separate arguments
+        if remainder:
+            if cmd == "eval":
+                args = [remainder.strip("\"'")]
+            else:
+                args = remainder.split(" ")
+        else:
+            args = []
+
+        # Create and return the job instance
+        return job_cls(cmd, args, client_id, globals=globals)
 
 
 class SequentialJob(Job):
@@ -266,10 +293,7 @@ def output(self):
             op = machine.reset if self.mode == "hard" else machine.soft_reset
         except AttributeError:
             raise OSError("Operation not supported on this platform")
-        if self.mode == "hard":
-            logging.critical(msg)
-        else:
-            logging.warning(msg)
+        logging.critical(msg)
 
         # Schedule reboot in three seconds
         async def reboot_callback(op):
@@ -282,6 +306,41 @@ async def reboot_callback(op):
         return BytesIO(msg.encode("utf-8"))
 
 
+class RunPyJob(Job):
+    """A job to evaluate Python script on the device."""
+
+    argc = 1
+
+    def output(self):
+        """Eval or exec given Python and return the result."""
+        expr = self.args[0]
+        try:
+            result = self.do_eval(expr)
+        except SyntaxError:  # Not an expression, try exec
+            result = self.do_exec(expr)
+        if isinstance(result, str):  # Ensure bytes output
+            result = result.encode("utf-8")
+        return BytesIO(result)
+
+    def do_eval(self, expr):
+        """Evaluate a Python expression and return the result."""
+        op = compile(expr, "<string>", "eval")
+        result = eval(op, self.globals, None)
+        return repr(result)
+
+    def do_exec(self, expr):
+        """Execute a Python statement and return the output."""
+        out_buf = BytesIO()
+        old_term = dupterm(out_buf)
+        try:
+            op = compile(expr, "<string>", "exec")
+            exec(op, self.globals, None)
+            result = out_buf.getvalue().strip()
+        finally:
+            dupterm(old_term)
+        return result
+
+
 # Map commands to associated job names
 COMMANDS = {
     "whoami": WhoAmIJob,
@@ -291,4 +350,5 @@ async def reboot_callback(op):
     "cp": PutFileJob,
     "ota": FirmwareUpdateJob,
     "reboot": RebootJob,
+    "eval": RunPyJob,
 }

mqterm/terminal.py

@@ -44,7 +44,11 @@ class MqttTerminal:
     BUFLEN = PKTLEN * 2  # payload size for MQTT messages
 
     def __init__(
-        self, mqtt_client, topic_prefix=None, logger=logging.getLogger("mqterm")
+        self,
+        mqtt_client,
+        topic_prefix=None,
+        logger=logging.getLogger("mqterm"),
+        globals={},
     ):
         self.mqtt_client = mqtt_client
         self.topic_prefix = topic_prefix
@@ -55,6 +59,7 @@ def __init__(
         self.out_view = memoryview(self.out_buffer)
         self.logger = logger
         self.jobs = {}
+        self.globals = globals
 
     async def connect(self):
         """Start processing messages in the input stream."""
@@ -102,7 +107,7 @@ async def update_job(self, client_id, seq, payload):
             self.logger.debug(f"Updated {job}, seq: {seq}")
         else:
             cmd = payload.decode("utf-8")
-            job = Job.from_cmd(cmd, client_id=client_id)
+            job = Job.from_cmd(cmd, client_id=client_id, globals=self.globals)
             self.jobs[client_id] = job
             self.logger.info(f"Created {job}")
 

tests/test_jobs.py

@@ -14,8 +14,10 @@
     PlatformInfoJob,
     PutFileJob,
     RebootJob,
+    RunPyJob,
     WhoAmIJob,
 )
+from tests.utils import Mock
 
 
 class TestJob(TestCase):
@@ -28,6 +30,20 @@ def test_from_cmd(self):
         with self.assertRaises(ValueError):
             Job.from_cmd("unknown")
 
+    def test_from_cmd_eval(self):
+        """Job should handle eval command"""
+        job = Job.from_cmd("eval '1 + 2'")
+        self.assertEqual(job.cmd, "eval")
+        self.assertEqual(job.args, ["1 + 2"])
+        self.assertIsInstance(job, RunPyJob)
+
+    def test_from_cmd_no_args(self):
+        """Job should handle commands with no arguments"""
+        job = Job.from_cmd("uname")
+        self.assertEqual(job.cmd, "uname")
+        self.assertEqual(job.args, [])
+        self.assertIsInstance(job, PlatformInfoJob)
+
     def test_str(self):
         """Job should have a string representation"""
         job = Job("cat", args=["file.txt"])
@@ -212,11 +228,11 @@ class TestRebootJob(TestCase):
     def setUp(self):
         # Turn off logging during tests
         self.logger = logging.getLogger()
-        self.old_level = self.logger.level
-        self.logger.setLevel(logging.CRITICAL)
+        self.handlers = self.logger.handlers[:]
+        self.logger.handlers = []
 
     def tearDown(self):
-        self.logger.setLevel(self.old_level)
+        self.logger.handlers = self.handlers
 
     def test_run_hard(self):
         """Reboot job should signal and perform a hard reboot"""
@@ -230,3 +246,43 @@ def test_run_soft(self):
         job = RebootJob("reboot", ["soft"])
         output = job.output().read().decode("utf-8").strip()
         self.assertEqual(output, "Performing soft reboot")
+
+
+class TestRunPyJob(TestCase):
+    # def setUp(self):
+    #     # Mock os.dupterm; not present in unix micropython
+    #     self.mock_dupterm = Mock()
+    #     try:
+    #         setattr(os, "dupterm", self.mock_dupterm)
+    #     except AttributeError:  # raises in CI only??
+    #         pass
+
+    # def tearDown(self):
+    #     # Remove mock
+    #     if hasattr(os, "dupterm"):
+    #         delattr(os, "dupterm")
+
+    def test_eval(self):
+        """RunPyJob should evaluate a Python expression and return result"""
+        job = RunPyJob("eval", ["1 + 2"])
+        output = job.output().read().decode("utf-8").strip()
+        self.assertEqual(output, "3")
+
+    def test_globals(self):
+        """RunPyJob should use provided globals for evaluation"""
+        job = RunPyJob("eval", ["x + y"], globals={"x": 1, "y": 2})
+        output = job.output().read().decode("utf-8").strip()
+        self.assertEqual(output, "3")
+
+    def test_exec(self):
+        """RunPyJob should execute a Python script with side effects"""
+        cmd = """
+            with open("output.txt", "w") as f:
+                f.write("Hello, World!")
+        """.strip()
+        job = RunPyJob("exec", [cmd])
+        job_output = job.output().read().decode("utf-8").strip()
+        file_output = open("output.txt").read().strip()
+        self.assertEqual(job_output, "")  # No output expected
+        self.assertEqual(file_output, "Hello, World!")
+        os.remove("output.txt")