From 32d838e60291f21699be2973ab229a0020be4122 Mon Sep 17 00:00:00 2001 From: Alex Forencich Date: Sat, 15 Aug 2026 16:11:04 -0700 Subject: [PATCH] axis: Update testbenches to use cocotb.parametrize Signed-off-by: Alex Forencich --- .../test_taxi_axis_adapter.py | 61 +++++----- .../test_taxi_axis_arb_mux.py | 79 ++++++------- .../test_taxi_axis_async_fifo.py | 86 +++++++------- .../test_taxi_axis_async_fifo_adapter.py | 86 +++++++------- .../test_taxi_axis_broadcast.py | 46 ++++---- .../test_taxi_axis_cobs_decode.py | 68 ++++++----- .../test_taxi_axis_cobs_encode.py | 76 ++++++------- .../taxi_axis_concat/test_taxi_axis_concat.py | 58 +++++----- .../taxi_axis_demux/test_taxi_axis_demux.py | 55 +++++---- .../tb/taxi_axis_fifo/test_taxi_axis_fifo.py | 74 ++++++------ .../test_taxi_axis_fifo_adapter.py | 74 ++++++------ .../tb/taxi_axis_mux/test_taxi_axis_mux.py | 64 +++++------ .../tb/taxi_axis_pad/test_taxi_axis_pad.py | 106 +++++++++--------- .../test_taxi_axis_pipeline_fifo.py | 70 +++++------- .../test_taxi_axis_pipeline_register.py | 61 +++++----- .../test_taxi_axis_register.py | 61 +++++----- .../taxi_axis_switch/test_taxi_axis_switch.py | 86 +++++++------- 17 files changed, 556 insertions(+), 655 deletions(-) diff --git a/src/axis/tb/taxi_axis_adapter/test_taxi_axis_adapter.py b/src/axis/tb/taxi_axis_adapter/test_taxi_axis_adapter.py index dc55f95..81b2d75 100644 --- a/src/axis/tb/taxi_axis_adapter/test_taxi_axis_adapter.py +++ b/src/axis/tb/taxi_axis_adapter/test_taxi_axis_adapter.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -57,6 +56,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = max(len(cocotb.top.s_axis.tdata), len(cocotb.top.m_axis.tdata)) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -100,6 +120,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -121,6 +142,11 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -163,39 +189,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = max(len(cocotb.top.s_axis.tdata), len(cocotb.top.m_axis.tdata)) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [run_test_tuser_assert]: - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_arb_mux/test_taxi_axis_arb_mux.py b/src/axis/tb/taxi_axis_arb_mux/test_taxi_axis_arb_mux.py index ee355c0..bb612b5 100644 --- a/src/axis/tb/taxi_axis_arb_mux/test_taxi_axis_arb_mux.py +++ b/src/axis/tb/taxi_axis_arb_mux/test_taxi_axis_arb_mux.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge, Event -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -58,6 +57,33 @@ class TB(object): await RisingEdge(self.dut.clk) +ports = 1 +if getattr(cocotb, 'top', None) is not None: + ports = len(cocotb.top.s_axis) + + +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), + ("port", list(range(ports))), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None, port=0): tb = TB(dut) @@ -107,6 +133,10 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("port", list(range(ports))), +) async def run_test_tuser_assert(dut, port=0): tb = TB(dut) @@ -128,6 +158,7 @@ async def run_test_tuser_assert(dut, port=0): await RisingEdge(dut.clk) +@cocotb.test(skip=(ports < 2)) async def run_arb_test(dut): tb = TB(dut) @@ -191,6 +222,11 @@ async def run_arb_test(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -252,47 +288,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - ports = len(cocotb.top.s_axis) - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.add_option("port", list(range(ports))) - factory.generate_tests() - - for test in [run_test_tuser_assert]: - factory = TestFactory(test) - factory.add_option("port", list(range(ports))) - factory.generate_tests() - - if ports > 1: - factory = TestFactory(run_arb_test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_async_fifo/test_taxi_axis_async_fifo.py b/src/axis/tb/taxi_axis_async_fifo/test_taxi_axis_async_fifo.py index ccd3d6d..d2d6051 100644 --- a/src/axis/tb/taxi_axis_async_fifo/test_taxi_axis_async_fifo.py +++ b/src/axis/tb/taxi_axis_async_fifo/test_taxi_axis_async_fifo.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -89,6 +88,27 @@ class TB(object): await RisingEdge(self.dut.m_clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -128,6 +148,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -154,6 +175,7 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause(dut): tb = TB(dut) @@ -182,6 +204,7 @@ async def run_test_init_sink_pause(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause_reset(dut): tb = TB(dut) @@ -210,6 +233,7 @@ async def run_test_init_sink_pause_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause_source_reset(dut): tb = TB(dut) @@ -242,6 +266,7 @@ async def run_test_init_sink_pause_source_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause_sink_reset(dut): tb = TB(dut) @@ -270,6 +295,7 @@ async def run_test_init_sink_pause_sink_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_in_source_reset(dut): tb = TB(dut) @@ -302,6 +328,7 @@ async def run_test_shift_in_source_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_in_sink_reset(dut): tb = TB(dut) @@ -326,6 +353,7 @@ async def run_test_shift_in_sink_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_out_source_reset(dut): tb = TB(dut) @@ -356,6 +384,7 @@ async def run_test_shift_out_source_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_out_sink_reset(dut): tb = TB(dut) @@ -382,6 +411,7 @@ async def run_test_shift_out_sink_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_pause(dut): tb = TB(dut) @@ -432,6 +462,7 @@ async def run_test_pause(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_overflow(dut): tb = TB(dut) @@ -493,6 +524,7 @@ async def run_test_overflow(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_oversize(dut): tb = TB(dut) @@ -532,6 +564,11 @@ async def run_test_oversize(dut): await RisingEdge(dut.s_clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -598,53 +635,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.s_clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [ - run_test_tuser_assert, - run_test_init_sink_pause, - run_test_init_sink_pause_reset, - run_test_init_sink_pause_source_reset, - run_test_init_sink_pause_sink_reset, - run_test_shift_in_source_reset, - run_test_shift_in_sink_reset, - run_test_shift_out_source_reset, - run_test_shift_out_sink_reset, - run_test_pause, - run_test_overflow, - run_test_oversize - ]: - - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_async_fifo_adapter/test_taxi_axis_async_fifo_adapter.py b/src/axis/tb/taxi_axis_async_fifo_adapter/test_taxi_axis_async_fifo_adapter.py index ec9f4fa..ad3aeb1 100644 --- a/src/axis/tb/taxi_axis_async_fifo_adapter/test_taxi_axis_async_fifo_adapter.py +++ b/src/axis/tb/taxi_axis_async_fifo_adapter/test_taxi_axis_async_fifo_adapter.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -89,6 +88,27 @@ class TB(object): await RisingEdge(self.dut.m_clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = max(len(cocotb.top.s_axis.tdata), len(cocotb.top.m_axis.tdata)) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -128,6 +148,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -154,6 +175,7 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause(dut): tb = TB(dut) @@ -182,6 +204,7 @@ async def run_test_init_sink_pause(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause_reset(dut): tb = TB(dut) @@ -210,6 +233,7 @@ async def run_test_init_sink_pause_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause_source_reset(dut): tb = TB(dut) @@ -242,6 +266,7 @@ async def run_test_init_sink_pause_source_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_init_sink_pause_sink_reset(dut): tb = TB(dut) @@ -270,6 +295,7 @@ async def run_test_init_sink_pause_sink_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_in_source_reset(dut): tb = TB(dut) @@ -302,6 +328,7 @@ async def run_test_shift_in_source_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_in_sink_reset(dut): tb = TB(dut) @@ -326,6 +353,7 @@ async def run_test_shift_in_sink_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_out_source_reset(dut): tb = TB(dut) @@ -356,6 +384,7 @@ async def run_test_shift_out_source_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_shift_out_sink_reset(dut): tb = TB(dut) @@ -382,6 +411,7 @@ async def run_test_shift_out_sink_reset(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_pause(dut): tb = TB(dut) @@ -432,6 +462,7 @@ async def run_test_pause(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_overflow(dut): tb = TB(dut) @@ -493,6 +524,7 @@ async def run_test_overflow(dut): await RisingEdge(dut.s_clk) +@cocotb.test() async def run_test_oversize(dut): tb = TB(dut) @@ -532,6 +564,11 @@ async def run_test_oversize(dut): await RisingEdge(dut.s_clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -598,53 +635,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.s_clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = max(len(cocotb.top.m_axis.tdata), len(cocotb.top.s_axis.tdata)) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [ - run_test_tuser_assert, - run_test_init_sink_pause, - run_test_init_sink_pause_reset, - run_test_init_sink_pause_source_reset, - run_test_init_sink_pause_sink_reset, - run_test_shift_in_source_reset, - run_test_shift_in_sink_reset, - run_test_shift_out_source_reset, - run_test_shift_out_sink_reset, - run_test_pause, - run_test_overflow, - run_test_oversize - ]: - - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_broadcast/test_taxi_axis_broadcast.py b/src/axis/tb/taxi_axis_broadcast/test_taxi_axis_broadcast.py index 39d4bad..5d8384f 100644 --- a/src/axis/tb/taxi_axis_broadcast/test_taxi_axis_broadcast.py +++ b/src/axis/tb/taxi_axis_broadcast/test_taxi_axis_broadcast.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -57,6 +56,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis[0].tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -98,30 +118,6 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.s_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_cobs_decode/test_taxi_axis_cobs_decode.py b/src/axis/tb/taxi_axis_cobs_decode/test_taxi_axis_cobs_decode.py index 10867b9..f48f71c 100644 --- a/src/axis/tb/taxi_axis_cobs_decode/test_taxi_axis_cobs_decode.py +++ b/src/axis/tb/taxi_axis_cobs_decode/test_taxi_axis_cobs_decode.py @@ -18,7 +18,6 @@ import cocotb_test.simulator import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -123,34 +122,6 @@ class TB(object): await RisingEdge(self.dut.clk) -async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): - - tb = TB(dut) - - await tb.reset() - - tb.set_idle_generator(idle_inserter) - tb.set_backpressure_generator(backpressure_inserter) - - test_frames = [payload_data(x) for x in payload_lengths()] - - for test_data in test_frames: - enc = cobs_encode(test_data) - test_frame = AxiStreamFrame(enc) - await tb.source.send(test_frame) - - for test_data in test_frames: - rx_frame = await tb.sink.recv() - - assert rx_frame.tdata == test_data - assert not rx_frame.tuser - - assert tb.sink.empty() - - await RisingEdge(dut.clk) - await RisingEdge(dut.clk) - - def cycle_pause(): return itertools.cycle([1, 1, 1, 0]) @@ -180,14 +151,39 @@ def prbs_payload(length): return bytearray([next(gen) for x in range(length)]) -if getattr(cocotb, 'top', None) is not None: +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [zero_payload, nonzero_incrementing_payload, nonzero_incrementing_payload_zero_framed, prbs_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) +async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [zero_payload, nonzero_incrementing_payload, nonzero_incrementing_payload_zero_framed, prbs_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() + tb = TB(dut) + + await tb.reset() + + tb.set_idle_generator(idle_inserter) + tb.set_backpressure_generator(backpressure_inserter) + + test_frames = [payload_data(x) for x in payload_lengths()] + + for test_data in test_frames: + enc = cobs_encode(test_data) + test_frame = AxiStreamFrame(enc) + await tb.source.send(test_frame) + + for test_data in test_frames: + rx_frame = await tb.sink.recv() + + assert rx_frame.tdata == test_data + assert not rx_frame.tuser + + assert tb.sink.empty() + + await RisingEdge(dut.clk) + await RisingEdge(dut.clk) # cocotb-test diff --git a/src/axis/tb/taxi_axis_cobs_encode/test_taxi_axis_cobs_encode.py b/src/axis/tb/taxi_axis_cobs_encode/test_taxi_axis_cobs_encode.py index 648df88..ea3f1fa 100644 --- a/src/axis/tb/taxi_axis_cobs_encode/test_taxi_axis_cobs_encode.py +++ b/src/axis/tb/taxi_axis_cobs_encode/test_taxi_axis_cobs_encode.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -124,6 +123,42 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + return list(range(1, 33))+list(range(253, 259))+[512]+[1]*64 + + +def zero_payload(length): + return bytearray(length) + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +def nonzero_incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(1, 256)), length)) + + +def nonzero_incrementing_payload_zero_framed(length): + return bytearray([0]+list(itertools.islice(itertools.cycle(range(1, 256)), length))+[0]) + + +def prbs_payload(length): + gen = prbs31() + return bytearray([next(gen) for x in range(length)]) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [zero_payload, nonzero_incrementing_payload, nonzero_incrementing_payload_zero_framed, prbs_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -158,45 +193,6 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - return list(range(1, 33))+list(range(253, 259))+[512]+[1]*64 - - -def zero_payload(length): - return bytearray(length) - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -def nonzero_incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(1, 256)), length)) - - -def nonzero_incrementing_payload_zero_framed(length): - return bytearray([0]+list(itertools.islice(itertools.cycle(range(1, 256)), length))+[0]) - - -def prbs_payload(length): - gen = prbs31() - return bytearray([next(gen) for x in range(length)]) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [zero_payload, nonzero_incrementing_payload, nonzero_incrementing_payload_zero_framed, prbs_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_concat/test_taxi_axis_concat.py b/src/axis/tb/taxi_axis_concat/test_taxi_axis_concat.py index 688caa6..8229b3f 100644 --- a/src/axis/tb/taxi_axis_concat/test_taxi_axis_concat.py +++ b/src/axis/tb/taxi_axis_concat/test_taxi_axis_concat.py @@ -19,8 +19,7 @@ import pytest import cocotb from cocotb.clock import Clock -from cocotb.triggers import RisingEdge, Event -from cocotb.regression import TestFactory +from cocotb.triggers import RisingEdge from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -58,6 +57,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*3+1)) + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -100,6 +120,11 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -146,35 +171,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*3+1)) - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_demux/test_taxi_axis_demux.py b/src/axis/tb/taxi_axis_demux/test_taxi_axis_demux.py index 11b0d03..8cfb340 100644 --- a/src/axis/tb/taxi_axis_demux/test_taxi_axis_demux.py +++ b/src/axis/tb/taxi_axis_demux/test_taxi_axis_demux.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -61,6 +60,33 @@ class TB(object): await RisingEdge(self.dut.clk) +ports = 1 +if getattr(cocotb, 'top', None) is not None: + ports = len(cocotb.top.m_axis) + + +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.s_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), + ("port", list(range(ports))), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None, port=0): tb = TB(dut) @@ -110,33 +136,6 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.s_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - ports = len(cocotb.top.m_axis) - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.add_option("port", list(range(ports))) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_fifo/test_taxi_axis_fifo.py b/src/axis/tb/taxi_axis_fifo/test_taxi_axis_fifo.py index 33b09b3..f7e818a 100644 --- a/src/axis/tb/taxi_axis_fifo/test_taxi_axis_fifo.py +++ b/src/axis/tb/taxi_axis_fifo/test_taxi_axis_fifo.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -59,6 +58,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -98,6 +118,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -124,6 +145,7 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_init_sink_pause(dut): tb = TB(dut) @@ -152,6 +174,7 @@ async def run_test_init_sink_pause(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_init_sink_pause_reset(dut): tb = TB(dut) @@ -180,6 +203,7 @@ async def run_test_init_sink_pause_reset(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_pause(dut): tb = TB(dut) @@ -218,6 +242,7 @@ async def run_test_pause(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_overflow(dut): tb = TB(dut) @@ -279,6 +304,7 @@ async def run_test_overflow(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_oversize(dut): tb = TB(dut) @@ -318,6 +344,11 @@ async def run_test_oversize(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -386,47 +417,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [ - run_test_tuser_assert, - run_test_init_sink_pause, - run_test_init_sink_pause_reset, - run_test_pause, - run_test_overflow, - run_test_oversize - ]: - - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_fifo_adapter/test_taxi_axis_fifo_adapter.py b/src/axis/tb/taxi_axis_fifo_adapter/test_taxi_axis_fifo_adapter.py index a8ee344..d798e7f 100644 --- a/src/axis/tb/taxi_axis_fifo_adapter/test_taxi_axis_fifo_adapter.py +++ b/src/axis/tb/taxi_axis_fifo_adapter/test_taxi_axis_fifo_adapter.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -59,6 +58,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = max(len(cocotb.top.s_axis.tdata), len(cocotb.top.m_axis.tdata)) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -98,6 +118,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -124,6 +145,7 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_init_sink_pause(dut): tb = TB(dut) @@ -152,6 +174,7 @@ async def run_test_init_sink_pause(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_init_sink_pause_reset(dut): tb = TB(dut) @@ -180,6 +203,7 @@ async def run_test_init_sink_pause_reset(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_pause(dut): tb = TB(dut) @@ -218,6 +242,7 @@ async def run_test_pause(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_overflow(dut): tb = TB(dut) @@ -279,6 +304,7 @@ async def run_test_overflow(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_oversize(dut): tb = TB(dut) @@ -318,6 +344,11 @@ async def run_test_oversize(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -384,47 +415,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = max(len(cocotb.top.m_axis.tdata), len(cocotb.top.s_axis.tdata)) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [ - run_test_tuser_assert, - run_test_init_sink_pause, - run_test_init_sink_pause_reset, - run_test_pause, - run_test_overflow, - run_test_oversize - ]: - - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_mux/test_taxi_axis_mux.py b/src/axis/tb/taxi_axis_mux/test_taxi_axis_mux.py index a362932..ae6cd86 100644 --- a/src/axis/tb/taxi_axis_mux/test_taxi_axis_mux.py +++ b/src/axis/tb/taxi_axis_mux/test_taxi_axis_mux.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -60,6 +59,33 @@ class TB(object): await RisingEdge(self.dut.clk) +ports = 1 +if getattr(cocotb, 'top', None) is not None: + ports = len(cocotb.top.s_axis) + + +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), + ("port", list(range(ports))), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None, port=0): tb = TB(dut) @@ -102,6 +128,10 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("port", list(range(ports))), +) async def run_test_tuser_assert(dut, port=0): tb = TB(dut) @@ -126,38 +156,6 @@ async def run_test_tuser_assert(dut, port=0): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - ports = len(cocotb.top.s_axis) - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.add_option("port", list(range(ports))) - factory.generate_tests() - - for test in [run_test_tuser_assert]: - factory = TestFactory(test) - factory.add_option("port", list(range(ports))) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_pad/test_taxi_axis_pad.py b/src/axis/tb/taxi_axis_pad/test_taxi_axis_pad.py index 1ada08e..bbcef31 100644 --- a/src/axis/tb/taxi_axis_pad/test_taxi_axis_pad.py +++ b/src/axis/tb/taxi_axis_pad/test_taxi_axis_pad.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge, FallingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -79,6 +78,36 @@ class TB(object): self.stats[stat] += int(getattr(self.dut, stat).value) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, max(byte_lanes*4, 128))) + [512, 1514, 9214] + [1]*10 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +underflow_drop = False +idle_list = [None, cycle_pause] +if getattr(cocotb, 'top', None) is not None: + if cocotb.top.UNDERFLOW_DROP_EN.value: + underflow_drop = True + idle_list = [None] + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("pad_en", [True, False]), + ("idle_inserter", idle_list), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, pad_en=True, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -128,6 +157,12 @@ async def run_test(dut, pad_en=True, payload_lengths=None, payload_data=None, id await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("pad_en", [True, False]), + ("idle_inserter", idle_list), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_pad(dut, pad_en=True, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -202,6 +237,12 @@ async def run_test_pad(dut, pad_en=True, idle_inserter=None, backpressure_insert await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("pad_en", [True, False]), + ("idle_inserter", idle_list), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_tuser_assert(dut, pad_en=True, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -284,6 +325,11 @@ async def run_test_tuser_assert(dut, pad_en=True, idle_inserter=None, backpressu await RisingEdge(dut.clk) +@cocotb.test(skip=(not underflow_drop)) +@cocotb.parametrize( + ("pad_en", [True, False]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_underflow(dut, pad_en=True, backpressure_inserter=None): tb = TB(dut) @@ -386,6 +432,12 @@ async def run_test_underflow(dut, pad_en=True, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("pad_en", [True, False]), + ("idle_inserter", idle_list), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, pad_en=True, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -438,58 +490,6 @@ async def run_stress_test(dut, pad_en=True, idle_inserter=None, backpressure_ins await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_lanes = data_width // 8 - return list(range(1, max(byte_lanes*4, 128))) + [512, 1514, 9214] + [1]*10 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("pad_en", [True, False]) - factory.add_option("payload_data", [incrementing_payload]) - if not cocotb.top.UNDERFLOW_DROP_EN.value: - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [ - run_test_pad, - run_test_tuser_assert, - ]: - - factory = TestFactory(test) - factory.add_option("pad_en", [True, False]) - if not cocotb.top.UNDERFLOW_DROP_EN.value: - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - if cocotb.top.UNDERFLOW_DROP_EN.value: - for test in [run_test_underflow]: - factory = TestFactory(test) - factory.add_option("pad_en", [True, False]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("pad_en", [True, False]) - if not cocotb.top.UNDERFLOW_DROP_EN.value: - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_pipeline_fifo/test_taxi_axis_pipeline_fifo.py b/src/axis/tb/taxi_axis_pipeline_fifo/test_taxi_axis_pipeline_fifo.py index a3895c9..20f1428 100644 --- a/src/axis/tb/taxi_axis_pipeline_fifo/test_taxi_axis_pipeline_fifo.py +++ b/src/axis/tb/taxi_axis_pipeline_fifo/test_taxi_axis_pipeline_fifo.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -57,6 +56,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -96,6 +116,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -117,6 +138,7 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_init_sink_pause(dut): tb = TB(dut) @@ -145,6 +167,7 @@ async def run_test_init_sink_pause(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_init_sink_pause_reset(dut): tb = TB(dut) @@ -173,6 +196,7 @@ async def run_test_init_sink_pause_reset(dut): await RisingEdge(dut.clk) +@cocotb.test() async def run_test_overflow(dut): tb = TB(dut) @@ -201,6 +225,11 @@ async def run_test_overflow(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -243,45 +272,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [ - run_test_tuser_assert, - run_test_init_sink_pause, - run_test_init_sink_pause_reset, - run_test_overflow - ]: - - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_pipeline_register/test_taxi_axis_pipeline_register.py b/src/axis/tb/taxi_axis_pipeline_register/test_taxi_axis_pipeline_register.py index f6ba18f..9c0a446 100644 --- a/src/axis/tb/taxi_axis_pipeline_register/test_taxi_axis_pipeline_register.py +++ b/src/axis/tb/taxi_axis_pipeline_register/test_taxi_axis_pipeline_register.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -57,6 +56,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -96,6 +116,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -117,6 +138,11 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -159,39 +185,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [run_test_tuser_assert]: - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_register/test_taxi_axis_register.py b/src/axis/tb/taxi_axis_register/test_taxi_axis_register.py index 49a69ed..47fe04a 100644 --- a/src/axis/tb/taxi_axis_register/test_taxi_axis_register.py +++ b/src/axis/tb/taxi_axis_register/test_taxi_axis_register.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -57,6 +56,27 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis.tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -96,6 +116,7 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() async def run_test_tuser_assert(dut): tb = TB(dut) @@ -117,6 +138,11 @@ async def run_test_tuser_assert(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -159,39 +185,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.m_axis.tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - for test in [run_test_tuser_assert]: - factory = TestFactory(test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/axis/tb/taxi_axis_switch/test_taxi_axis_switch.py b/src/axis/tb/taxi_axis_switch/test_taxi_axis_switch.py index 62cd9f1..66ce09d 100644 --- a/src/axis/tb/taxi_axis_switch/test_taxi_axis_switch.py +++ b/src/axis/tb/taxi_axis_switch/test_taxi_axis_switch.py @@ -20,7 +20,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge, Event -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -59,6 +58,36 @@ class TB(object): await RisingEdge(self.dut.clk) +s_count = 1 +m_count = 1 +if getattr(cocotb, 'top', None) is not None: + s_count = len(cocotb.top.s_axis) + m_count = len(cocotb.top.m_axis) + + +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +def size_list(): + data_width = len(cocotb.top.m_axis[0].tdata) + byte_lanes = data_width // 8 + return list(range(1, byte_lanes*4+1))+[512]+[1]*64 + + +def incrementing_payload(length): + return bytearray(itertools.islice(itertools.cycle(range(256)), length)) + + +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), + ("s", range(min(s_count, 2))), + ("m", range(min(m_count, 2))), +) async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None, s=0, m=0): tb = TB(dut) @@ -110,6 +139,11 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("s", range(min(s_count, 2))), + ("m", range(min(m_count, 2))), +) async def run_test_tuser_assert(dut, s=0, m=0): tb = TB(dut) @@ -134,6 +168,7 @@ async def run_test_tuser_assert(dut, s=0, m=0): await RisingEdge(dut.clk) +@cocotb.test(skip=(s_count < 2)) async def run_arb_test(dut): tb = TB(dut) @@ -199,6 +234,11 @@ async def run_arb_test(dut): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -264,50 +304,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - data_width = len(cocotb.top.s_axis[0].tdata) - byte_width = data_width // 8 - return list(range(1, byte_width*4+1))+[512]+[1]*64 - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - s_count = len(cocotb.top.s_axis) - m_count = len(cocotb.top.m_axis) - - factory = TestFactory(run_test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.add_option("s", range(min(s_count, 2))) - factory.add_option("m", range(min(m_count, 2))) - factory.generate_tests() - - for test in [run_test_tuser_assert]: - factory = TestFactory(test) - factory.add_option("s", range(min(s_count, 2))) - factory.add_option("m", range(min(m_count, 2))) - factory.generate_tests() - - if s_count > 1: - factory = TestFactory(run_arb_test) - factory.generate_tests() - - factory = TestFactory(run_stress_test) - factory.add_option("idle_inserter", [None, cycle_pause]) - factory.add_option("backpressure_inserter", [None, cycle_pause]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__)