Something went wrong. Try again.
OR-1 dataflow CPU sketch
Something went wrong. Try again.
19 kB · 584 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585"""Tests for the Lower pass (CST → IRGraph transformation).
Tests verify:- Instruction definition (inst_def) → IRNode with opcode, placement, named args- Plain edges → IREdge with correct source, dest, ports- Strong/weak inline edges → anonymous nodes + wiring- Data definitions → IRDataDef with SM ID, cell address, packed values- System pragma → SystemConfig with pe_count, sm_count, etc.- Function scoping → qualified names, nested IRRegions- Location directives → LOCATION IRRegions- Error handling → reserved names, duplicate definitions"""
from tests.pipeline import parse_and_lower
from asm.ir import RegionKind, SourceLocfrom asm.errors import ErrorCategoryfrom cm_inst import ArithOp, LogicOp, MemOp, Port, RoutingOp
class TestInstDef: """Tests for instruction definition (AC2.1, AC2.8, AC2.9)."""
def test_basic_instruction(self, parser): """Parse simple instruction definition.""" graph = parse_and_lower(parser, """\ &my_add <| add """)
assert "&my_add" in graph.nodes node = graph.nodes["&my_add"] assert node.opcode == ArithOp.ADD assert node.name == "&my_add"
def test_instruction_with_const(self, parser): """Parse instruction with constant operand.""" graph = parse_and_lower(parser, """\ &my_const <| const, 42 """)
assert "&my_const" in graph.nodes node = graph.nodes["&my_const"] assert node.opcode == RoutingOp.CONST assert node.const == 42
def test_instruction_with_hex_const(self, parser): """Parse instruction with hexadecimal constant.""" graph = parse_and_lower(parser, """\ &mask <| const, 0xFF """)
assert "&mask" in graph.nodes node = graph.nodes["&mask"] assert node.const == 0xFF
def test_instruction_with_pe_placement(self, parser): """Parse instruction with PE placement qualifier (AC2.8).""" graph = parse_and_lower(parser, """\ &my_add|pe0 <| add """)
assert "&my_add" in graph.nodes node = graph.nodes["&my_add"] assert node.pe == 0
def test_instruction_with_pe_placement_nonzero(self, parser): """Parse instruction with non-zero PE placement.""" graph = parse_and_lower(parser, """\ &result|pe2 <| pass """)
assert "&result" in graph.nodes node = graph.nodes["&result"] assert node.pe == 2
def test_instruction_with_named_args(self, parser): """Parse instruction with named arguments (AC2.9).""" graph = parse_and_lower(parser, """\ &serial <| ior, dest=0x45 """)
# ior is not in MNEMONIC_TO_OP, so we should have an error # but the instruction should still be created or we should check errors assert len(graph.errors) > 0
def test_shift_instruction(self, parser): """Parse shift instruction.""" graph = parse_and_lower(parser, """\ &shift_left <| shl """)
assert "&shift_left" in graph.nodes node = graph.nodes["&shift_left"] assert node.opcode == ArithOp.SHL
class TestPlainEdge: """Tests for plain edges (AC2.2)."""
def test_basic_plain_edge(self, parser): """Parse basic plain edge.""" graph = parse_and_lower(parser, """\ &a <| pass &b <| add &a |> &b:L """)
assert len(graph.edges) == 1 edge = graph.edges[0] assert edge.source == "&a" assert edge.dest == "&b" assert edge.port == Port.L
def test_plain_edge_to_right_port(self, parser): """Parse plain edge to right port.""" graph = parse_and_lower(parser, """\ &a <| pass &b <| add &a |> &b:R """)
assert len(graph.edges) == 1 edge = graph.edges[0] assert edge.port == Port.R
def test_plain_edge_fanout(self, parser): """Parse fanout (one source to multiple destinations).""" graph = parse_and_lower(parser, """\ &a <| pass &b <| add &c <| sub &a |> &b:L, &c:R """)
assert len(graph.edges) == 2 assert graph.edges[0].dest == "&b" assert graph.edges[0].port == Port.L assert graph.edges[1].dest == "&c" assert graph.edges[1].port == Port.R
def test_plain_edge_with_source_port(self, parser): """Parse plain edge with source port specification.""" graph = parse_and_lower(parser, """\ &a:L <| pass &b <| add &a:L |> &b:L """)
assert len(graph.edges) == 1 edge = graph.edges[0] assert edge.source_port == Port.L
class TestStrongEdge: """Tests for strong inline edges (AC2.3)."""
def test_basic_strong_edge(self, parser): """Parse basic strong inline edge.""" graph = parse_and_lower(parser, """\ &a <| pass &b <| pass &c <| pass &d <| pass add &a, &b |> &c, &d """)
# Should create anonymous node anon_nodes = [n for n in graph.nodes.keys() if n.startswith("&__anon_")] assert len(anon_nodes) == 1 anon_name = anon_nodes[0]
anon_node = graph.nodes[anon_name] assert anon_node.opcode == ArithOp.ADD
# Should create 4 edges: 2 inputs, 2 outputs assert len(graph.edges) == 4
# Verify input edges input_edges = [e for e in graph.edges if e.dest == anon_name] assert len(input_edges) == 2 left_input = [e for e in input_edges if e.port == Port.L][0] right_input = [e for e in input_edges if e.port == Port.R][0] assert left_input.source == "&a" assert right_input.source == "&b"
# Verify output edges output_edges = [e for e in graph.edges if e.source == anon_name] assert len(output_edges) == 2
def test_strong_edge_anonymous_name_format(self, parser): """Verify anonymous nodes have correct naming.""" graph = parse_and_lower(parser, """\ &a <| pass &b <| pass add &a |> &b """)
anon_nodes = [n for n in graph.nodes.keys() if n.startswith("&__anon_")] assert len(anon_nodes) == 1 assert anon_nodes[0].startswith("&__anon_")
class TestWeakEdge: """Tests for weak inline edges (AC2.4)."""
def test_basic_weak_edge(self, parser): """Parse basic weak inline edge.""" graph = parse_and_lower(parser, """\ &a <| pass &b <| pass &c <| pass &d <| pass &c, &d sub <| &a, &b """)
# Should create anonymous node anon_nodes = [n for n in graph.nodes.keys() if n.startswith("&__anon_")] assert len(anon_nodes) == 1 anon_name = anon_nodes[0]
anon_node = graph.nodes[anon_name] assert anon_node.opcode == ArithOp.SUB
def test_weak_edge_equivalent_to_strong(self, parser): """Verify weak edge produces same IR as equivalent strong edge.""" # Parse weak edge version graph_weak = parse_and_lower(parser, """\ &a <| pass &b <| pass &c <| pass &d <| pass &c, &d sub <| &a, &b """)
# Parse strong edge version graph_strong = parse_and_lower(parser, """\ &a <| pass &b <| pass &c <| pass &d <| pass sub &a, &b |> &c, &d """)
# Both should have one anonymous node anon_weak = [n for n in graph_weak.nodes.keys() if n.startswith("&__anon_")] anon_strong = [n for n in graph_strong.nodes.keys() if n.startswith("&__anon_")] assert len(anon_weak) == 1 assert len(anon_strong) == 1
# Both should have the same opcodes for the anon nodes assert graph_weak.nodes[anon_weak[0]].opcode == graph_strong.nodes[anon_strong[0]].opcode
class TestDataDef: """Tests for data definitions (AC2.5, AC2.6)."""
def test_basic_data_def(self, parser): """Parse basic data definition.""" graph = parse_and_lower(parser, """\ @hello|sm0:0 = 0x05 """)
assert len(graph.data_defs) == 1 data_def = graph.data_defs[0] assert data_def.name == "@hello" assert data_def.sm_id == 0 assert data_def.cell_addr == 0 assert data_def.value == 0x05
def test_data_def_with_different_sm(self, parser): """Parse data definition with different SM.""" graph = parse_and_lower(parser, """\ @data|sm1:2 = 0x42 """)
data_def = graph.data_defs[0] assert data_def.sm_id == 1 assert data_def.cell_addr == 2
def test_data_def_char_pair_big_endian(self, parser): """Parse data definition with char pair (big-endian packing) (AC2.6).""" graph = parse_and_lower(parser, """\ @hello|sm0:0 = 'h', 'e' """)
data_def = graph.data_defs[0] # 'h' = 0x68, 'e' = 0x65 # Big-endian: (0x68 << 8) | 0x65 = 0x6865 expected = (ord('h') << 8) | ord('e') assert data_def.value == expected
def test_data_def_char_pair_he_le(self, parser): """Verify big-endian packing of char pair.""" graph = parse_and_lower(parser, """\ @data|sm0:1 = 'l', 'l' """)
data_def = graph.data_defs[0] expected = (ord('l') << 8) | ord('l') assert data_def.value == expected
class TestSystemConfig: """Tests for system pragma (AC2.7)."""
def test_system_pragma_minimal(self, parser): """Parse minimal system pragma.""" graph = parse_and_lower(parser, """\ @system pe=4, sm=1 """)
assert graph.system is not None assert graph.system.pe_count == 4 assert graph.system.sm_count == 1 assert graph.system.iram_capacity == 256 # default assert graph.system.frame_count == 8 # default
def test_system_pragma_full(self, parser): """Parse full system pragma.""" graph = parse_and_lower(parser, """\ @system pe=2, sm=1, iram=128, frames=2 """)
assert graph.system.pe_count == 2 assert graph.system.sm_count == 1 assert graph.system.iram_capacity == 128 assert graph.system.frame_count == 2
def test_system_pragma_hex_values(self, parser): """Parse system pragma with hexadecimal values.""" graph = parse_and_lower(parser, """\ @system pe=0x04, sm=0x01 """)
assert graph.system.pe_count == 4 assert graph.system.sm_count == 1
class TestFunctionScoping: """Tests for function scoping (AC3.1, AC3.2, AC3.3, AC3.4)."""
def test_label_inside_function_qualified(self, parser): """Verify labels inside functions are qualified (AC3.1).""" graph = parse_and_lower(parser, """\ $main |> { &add <| add } """)
# Label should be qualified in the function region assert len(graph.regions) == 1 region = graph.regions[0] assert "$main.&add" in region.body.nodes node = region.body.nodes["$main.&add"] assert node.opcode == ArithOp.ADD
def test_global_node_not_qualified(self, parser): """Verify @nodes are never qualified (AC3.2).""" graph = parse_and_lower(parser, """\ @global <| pass """)
# Should not be qualified assert "@global" in graph.nodes assert "$main.@global" not in graph.nodes
def test_top_level_label_not_qualified(self, parser): """Verify top-level labels are not qualified (AC3.3).""" graph = parse_and_lower(parser, """\ &top <| pass """)
# Should not be qualified assert "&top" in graph.nodes assert "$main.&top" not in graph.nodes
def test_same_label_in_different_functions(self, parser): """Verify functions can each define &add without collision (AC3.4).""" graph = parse_and_lower(parser, """\ $foo |> { &add <| add } $bar |> { &add <| sub } """)
# Both should exist with different names in their respective regions assert len(graph.regions) == 2 foo_region = next(r for r in graph.regions if r.tag == "$foo") bar_region = next(r for r in graph.regions if r.tag == "$bar") assert "$foo.&add" in foo_region.body.nodes assert "$bar.&add" in bar_region.body.nodes assert foo_region.body.nodes["$foo.&add"].opcode == ArithOp.ADD assert bar_region.body.nodes["$bar.&add"].opcode == ArithOp.SUB
class TestRegions: """Tests for regions (AC3.7, AC3.8)."""
def test_function_region_creation(self, parser): """Verify function creates FUNCTION region (AC3.7).""" graph = parse_and_lower(parser, """\ $func |> { &a <| add } """)
assert len(graph.regions) == 1 region = graph.regions[0] assert region.tag == "$func" assert region.kind == RegionKind.FUNCTION assert "$func.&a" in region.body.nodes
def test_location_directive_creates_region(self, parser): """Verify location directive creates LOCATION region (AC3.8).""" graph = parse_and_lower(parser, """\ @data_section|sm0: """)
assert len(graph.regions) == 1 region = graph.regions[0] assert region.tag == "@data_section" assert region.kind == RegionKind.LOCATION
def test_location_directive_with_label_and_colon(self, parser): """AC6.1: Location directive with label and trailing colon.""" graph = parse_and_lower(parser, """\ §ion: """) # A bare label with colon becomes a location_dir assert len(graph.regions) == 1 region = graph.regions[0] assert region.kind == RegionKind.LOCATION
class TestErrorCases: """Tests for error handling (AC3.5, AC3.6)."""
def test_reserved_name_system_error(self, parser): """Verify reserved name @system produces error (AC3.5).""" graph = parse_and_lower(parser, """\ @system <| add """)
# Should have an error (note: @system is a keyword, might parse as pragma) # If it parses as inst_def, check for NAME error assert len(graph.errors) > 0 assert any(e.category == ErrorCategory.NAME for e in graph.errors)
def test_duplicate_label_in_function_error(self, parser): """Verify duplicate labels in same function produce error (AC3.6).""" graph = parse_and_lower(parser, """\ $main |> { &add <| add &add <| sub } """)
# Should have a SCOPE error assert len(graph.errors) > 0 assert any(e.category == ErrorCategory.SCOPE for e in graph.errors)
def test_duplicate_label_at_top_level_error(self, parser): """Verify duplicate labels at top level produce error.""" graph = parse_and_lower(parser, """\ &label <| add &label <| sub """)
# Should have a SCOPE error assert len(graph.errors) > 0 assert any(e.category == ErrorCategory.SCOPE for e in graph.errors)
class TestMemOps: """Tests for memory operations."""
def test_read_op(self, parser): """Parse READ operation.""" graph = parse_and_lower(parser, """\ &cell <| read """)
node = graph.nodes["&cell"] assert node.opcode == MemOp.READ
def test_write_op(self, parser): """Parse WRITE operation.""" graph = parse_and_lower(parser, """\ &cell <| write """)
node = graph.nodes["&cell"] assert node.opcode == MemOp.WRITE
def test_rd_inc_op(self, parser): """Parse RD_INC operation.""" graph = parse_and_lower(parser, """\ &cell <| rd_inc """)
node = graph.nodes["&cell"] assert node.opcode == MemOp.RD_INC
class TestEdgeCases: """Tests for edge cases and integration."""
def test_empty_program(self, parser): """Parse empty program.""" graph = parse_and_lower(parser, "")
assert len(graph.nodes) == 0 assert len(graph.edges) == 0
def test_program_with_comments(self, parser): """Parse program with comments.""" graph = parse_and_lower(parser, """\ &my_add <| add ; this is a comment &a |> &my_add:L ; wire a to add left """)
assert "&my_add" in graph.nodes assert len(graph.edges) == 1
def test_multiple_instructions(self, parser): """Parse multiple instructions.""" graph = parse_and_lower(parser, """\ &a <| add &b <| sub &c <| pass """)
assert len(graph.nodes) == 3 assert "&a" in graph.nodes assert "&b" in graph.nodes assert "&c" in graph.nodes
def test_complex_graph(self, parser): """Parse a more complex program.""" graph = parse_and_lower(parser, """\ @system pe=4, sm=1
&init <| const, 0 &loop_add <| add &cmp <| lte &branch <| breq
&init |> &loop_add:L &loop_add |> &cmp:L &cmp |> &branch:L """)
assert graph.system.pe_count == 4 assert len(graph.nodes) == 4 assert len(graph.edges) == 3
class TestScalingAnonymousCounters: """Tests that anonymous counter properly increments."""
def test_multiple_strong_edges_increment_counter(self, parser): """Verify each strong edge gets unique anonymous name.""" graph = parse_and_lower(parser, """\ &a <| pass &b <| pass &c <| pass &d <| pass &e <| pass &f <| pass add &a |> &b sub &c |> &d inc &e |> &f """)
anon_nodes = [n for n in graph.nodes.keys() if n.startswith("&__anon_")] assert len(anon_nodes) == 3 # Verify they have different counter values counters = set() for name in anon_nodes: counter_str = name.split("_")[-1] counters.add(int(counter_str)) assert len(counters) == 3