"""Tests for the risc-v-gsp-rm-debugger MCP server plugin (roadmap R3). The protocol and tool tests drive an in-memory Session over a stub RSP-shaped client, so they need no QEMU and no simulator. A standalone subprocess test launches the native backend end-to-end and skips when that backend is absent. """ from __future__ import annotations import io import json import os import subprocess import sys import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) 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 BASE = 0x80000000 class StubClient: """RSP-shaped in-memory client: enough for Session without an emulator.""" def __init__(self, size: int = 0x20000): self.base = BASE self.mem = bytearray(size) self.regs = {"pc": BASE, "ra": 0, "sp": BASE + size, "a0": 0, "a1": 0} self.sw_bps: set[int] = set() self.hw_bps: set[int] = set() self.watches: dict[tuple[int, int, str], None] = {} self.closed = False self.features = "swbreak+;hwbreak+" def _span(self, addr: int, size: int) -> int: start = addr - self.base if start < 0 or start + size > len(self.mem): raise ValueError(f"out of range: {addr:#x}+{size}") return start def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes: start = self._span(addr, size) return bytes(self.mem[start : start + size]) def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None: start = self._span(addr, len(data)) self.mem[start : start + len(data)] = data def read_registers(self) -> dict[str, int]: return dict(self.regs) def read_reg(self, name: str) -> int: return int(self.regs.get(str(name), 0)) def write_reg(self, name: str, value: int) -> None: self.regs[str(name)] = int(value) @property def pc(self) -> int: return self.regs["pc"] def set_sw_break(self, addr: int, kind: int = 4) -> None: self.sw_bps.add(addr) def clear_sw_break(self, addr: int, kind: int = 4) -> None: self.sw_bps.discard(addr) def set_hw_break(self, addr: int, kind: int = 4) -> None: self.hw_bps.add(addr) def clear_hw_break(self, addr: int, kind: int = 4) -> None: self.hw_bps.discard(addr) def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: self.watches[(addr, size, kind)] = None def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: self.watches.pop((addr, size, kind), None) def continue_(self, timeout: float | None = None) -> StopReply: return StopReply(raw="T05", signal=5) def step(self, timeout: float | None = None) -> StopReply: return StopReply(raw="T05", signal=5) def interrupt(self, timeout: float | None = None) -> StopReply: return StopReply(raw="T02", signal=2) def close(self) -> None: self.closed = True class McpTestCase(unittest.TestCase): def setUp(self): self.client = StubClient() symbols = SymbolDB() symbols.add(Symbol(addr=BASE + 0x100, name="rpc_ping", size=0x20, source="test")) symbols.add(Symbol(addr=BASE + 0x200, name="gsp_ctrl", size=0x40, source="test")) self.session = Session(client=self.client, symbols=symbols, record=False, name="mcp-test") self.addCleanup(self.session.close) def make_server(self, **kw): self.inp = io.StringIO() self.out = io.StringIO() self.log = io.StringIO() options = dict(input_stream=self.inp, output_stream=self.out, log_stream=self.log) options.update(kw) return mcp.GreenEyeMCPServer(self.session, **options) def request(self, server, method, params=None, request_id=1): message = {"jsonrpc": "2.0", "id": request_id, "method": method} if params is not None: message["params"] = params return server.handle_message(message) def call_tool(self, server, name, **arguments): response = self.request(server, "tools/call", {"name": name, "arguments": arguments}, request_id=9) self.assertIn("result", response, response) result = response["result"] payload = json.loads(result["content"][0]["text"]) return payload, result.get("isError", False) class ProtocolTests(McpTestCase): def test_initialize(self): response = self.request( self.make_server(), "initialize", {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}, ) result = response["result"] self.assertEqual(response["jsonrpc"], "2.0") self.assertEqual(result["protocolVersion"], "2024-11-05") self.assertEqual(result["serverInfo"]["name"], "greeneye-mcp") self.assertIn("tools", result["capabilities"]) def test_initialize_negotiation_and_fallback(self): server = self.make_server() self.assertEqual( self.request(server, "initialize", {"protocolVersion": "2025-03-26"})["result"]["protocolVersion"], "2025-03-26", ) self.assertEqual( self.request(server, "initialize", {"protocolVersion": "1999-01-01"})["result"]["protocolVersion"], mcp.DEFAULT_PROTOCOL, ) def test_initialized_notification_has_no_response(self): server = self.make_server() self.assertIsNone(server.handle_message({"jsonrpc": "2.0", "method": "notifications/initialized"})) def test_ping(self): self.assertEqual(self.request(self.make_server(), "ping")["result"], {}) def test_tools_list_catalogue(self): tools = self.request(self.make_server(), "tools/list")["result"]["tools"] names = [tool["name"] for tool in tools] 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"]) def test_unknown_method(self): response = self.request(self.make_server(), "no/such") self.assertEqual(response["error"]["code"], mcp.METHOD_NOT_FOUND) self.assertEqual(response["id"], 1) def test_unknown_tool(self): response = self.request(self.make_server(), "tools/call", {"name": "nope", "arguments": {}}) self.assertEqual(response["error"]["code"], mcp.INVALID_PARAMS) def test_tools_call_before_initialize_is_allowed(self): payload, is_error = self.call_tool(self.make_server(), "session_info") self.assertFalse(is_error) self.assertEqual(payload["name"], "mcp-test") def test_parse_error_is_reported(self): server = self.make_server() server.process_line("{not json") line = self.out.getvalue().strip() self.assertEqual(json.loads(line)["error"]["code"], mcp.PARSE_ERROR) def test_serve_stream_end_to_end_and_no_stdout_pollution(self): server = self.make_server() lines = [ json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}), json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}), json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}), json.dumps( {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "session_info", "arguments": {}}} ), ] server.serve_stream(io.StringIO("\n".join(lines) + "\n"), self.out) raw = self.out.getvalue() responses = [json.loads(line) for line in raw.splitlines()] self.assertEqual([response["id"] for response in responses], [1, 2, 3]) self.assertTrue(all(response["jsonrpc"] == "2.0" for response in responses)) self.assertNotIn("[greeneye-mcp]", raw) self.assertNotIn("[greeneye-mcp]", self.inp.getvalue()) class ReadOnlyToolTests(McpTestCase): def test_symbols_resolve(self): payload, is_error = self.call_tool(self.make_server(), "symbols_resolve", target="rpc_ping") self.assertFalse(is_error) self.assertEqual(payload["address"], BASE + 0x100) self.assertEqual(payload["symbol"], "rpc_ping") def test_read_mem_and_hexdump(self): self.client.mem[0x10:0x14] = b"\xde\xad\xbe\xef" server = self.make_server() payload, is_error = self.call_tool(server, "read_mem", addr=BASE + 0x10, size=4) self.assertFalse(is_error) self.assertEqual(payload["hex"], "deadbeef") payload, is_error = self.call_tool(server, "hexdump", addr=BASE + 0x10, size=4) self.assertFalse(is_error) self.assertIn("de ad be ef", payload["text"]) def test_bp_set_and_clear(self): server = self.make_server() payload, is_error = self.call_tool(server, "bp_set", target="rpc_ping") self.assertFalse(is_error) self.assertIn(BASE + 0x100, self.client.sw_bps) payload, is_error = self.call_tool(server, "bp_clear", target="rpc_ping") self.assertFalse(is_error) self.assertEqual(payload["cleared"], 1) self.assertNotIn(BASE + 0x100, self.client.sw_bps) def test_watch_set(self): payload, is_error = self.call_tool(self.make_server(), "watch_set", target="gsp_ctrl", size=8, kind="w") self.assertFalse(is_error) self.assertEqual(payload["addr"], BASE + 0x200) self.assertIn((BASE + 0x200, 8, "w"), self.client.watches) def test_snapshot_digest(self): payload, is_error = self.call_tool(self.make_server(), "snapshot", addr=BASE, size=16, label="unit") self.assertFalse(is_error) self.assertEqual(payload["size"], 16) self.assertEqual(len(payload["sha256"]), 64) def test_session_info_capabilities(self): payload, is_error = self.call_tool( self.make_server(allow_write=True, allow_inject=True), "session_info" ) self.assertFalse(is_error) self.assertEqual(payload["name"], "mcp-test") self.assertTrue(payload["capabilities"]["write"]) self.assertTrue(payload["capabilities"]["inject"]) self.assertFalse(payload["capabilities"]["trace"]) def test_plugins_list(self): payload, is_error = self.call_tool(self.make_server(), "plugins_list") self.assertFalse(is_error) self.assertIn("mcp_server.py", payload["bundled"]) def test_analyze_current_reads_active_recording(self): with tempfile.TemporaryDirectory(prefix="mcp-rec-") as directory: session = Session(client=StubClient(), symbols=SymbolDB(), record=directory, name="active") try: server = mcp.GreenEyeMCPServer( session, input_stream=io.StringIO(), output_stream=io.StringIO(), log_stream=io.StringIO(), ) payload, is_error = self.call_tool(server, "analyze_current") self.assertFalse(is_error) self.assertIn("# risc-v-gsp-rm-debugger session summary", payload["report"]) finally: 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() payload, is_error = self.call_tool(server, "write_mem", addr=BASE + 0x10, hex="0102") self.assertTrue(is_error) self.assertEqual(payload["error"], "write_disabled") self.assertEqual(bytes(self.client.mem[0x10:0x12]), b"\x00\x00") def test_write_mem_enabled(self): server = self.make_server(allow_write=True) payload, is_error = self.call_tool(server, "write_mem", addr=BASE + 0x10, hex="0x01 02") self.assertFalse(is_error) self.assertEqual(payload["written"], 2) self.assertEqual(bytes(self.client.mem[0x10:0x12]), b"\x01\x02") def test_call_function_refused_without_allow_write(self): payload, is_error = self.call_tool(self.make_server(), "call_function", target="rpc_ping", args=[1]) self.assertTrue(is_error) self.assertEqual(payload["error"], "write_disabled") def test_inject_refused_without_allow_inject(self): payload, is_error = self.call_tool(self.make_server(), "inject_service", service="/nonexistent.elf") self.assertTrue(is_error) self.assertEqual(payload["error"], "inject_disabled") def test_inject_missing_file_after_gate(self): payload, is_error = self.call_tool( self.make_server(allow_inject=True), "inject_service", service="/nonexistent.elf" ) self.assertTrue(is_error) self.assertEqual(payload["error"], "invalid_arguments") class AvailabilityTests(McpTestCase): def test_analyze_current_without_recording(self): payload, is_error = self.call_tool(self.make_server(), "analyze_current") self.assertTrue(is_error) self.assertEqual(payload["error"], "not_available") def test_trace_start_on_non_native_backend(self): payload, is_error = self.call_tool(self.make_server(), "trace_start") self.assertTrue(is_error) self.assertEqual(payload["error"], "not_available") def test_trace_stop_without_active_trace(self): payload, is_error = self.call_tool(self.make_server(), "trace_stop") self.assertTrue(is_error) self.assertEqual(payload["error"], "invalid_arguments") def test_mutation_missing_script_is_structured(self): server = self.make_server() original = mcp.MUTATION_SCRIPT mcp.MUTATION_SCRIPT = "/nonexistent/mutation_harness.py" try: payload, is_error = self.call_tool(server, "mutation_run") finally: mcp.MUTATION_SCRIPT = original self.assertTrue(is_error) self.assertEqual(payload["error"], "not_available") def test_portability_missing_script_is_structured(self): server = self.make_server() original = mcp.MATRIX_SCRIPT mcp.MATRIX_SCRIPT = "/nonexistent/portability_matrix.py" try: payload, is_error = self.call_tool(server, "portability_matrix_run", targets=["/x.elf"]) finally: mcp.MATRIX_SCRIPT = original self.assertTrue(is_error) self.assertEqual(payload["error"], "not_available") def test_probe_missing_elf(self): payload, is_error = self.call_tool(self.make_server(), "probe_firmware", elf="/nonexistent/fw.elf") self.assertTrue(is_error) self.assertEqual(payload["error"], "invalid_arguments") class CloseAndTransportTests(McpTestCase): def test_close_stops_serving_but_leaves_plugin_session(self): server = self.make_server(owns_session=False) payload, is_error = self.call_tool(server, "close") self.assertFalse(is_error) self.assertFalse(payload["session_closed"]) self.assertTrue(server._stop) self.assertFalse(self.client.closed) def test_close_closes_owned_session(self): server = self.make_server(owns_session=True) payload, is_error = self.call_tool(server, "close") self.assertFalse(is_error) self.assertTrue(payload["session_closed"]) self.assertTrue(self.client.closed) def test_parse_tcp_is_loopback_only(self): self.assertEqual(mcp.parse_tcp("127.0.0.1:8765"), ("127.0.0.1", 8765)) self.assertEqual(mcp.parse_tcp("localhost:0")[0], "localhost") with self.assertRaises(Exception): mcp.parse_tcp("0.0.0.0:8765") class StandaloneEntryTests(unittest.TestCase): @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing") def test_help(self): result = subprocess.run( [sys.executable, MCP_SERVER, "--help"], capture_output=True, text=True, cwd=TOOL_ROOT ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("--stdio", result.stdout) self.assertIn("--allow-write", result.stdout) @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing") def test_stdio_native_session(self): try: import greeneye.sim_backend # noqa: F401 except Exception as exc: self.skipTest(f"native backend unavailable: {exc}") with tempfile.TemporaryDirectory(prefix="mcp-native-") 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": "session_info", "arguments": {}}}, {"jsonrpc": "2.0", "id": 3, "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"], input=payload, capture_output=True, text=True, cwd=TOOL_ROOT, env=env, timeout=180, ) self.assertEqual(result.returncode, 0, result.stderr[-2000:]) responses = [json.loads(line) for line in result.stdout.splitlines() if line.strip()] self.assertEqual(len(responses), 3) by_id = {response.get("id"): response for response in responses} info = json.loads(by_id[2]["result"]["content"][0]["text"]) self.assertEqual(info["backend"], "native") 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()