1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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)
|
