diff --git a/axi/tb/axis_cocotb_loopback_test/Makefile b/axi/tb/axis_cocotb_loopback_test/Makefile new file mode 100644 index 0000000..2918e17 --- /dev/null +++ b/axi/tb/axis_cocotb_loopback_test/Makefile @@ -0,0 +1,23 @@ +TOPLEVEL_LANG = verilog +SIM ?= verilator + +PWD := $(shell pwd) +RTL_DIR ?= $(PWD)/../../rtl +TB_DIR ?= $(PWD) + +TOPLEVEL = tb_axis_loopback +MODULE = test_axis_loopback + +VERILOG_SOURCES += $(RTL_DIR)/axis_if.sv +VERILOG_SOURCES += $(RTL_DIR)/axis_flat_to_if.sv +VERILOG_SOURCES += $(RTL_DIR)/axis_if_to_flat.sv +VERILOG_SOURCES += $(RTL_DIR)/axis_desc_flat_to_if.sv +VERILOG_SOURCES += $(RTL_DIR)/axis_desc_if_to_flat.sv +VERILOG_SOURCES += $(TB_DIR)/axis_loopback.sv +VERILOG_SOURCES += $(TB_DIR)/tb_axis_loopback.sv + +COMPILE_ARGS += -I$(RTL_DIR) +EXTRA_ARGS += --trace +EXTRA_ARGS += --trace-structs + +include $(shell cocotb-config --makefiles)/Makefile.sim diff --git a/axi/tb/axis_cocotb_loopback_test/test_axis_loopback.py b/axi/tb/axis_cocotb_loopback_test/test_axis_loopback.py new file mode 100644 index 0000000..823c1ad --- /dev/null +++ b/axi/tb/axis_cocotb_loopback_test/test_axis_loopback.py @@ -0,0 +1,126 @@ +import random + +import cocotb +from cocotb.clock import Clock +from cocotb.triggers import RisingEdge, Timer, with_timeout + + +CLK_PERIOD_NS = 10 + + +async def reset_dut(dut): + dut.rst.value = 1 + + dut.s_axis_tdata.value = 0 + dut.s_axis_tkeep.value = 0 + dut.s_axis_tstrb.value = 0 + dut.s_axis_tlast.value = 0 + dut.s_axis_tid.value = 0 + dut.s_axis_tdest.value = 0 + dut.s_axis_tuser.value = 0 + dut.s_axis_tvalid.value = 0 + dut.m_axis_tready.value = 0 + + for _ in range(5): + await RisingEdge(dut.clk) + + dut.rst.value = 0 + + for _ in range(2): + await RisingEdge(dut.clk) + + +async def send_axis_beat(dut, beat): + dut.s_axis_tdata.value = beat["data"] + dut.s_axis_tkeep.value = beat["keep"] + dut.s_axis_tstrb.value = beat["strb"] + dut.s_axis_tlast.value = beat["last"] + dut.s_axis_tid.value = beat["id"] + dut.s_axis_tdest.value = beat["dest"] + dut.s_axis_tuser.value = beat["user"] + dut.s_axis_tvalid.value = 1 + + while True: + await RisingEdge(dut.clk) + if int(dut.s_axis_tready.value): + break + + dut.s_axis_tvalid.value = 0 + + +async def recv_axis_beats(dut, count, ready_pattern=None): + beats = [] + cycle = 0 + + while len(beats) < count: + ready = 1 + if ready_pattern is not None: + ready = ready_pattern[cycle % len(ready_pattern)] + dut.m_axis_tready.value = ready + + await RisingEdge(dut.clk) + + if ready and int(dut.m_axis_tvalid.value): + beats.append({ + "data": int(dut.m_axis_tdata.value), + "keep": int(dut.m_axis_tkeep.value), + "strb": int(dut.m_axis_tstrb.value), + "last": int(dut.m_axis_tlast.value), + "id": int(dut.m_axis_tid.value), + "dest": int(dut.m_axis_tdest.value), + "user": int(dut.m_axis_tuser.value), + }) + + cycle += 1 + + dut.m_axis_tready.value = 0 + return beats + + +@cocotb.test() +async def test_axis_loopback_basic(dut): + cocotb.start_soon(Clock(dut.clk, CLK_PERIOD_NS, units="ns").start()) + await reset_dut(dut) + + tx = [ + {"data": 0x1122334455667788, "keep": 0xFF, "strb": 0xFF, + "last": 0, "id": 1, "dest": 2, "user": 3}, + {"data": 0xAABBCCDDEEFF0011, "keep": 0xFF, "strb": 0xFF, + "last": 1, "id": 1, "dest": 2, "user": 4}, + ] + + rx_task = cocotb.start_soon(recv_axis_beats(dut, len(tx))) + + for beat in tx: + await send_axis_beat(dut, beat) + + rx = await with_timeout(rx_task, 1, "us") + assert rx == tx + + +@cocotb.test() +async def test_axis_loopback_with_backpressure(dut): + cocotb.start_soon(Clock(dut.clk, CLK_PERIOD_NS, units="ns").start()) + await reset_dut(dut) + + random.seed(1) + tx = [] + for i in range(16): + tx.append({ + "data": random.getrandbits(64), + "keep": 0xFF, + "strb": 0xFF, + "last": int(i == 15), + "id": i & 0xFF, + "dest": (i + 1) & 0xFF, + "user": (i + 2) & 0xFF, + }) + + rx_task = cocotb.start_soon(recv_axis_beats( + dut, len(tx), ready_pattern=[1, 0, 1, 1, 0])) + + for beat in tx: + await send_axis_beat(dut, beat) + + rx = await with_timeout(rx_task, 5, "us") + assert rx == tx