"""Tests for monitor/graph_json.py. Verifies: - or1-monitor.AC3.1 (partial): Graph structure for rendering - or1-monitor.AC3.7: PE state serialization - or1-monitor.AC3.8: SM state serialization """ from cm_inst import ArithOp, MemOp, Port, FrameDest, Instruction, OutputStyle from asm.ir import IRGraph, IRNode, IREdge, SourceLoc, SystemConfig from emu.events import TokenReceived, Matched, Executed, Emitted from monitor.graph_json import graph_to_monitor_json, graph_loaded_json from monitor.snapshot import StateSnapshot, PESnapshot, SMSnapshot, SMCellSnapshot from sm_mod import Presence from tokens import DyadToken class TestGraphJsonStructure: """Test basic JSON structure output.""" def test_graph_loaded_json_structure(self): """Test graph_loaded_json returns correct structure.""" # Create minimal IR graph node = IRNode( name="&test", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, loc=SourceLoc(1, 1), ) ir_graph = IRGraph( nodes={"&test": node}, edges=[], system=SystemConfig(pe_count=1, sm_count=0), ) # Create Instruction for IRAM inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) # Create minimal snapshot pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) # Check structure assert result["type"] == "graph_loaded" assert "graph" in result assert "nodes" in result["graph"] assert "edges" in result["graph"] assert "state" in result assert "pes" in result["state"] assert "sms" in result["state"] assert result["sim_time"] == 0.0 assert result["finished"] is False def test_graph_loaded_json_node_fields(self): """Test node fields in graph_loaded_json.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=5, act_id=1, const=42, loc=SourceLoc(1, 1), ) ir_graph = IRGraph( nodes={"&add": node}, edges=[], system=SystemConfig(pe_count=1, sm_count=0), ) inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={5: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) nodes = result["graph"]["nodes"] assert len(nodes) == 1 node_json = nodes[0] assert node_json["id"] == "&add" assert node_json["opcode"] == "add" assert node_json["pe"] == 0 assert node_json["iram_offset"] == 5 assert node_json["act_id"] == 1 assert node_json["const"] == 42 # Check execution overlay fields are present and False assert node_json["active"] is False assert node_json["matched"] is False assert node_json["executed"] is False def test_graph_loaded_json_pe_state(self): """Test PE state serialization.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=5, act_id=1, ) ir_graph = IRGraph(nodes={"&add": node}) inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={5: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) pe_state = result["state"]["pes"]["0"] assert "iram" in pe_state assert "frames" in pe_state assert "tag_store" in pe_state assert "free_frames" in pe_state assert pe_state["input_queue_size"] == 0 def test_graph_loaded_json_sm_state(self): """Test SM state serialization.""" ir_graph = IRGraph() cell_snap = SMCellSnapshot( pres=Presence.FULL, data_l=42, data_r=None, ) sm_snap = SMSnapshot( sm_id=0, cells={10: cell_snap}, deferred_read=None, t0_store=(), input_queue=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={}, sms={0: sm_snap}, ) result = graph_loaded_json(ir_graph, snapshot) sm_state = result["state"]["sms"]["0"] assert "cells" in sm_state assert "10" in sm_state["cells"] assert sm_state["cells"]["10"]["presence"] == "FULL" assert sm_state["cells"]["10"]["data_l"] == 42 assert sm_state["deferred_read"] is None assert sm_state["t0_store_size"] == 0 class TestGraphJsonWithEvents: """Test execution overlay with events.""" def test_monitor_json_with_token_received(self): """Test TokenReceived sets active flag.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node}) token = DyadToken(target=0, offset=0, act_id=0, data=42, port=Port.L) event = TokenReceived(time=1.0, component="pe:0", token=token) inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={0: pe_snap}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) assert result["type"] == "monitor_update" # Find the node in the result nodes = result["graph"]["nodes"] node_json = next((n for n in nodes if n["id"] == "&add"), None) assert node_json is not None assert node_json["active"] is True def test_monitor_json_with_matched(self): """Test Matched event sets matched flag.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=5, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node}) event = Matched(time=1.0, component="pe:0", left=42, right=7, act_id=0, offset=5, frame_id=0) inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={5: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={0: pe_snap}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) nodes = result["graph"]["nodes"] node_json = next((n for n in nodes if n["id"] == "&add"), None) assert node_json is not None assert node_json["matched"] is True def test_monitor_json_with_executed(self): """Test Executed event sets executed flag.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node}) event = Executed(time=1.0, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False) inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={0: pe_snap}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) nodes = result["graph"]["nodes"] node_json = next((n for n in nodes if n["id"] == "&add"), None) assert node_json is not None assert node_json["executed"] is True def test_monitor_json_events_serialization(self): """Test events are properly serialized.""" ir_graph = IRGraph() token = DyadToken(target=0, offset=0, act_id=0, data=42, port=Port.L) events = [ TokenReceived(time=1.0, component="pe:0", token=token), Executed(time=1.1, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False), ] snapshot = StateSnapshot( sim_time=1.1, next_time=2.0, pes={}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, events) assert "events" in result assert len(result["events"]) == 2 event_json_0 = result["events"][0] assert event_json_0["type"] == "TokenReceived" assert event_json_0["time"] == 1.0 assert event_json_0["component"] == "pe:0" assert "details" in event_json_0 event_json_1 = result["events"][1] assert event_json_1["type"] == "Executed" assert event_json_1["time"] == 1.1 assert event_json_1["component"] == "pe:0" class TestGraphJsonEdges: """Test edge serialization.""" def test_edge_serialization(self): """Test edges are properly serialized.""" source = IRNode( name="&a", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) dest = IRNode( name="&b", opcode=ArithOp.SUB, pe=0, iram_offset=1, act_id=0, ) edge = IREdge( source="&a", dest="&b", port=Port.L, source_port=Port.L, ) ir_graph = IRGraph( nodes={"&a": source, "&b": dest}, edges=[edge], ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) edges = result["graph"]["edges"] assert len(edges) == 1 edge_json = edges[0] assert edge_json["source"] == "&a" assert edge_json["target"] == "&b" assert edge_json["port"] == "L" # source_port can be None or "L" depending on how it's set assert edge_json["source_port"] in [None, "L"] assert edge_json["token_flow"] is False def test_edge_token_flow(self): """Test token_flow flag is set on emitted edges.""" source = IRNode( name="&a", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) dest = IRNode( name="&b", opcode=ArithOp.SUB, pe=1, # Destination on a different PE iram_offset=1, act_id=0, ) edge = IREdge(source="&a", dest="&b", port=Port.L) ir_graph = IRGraph( nodes={"&a": source, "&b": dest}, edges=[edge], ) # Emitted event with token targeting PE 1 should mark the edge token = DyadToken(target=1, offset=1, act_id=0, data=42, port=Port.L) event = Emitted(time=1.0, component="pe:0", token=token) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) edges = result["graph"]["edges"] assert len(edges) == 1 edge_json = edges[0] # Edge token_flow should be True when an Emitted event targets the destination PE assert edge_json["token_flow"] is True class TestGraphJsonFinished: """Test finished flag.""" def test_finished_false_when_next_time_not_inf(self): """Test finished=False when next_time is finite.""" ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) assert result["finished"] is False def test_finished_true_when_next_time_inf(self): """Test finished=True when next_time is infinity.""" ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=10.0, next_time=float('inf'), pes={}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) assert result["finished"] is True class TestTagStoreSerialisation: """Test non-empty tag_store JSON serialization in PE state.""" def test_non_empty_tag_store_serialization(self): """Test that tag_store with non-empty mapping serializes correctly.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node}) inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) # Create PESnapshot with non-empty tag_store: act_id 0 maps to frame 2, lane 1 pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={0: (2, 1)}, # act_id=0 -> (frame_id=2, lane=1) presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) pe_state = result["state"]["pes"]["0"] # Verify tag_store is correctly serialized assert "tag_store" in pe_state assert "0" in pe_state["tag_store"] # act_id 0 should be a string key "0" assert pe_state["tag_store"]["0"]["frame_id"] == 2 assert pe_state["tag_store"]["0"]["lane"] == 1 assert pe_state["lane_count"] == 4 def test_multiple_entries_tag_store_serialization(self): """Test tag_store with multiple act_id entries serializes correctly.""" ir_graph = IRGraph() inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) # Create PESnapshot with multiple tag_store entries pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={ 0: (2, 1), 1: (3, 0), 5: (7, 2), }, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, ) result = graph_loaded_json(ir_graph, snapshot) pe_state = result["state"]["pes"]["0"] # Verify all entries are correctly serialized assert "tag_store" in pe_state assert pe_state["tag_store"]["0"]["frame_id"] == 2 assert pe_state["tag_store"]["0"]["lane"] == 1 assert pe_state["tag_store"]["1"]["frame_id"] == 3 assert pe_state["tag_store"]["1"]["lane"] == 0 assert pe_state["tag_store"]["5"]["frame_id"] == 7 assert pe_state["tag_store"]["5"]["lane"] == 2 assert pe_state["lane_count"] == 4 class TestAC72_NewEventTypesSerialization: """AC7.2: Verify new frame-based event types serialize correctly to JSON.""" def test_frame_allocated_event_serialization(self): """FrameAllocated event should serialize with correct type and fields.""" from emu.events import FrameAllocated event = FrameAllocated( time=5.0, component="pe:0", act_id=1, frame_id=2, lane=0, ) ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=5.0, next_time=6.0, pes={}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) # Extract the events from result assert "events" in result events_list = result["events"] assert len(events_list) == 1 event_json = events_list[0] assert event_json["type"] == "FrameAllocated" assert event_json["time"] == 5.0 assert event_json["component"] == "pe:0" assert "details" in event_json assert event_json["details"]["act_id"] == 1 assert event_json["details"]["frame_id"] == 2 assert event_json["details"]["lane"] == 0 def test_frame_freed_event_serialization(self): """FrameFreed event should serialize with correct type and fields.""" from emu.events import FrameFreed event = FrameFreed( time=10.0, component="pe:1", act_id=3, frame_id=4, lane=0, frame_freed=True, ) ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=10.0, next_time=11.0, pes={}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) events_list = result["events"] assert len(events_list) == 1 event_json = events_list[0] assert event_json["type"] == "FrameFreed" assert event_json["time"] == 10.0 assert event_json["component"] == "pe:1" assert "details" in event_json assert event_json["details"]["act_id"] == 3 assert event_json["details"]["frame_id"] == 4 assert event_json["details"]["lane"] == 0 assert event_json["details"]["frame_freed"] == True def test_frame_slot_written_event_serialization(self): """FrameSlotWritten event should serialize with correct type and fields.""" from emu.events import FrameSlotWritten event = FrameSlotWritten( time=7.0, component="pe:0", frame_id=1, slot=5, value=0x1234, ) ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=7.0, next_time=8.0, pes={}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) events_list = result["events"] assert len(events_list) == 1 event_json = events_list[0] assert event_json["type"] == "FrameSlotWritten" assert event_json["time"] == 7.0 assert event_json["component"] == "pe:0" assert "details" in event_json assert event_json["details"]["frame_id"] == 1 assert event_json["details"]["slot"] == 5 assert event_json["details"]["value"] == 0x1234 def test_token_rejected_event_serialization(self): """TokenRejected event should serialize with correct type and fields.""" from emu.events import TokenRejected from tokens import DyadToken token = DyadToken( target=0, offset=1, act_id=99, data=0x5678, port=Port.L, ) event = TokenRejected( time=3.0, component="pe:0", token=token, reason="invalid act_id", ) ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=3.0, next_time=4.0, pes={}, sms={}, ) result = graph_to_monitor_json(ir_graph, snapshot, [event]) events_list = result["events"] assert len(events_list) == 1 event_json = events_list[0] assert event_json["type"] == "TokenRejected" assert event_json["time"] == 3.0 assert event_json["component"] == "pe:0" assert "details" in event_json assert event_json["details"]["reason"] == "invalid act_id"