diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_mcp.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_mcp.py | 422 |
1 files changed, 421 insertions, 1 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py index 2df8a4a..87a85a8 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py @@ -19,11 +19,14 @@ TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) PLUGINS_DIR = os.path.join(TOOL_ROOT, "plugins") MCP_SERVER = os.path.join(PLUGINS_DIR, "mcp_server.py") DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +SERVICE_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf") for _path in (TOOL_ROOT, PLUGINS_DIR): if _path not in sys.path: sys.path.insert(0, _path) import mcp_server as mcp # noqa: E402 +import greeneye as ge # noqa: E402 +from greeneye.inject import SERVICE_REGISTRY_ATTR, InjectedService # noqa: E402 from greeneye.rsp import StopReply # noqa: E402 from greeneye.session import Session # noqa: E402 from greeneye.symbols import Symbol, SymbolDB # noqa: E402 @@ -167,8 +170,19 @@ class ProtocolTests(McpTestCase): def test_tools_list_catalogue(self): tools = self.request(self.make_server(), "tools/list")["result"]["tools"] names = [tool["name"] for tool in tools] - self.assertEqual(len(names), 24) + self.assertGreaterEqual(len(names), 24) + self.assertEqual(len(names), len(mcp.TOOLS)) self.assertEqual(set(names), mcp._TOOL_NAMES) + for expected in ( + "backends_list", + "session_launch", + "service_list", + "sim_cycles", + "sim_device_state", + "sim_profile", + "sim_regions", + ): + self.assertIn(expected, names) for tool in tools: self.assertEqual(tool["inputSchema"]["type"], "object") self.assertTrue(tool["description"]) @@ -283,6 +297,328 @@ class ReadOnlyToolTests(McpTestCase): session.close() +class StubCycleModel: + """Minimal CycleModel shape for MCP sim-tool tests (no sim/ import needed).""" + + name = "rv64imac-modeled/1" + + def describe(self) -> dict: + return {"name": self.name, "alu": 1, "memory_latency": 3, "fidelity": "modeled"} + + +class StubMachine: + """Machine-shaped object exposing the public surface the sim tools read.""" + + def __init__(self): + self.cycles = 120 + self.insns = 100 + self.cycle_model = StubCycleModel() + self.unknown_mmio = "lenient" + self.bias = 0x80000000 + self.ram_base = BASE + self.ram_size = 0x20000 + self.mmio_log = [{"pc": BASE + 4, "addr": 0x2000000000000A00, "size": 8, "rw": "r"}] + self.mmio_unknown = 1 + self.device_map = None + + +class StubTarget: + """SimTarget-shaped handle so the server classifies the session as native.""" + + keep_alive = True + + def __init__(self, machine): + self.machine = machine + self.mmio_log = machine.mmio_log + self.mmio_unknown = machine.mmio_unknown + self.ram_base = BASE + self.memory_mb = 256 + + def kill(self): + pass + + +class NativeSimToolTests(McpTestCase): + def setUp(self): + super().setUp() + self.client.trace = None # backend detection: native + self.machine = StubMachine() + self.session.target = StubTarget(self.machine) + + def test_backends_list(self): + payload, is_error = self.call_tool(self.make_server(owns_session=True), "backends_list") + self.assertFalse(is_error) + self.assertEqual(payload["active"], "native") + self.assertTrue(payload["launchable"]) + names = {entry["name"] for entry in payload["backends"]} + self.assertEqual(names, {"native", "qemu"}) + + def test_sim_cycles(self): + payload, is_error = self.call_tool(self.make_server(), "sim_cycles") + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["insns"], 100) + self.assertEqual(payload["cycles"], 120) + self.assertEqual(payload["cycles_per_instruction"], 1.2) + self.assertEqual(payload["cycle_model"]["name"], "rv64imac-modeled/1") + self.assertIn("modeled", payload["fidelity"]) + + def test_sim_device_state_reports_absent_device_model(self): + payload, is_error = self.call_tool(self.make_server(), "sim_device_state", limit=4) + self.assertFalse(is_error) + self.assertEqual(payload["mmio_unknown"], 1) + self.assertEqual(len(payload["mmio_log"]), 1) + self.assertEqual(payload["unknown_mmio"], "lenient") + self.assertIn("device_model", payload) + if "error" in payload["device_model"]: + self.assertEqual(payload["device_model"]["error"], "not_available") + + def test_sim_device_state_serialises_attached_registry(self): + class Registry: + def state(self): + return {"schema": "greeneye-sim-devices/1", "devices": []} + + self.machine.devices = Registry() + payload, is_error = self.call_tool(self.make_server(), "sim_device_state") + self.assertFalse(is_error) + self.assertEqual(payload["device_model"]["source"], "machine.devices.state") + self.assertEqual(payload["device_model"]["state"]["schema"], "greeneye-sim-devices/1") + json.dumps(payload) # the tool result must stay JSON-serialisable + + def test_sim_profile(self): + payload, is_error = self.call_tool(self.make_server(), "sim_profile") + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["bias"], 0x80000000) + self.assertEqual(payload["cycles"], 120) + self.assertIn("session_meta", payload) + + def test_sim_regions(self): + payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=4) + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["bias"], 0x80000000) + self.assertEqual(payload["unknown_mmio"], "lenient") + regions = {region["name"]: region for region in payload["regions"]} + self.assertIn("ram", regions) + self.assertIn("uart", regions) + self.assertEqual(regions["ram"]["base"], BASE) + self.assertEqual(regions["ram"]["size"], 0x20000) + self.assertEqual(regions["ram"]["end"], BASE + 0x20000) + self.assertEqual(regions["uart"]["base"], 0x10000000) + self.assertEqual(regions["uart"]["size"], 0x100) + self.assertIsNone(payload["device_map_entries"]) + self.assertEqual(payload["device_map"], []) + self.assertEqual(len(payload["observed_mmio_pages"]), 1) + page = payload["observed_mmio_pages"][0] + self.assertEqual(page["page"], 0x2000000000000A00 & ~0xFFF) + self.assertEqual(page["reads"], 1) + self.assertEqual(page["writes"], 0) + self.assertEqual(page["addresses"], [0x2000000000000A00]) + self.assertEqual(page["pcs"], [BASE + 4]) + + def test_sim_regions_device_map_entries(self): + class Entry: + address = 0x2000000000000A00 + width = 8 + kind = "r" + hits = 3 + first_epc_fw = 0x4009014 + source = "probe" + + class Map: + def entries(self): + return [Entry()] + + self.machine.device_map = Map() + payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=8) + self.assertFalse(is_error) + self.assertEqual(payload["device_map_entries"], 1) + self.assertEqual(payload["device_map"][0]["address"], 0x2000000000000A00) + self.assertEqual(payload["device_map"][0]["first_epc_fw"], 0x4009014) + self.assertEqual(payload["device_map"][0]["source"], "probe") + + def test_sim_tools_refuse_non_native(self): + # a fresh stub client without .trace is not a native session + client = StubClient() + session = Session(client=client, symbols=self.session.symbols, record=False, name="stub-qemu") + server = mcp.GreenEyeMCPServer( + session, input_stream=io.StringIO(), output_stream=io.StringIO(), log_stream=io.StringIO() + ) + try: + for name in ("sim_cycles", "sim_device_state", "sim_profile", "sim_regions"): + payload, is_error = self.call_tool(server, name) + self.assertTrue(is_error, name) + self.assertEqual(payload["error"], "not_available", name) + finally: + session.close() + + +class ServiceListTests(McpTestCase): + def test_service_list_empty_with_slots(self): + self.session.memory.write_u64(BASE + 0x200 + 0x150 + 6 * 8, 0x81000000) + payload, is_error = self.call_tool(self.make_server(), "service_list") + self.assertFalse(is_error) + self.assertEqual(payload["count"], 0) + self.assertEqual(payload["mailbox_cmd_base"], 16) + self.assertEqual(payload["slots"][6]["entry"], 0x81000000) + self.assertEqual(payload["slots"][6]["cmd"], 22) + + def test_service_list_reports_registry(self): + service = InjectedService( + path="/tmp/svc.elf", + name="svc.elf", + load_addr=0x81000000, + entry_addr=0x81000100, + dispatch_offset=0x100, + slot=6, + size=64, + ) + setattr(self.session, SERVICE_REGISTRY_ATTR, [service]) + payload, is_error = self.call_tool(self.make_server(), "service_list") + self.assertFalse(is_error) + self.assertEqual(payload["count"], 1) + self.assertEqual(payload["services"][0]["entry_addr"], 0x81000100) + self.assertEqual(payload["services"][0]["slot"], 6) + + +class SessionLaunchTests(McpTestCase): + def test_refused_in_plugin_mode(self): + payload, is_error = self.call_tool(self.make_server(owns_session=False), "session_launch", backend="native") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "not_available") + + def test_invalid_backend(self): + payload, is_error = self.call_tool(self.make_server(owns_session=True), "session_launch", backend="mips") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_missing_kernel(self): + payload, is_error = self.call_tool( + self.make_server(owns_session=True), "session_launch", backend="native", kernel="/nonexistent.elf" + ) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_qemu_trace_argument_refused(self): + payload, is_error = self.call_tool( + self.make_server(owns_session=True), "session_launch", backend="qemu", trace=True + ) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)") + def test_launch_native_replaces_owned_session(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + server = self.make_server(owns_session=True) + # Contract (docs/20 §5/§7): session_launch is session-class and allowed + # without --allow-write; it only replaces the local served session. + self.assertFalse(server.allow_write) + with tempfile.TemporaryDirectory(prefix="mcp-launch-") as directory: + payload, is_error = self.call_tool( + server, "session_launch", backend="native", kernel=DEMO_ELF, record=directory + ) + self.assertFalse(is_error, payload) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["session"]["kernel"], DEMO_ELF) + self.assertTrue(self.client.closed) + cycles, is_error = self.call_tool(server, "sim_cycles") + self.assertFalse(is_error, cycles) + self.assertEqual(cycles["backend"], "native") + self.assertIsNotNone(cycles["cycle_model"]) + regions, is_error = self.call_tool(server, "sim_regions") + self.assertFalse(is_error, regions) + self.assertIn("ram", {region["name"] for region in regions["regions"]}) + server.session.close() + + +@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)") +class NativeSessionToolTests(unittest.TestCase): + """End-to-end tool checks against a real native session (demo firmware).""" + + def setUp(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + self.session = ge.launch(kernel=DEMO_ELF, backend="native", record=None, name="native-mcp") + self.machine = self.session.target.machine + self.entry = self.machine.pc + self.addCleanup(self.session.close) + self.inp = io.StringIO() + self.out = io.StringIO() + self.log = io.StringIO() + + def make_server(self, **kw): + return mcp.GreenEyeMCPServer( + self.session, input_stream=self.inp, output_stream=self.out, log_stream=self.log, **kw + ) + + def call_tool(self, server, name, **arguments): + response = server.handle_message( + {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": name, "arguments": arguments}} + ) + self.assertIn("result", response, response) + result = response["result"] + return json.loads(result["content"][0]["text"]), result.get("isError", False) + + def test_native_step_budget(self): + server = self.make_server() + self.assertEqual(self.machine.insns, 0) + payload, is_error = self.call_tool(server, "step", count=3) + self.assertFalse(is_error, payload) + self.assertEqual(payload["steps"], 3) + self.assertEqual(self.machine.insns, 3) + self.assertNotEqual(self.machine.pc, self.entry) + self.assertEqual(payload["pc"], self.machine.pc) + + def test_native_step_budget_cap(self): + payload, is_error = self.call_tool(self.make_server(), "step", count=10001) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_native_run_budget_and_cycles(self): + server = self.make_server() + budget = 0.02 + before = self.machine.insns + payload, is_error = self.call_tool(server, "run", duration=budget) + self.assertFalse(is_error, payload) + self.assertGreaterEqual(payload["stops"], 1) + insns = self.machine.insns + self.assertGreater(insns, before) + self.assertLessEqual(insns, int(budget * 1_000_000) + 2000) + cycles, is_error = self.call_tool(server, "sim_cycles") + self.assertFalse(is_error, cycles) + self.assertEqual(cycles["backend"], "native") + self.assertEqual(cycles["insns"], insns) + self.assertGreater(cycles["cycles"], 0) + self.assertEqual(cycles["accounting_enabled"], True) + + def test_native_sim_regions(self): + payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=8) + self.assertFalse(is_error, payload) + self.assertEqual(payload["backend"], "native") + regions = {region["name"]: region for region in payload["regions"]} + self.assertEqual(regions["ram"]["base"], 0x80000000) + self.assertEqual(regions["ram"]["size"], 256 * 1024 * 1024) + self.assertEqual(regions["uart"]["base"], 0x10000000) + image_regions = [region for region in payload["regions"] if region["kind"] == "firmware"] + self.assertTrue(image_regions) + self.assertGreaterEqual(image_regions[0]["base"], 0x80000000) + self.assertTrue(image_regions[0]["flags"]["exec"]) + self.assertIsNone(payload["device_map_entries"]) + self.assertEqual(payload["observed_mmio_pages"], []) + + def test_native_session_info_reports_native(self): + payload, is_error = self.call_tool(self.make_server(), "session_info") + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertTrue(payload["capabilities"]["trace"]) + + class GatingTests(McpTestCase): def test_write_mem_refused_without_allow_write(self): server = self.make_server() @@ -426,6 +762,90 @@ class StandaloneEntryTests(unittest.TestCase): self.assertFalse(by_id[3]["result"].get("isError", False)) self.assertTrue(any(name.endswith("-mcp") for name in os.listdir(dumps))) + @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing") + @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)") + def test_stdio_native_sim_tools_injection_and_trace(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + if not os.path.isfile(SERVICE_ELF): + build = subprocess.run( + ["make", "-C", os.path.join(TOOL_ROOT, "sdk")], capture_output=True, text=True + ) + if build.returncode != 0 or not os.path.isfile(SERVICE_ELF): + self.skipTest(f"SDK build failed: {build.stderr[-200:]}") + with tempfile.TemporaryDirectory(prefix="mcp-native-sim-") as dumps: + env = dict(os.environ, GREENEYE_DUMPS=dumps) + requests = [ + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05"}}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "backends_list", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "sim_cycles", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "sim_profile", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "sim_device_state", "arguments": {"limit": 2}}}, + {"jsonrpc": "2.0", "id": 6, "method": "tools/call", "params": {"name": "trace_start", "arguments": {"exec": False, "max_events": 10000}}}, + {"jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "trace_stop", "arguments": {}}}, + { + "jsonrpc": "2.0", + "id": 8, + "method": "tools/call", + "params": { + "name": "inject_service", + "arguments": {"service": SERVICE_ELF, "slot": 6, "cmd": 1, "arg0": 65, "path": "both"}, + }, + }, + {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": "service_list", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 10, "method": "tools/call", "params": {"name": "close", "arguments": {}}}, + ] + payload = "".join(json.dumps(request) + "\n" for request in requests) + result = subprocess.run( + [sys.executable, MCP_SERVER, "--stdio", "--backend", "native", "--record", "--allow-inject"], + input=payload, + capture_output=True, + text=True, + cwd=TOOL_ROOT, + env=env, + timeout=300, + ) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + responses = [json.loads(line) for line in result.stdout.splitlines() if line.strip()] + by_id = {response.get("id"): response for response in responses} + self.assertEqual(sorted(by_id), [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + + def call_result(request_id): + response = by_id[request_id] + self.assertIn("result", response, response) + body = response["result"] + self.assertFalse(body.get("isError", False), body) + return json.loads(body["content"][0]["text"]) + + backends = call_result(2) + self.assertEqual(backends["active"], "native") + cycles = call_result(3) + self.assertEqual(cycles["backend"], "native") + self.assertIsNotNone(cycles["cycle_model"]) + profile = call_result(4) + self.assertEqual(profile["backend"], "native") + self.assertEqual(profile["kernel"], DEMO_ELF) + device_state = call_result(5) + self.assertIn("device_model", device_state) + trace_start = call_result(6) + self.assertTrue(trace_start["tracing"]) + trace_stop = call_result(7) + self.assertFalse(trace_stop["tracing"]) + self.assertTrue(str(trace_stop["blob"]).endswith("trace.jsonl")) + injected = call_result(8) + self.assertEqual(injected["remote"], 0x41 ^ 0xBEEF) + self.assertEqual(injected["firmware"], 0x41 ^ 0xBEEF) + services = call_result(9) + self.assertEqual(services["count"], 1) + self.assertEqual(services["services"][0]["entry_addr"], injected["service"]["entry_addr"]) + slots = services["slots"] + if slots: + self.assertEqual(slots[6]["entry"], injected["service"]["entry_addr"]) + self.assertFalse(by_id[10]["result"].get("isError", False)) + if __name__ == "__main__": unittest.main() |