1 files changed, 95 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/rmharness/tests/test_rmharness_plugin.py b/debug/risc-v-gsp-rm-vm/rmharness/tests/test_rmharness_plugin.py
new file mode 100644
index 0000000..8d74f6e
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/rmharness/tests/test_rmharness_plugin.py
@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Unit-style check for plugins/rmharness_plugin.py (stdlib only).
+
+Run from the risc-v-gsp-rm-vm root:
+
+    python3 rmharness/tests/test_rmharness_plugin.py
+
+It loads the plugin with a fake ``ge`` object and verifies:
+  * register() adds the rmtool/rmtop/rmq commands;
+  * write arguments are refused and recorded as rm_harness_skip;
+  * a real ``rmtool --version`` run is recorded as rm_harness_run with a blob.
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import unittest
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_TOOL_ROOT = os.path.dirname(os.path.dirname(_HERE))
+_PLUGIN = os.path.join(_TOOL_ROOT, "plugins", "rmharness_plugin.py")
+
+
+class FakeRecorder:
+    def __init__(self, directory=None):
+        self.dir = directory
+        self.events = []
+        self.blobs = []
+
+    def event(self, event_type, **fields):
+        self.events.append(dict(type=event_type, **fields))
+
+    def blob(self, name, data):
+        self.blobs.append((name, data))
+        return name
+
+
+class FakeGe:
+    def __init__(self, directory=None):
+        self.recorder = FakeRecorder(directory)
+        self.commands = {}
+        self.logs = []
+
+    def log(self, message, **fields):
+        self.logs.append(message)
+
+    def add_command(self, name, fn, help=""):
+        self.commands[name] = (fn, help)
+
+
+def load_plugin():
+    spec = importlib.util.spec_from_file_location("rmharness_plugin_test",
+                                                  _PLUGIN)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class RmHarnessPluginTest(unittest.TestCase):
+    def setUp(self):
+        self.plugin = load_plugin()
+        self.ge = FakeGe()
+        self.plugin.register(self.ge)
+
+    def test_registers_commands(self):
+        self.assertEqual(sorted(self.ge.commands),
+                         ["rmq", "rmtool", "rmtop"])
+        self.assertTrue(any("rmharness" in msg for msg in self.ge.logs))
+
+    def test_blocked_write_is_refused(self):
+        fn, _help = self.ge.commands["rmtool"]
+        rc = fn(None, ["mem", "alloc", "-d", "0", "--size=1G"])
+        self.assertEqual(rc, 2)
+        types = [e["type"] for e in self.ge.recorder.events]
+        self.assertIn("rm_harness_skip", types)
+        self.assertNotIn("rm_harness_run", types)
+
+    def test_readonly_run_is_recorded_with_blob(self):
+        if not os.path.isfile(
+                os.path.join(_TOOL_ROOT, "rmharness", "rmtool.sh")):
+            self.skipTest("rmharness wrapper not present")
+        fn, _help = self.ge.commands["rmtool"]
+        rc = fn(None, ["--version"])
+        self.assertEqual(rc, 0)
+        types = [e["type"] for e in self.ge.recorder.events]
+        self.assertIn("rm_harness_run", types)
+        self.assertTrue(self.ge.recorder.blobs)
+        run = [e for e in self.ge.recorder.events
+               if e["type"] == "rm_harness_run"][0]
+        self.assertTrue(run["readonly"])
+        self.assertEqual(run["tool"], "rmtool")
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)