tests: add axis loopback test

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

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