From 1cf84f107df384ba5177dbf9d321f3076cac408f Mon Sep 17 00:00:00 2001 From: Alex Forencich Date: Sat, 15 Aug 2026 13:12:34 -0700 Subject: [PATCH] xfcp: Update testbenches to use cocotb.parametrize Signed-off-by: Alex Forencich --- .../test_taxi_xfcp_if_uart.py | 40 +++++++------- .../test_taxi_xfcp_mod_apb.py | 34 ++++++------ .../test_taxi_xfcp_mod_axi.py | 34 ++++++------ .../test_taxi_xfcp_mod_axil.py | 34 ++++++------ .../test_taxi_xfcp_mod_i2c_master.py | 44 +++++++++------- .../taxi_xfcp_switch/test_taxi_xfcp_switch.py | 52 +++++++++---------- 6 files changed, 125 insertions(+), 113 deletions(-) diff --git a/src/xfcp/tb/taxi_xfcp_if_uart/test_taxi_xfcp_if_uart.py b/src/xfcp/tb/taxi_xfcp_if_uart/test_taxi_xfcp_if_uart.py index 0805fca..f08808d 100644 --- a/src/xfcp/tb/taxi_xfcp_if_uart/test_taxi_xfcp_if_uart.py +++ b/src/xfcp/tb/taxi_xfcp_if_uart/test_taxi_xfcp_if_uart.py @@ -19,7 +19,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, AxiStreamSource, AxiStreamSink from cocotbext.uart import UartSource, UartSink @@ -64,6 +63,19 @@ class TB(object): await RisingEdge(self.dut.clk) +def size_list(): + return list(range(1, 16)) + [128] + + +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]), +) async def run_test_tx(dut, payload_lengths=None, payload_data=None): tb = TB(dut) @@ -98,6 +110,11 @@ async def run_test_tx(dut, payload_lengths=None, payload_data=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("payload_lengths", [size_list]), + ("payload_data", [incrementing_payload]), +) async def run_test_rx(dut, payload_lengths=None, payload_data=None): tb = TB(dut) @@ -126,27 +143,6 @@ async def run_test_rx(dut, payload_lengths=None, payload_data=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -def size_list(): - return list(range(1, 16)) + [128] - - -def incrementing_payload(length): - return bytearray(itertools.islice(itertools.cycle(range(256)), length)) - - -if getattr(cocotb, 'top', None) is not None: - - for test in [run_test_tx, run_test_rx]: - factory = TestFactory(test) - factory.add_option("payload_lengths", [size_list]) - factory.add_option("payload_data", [incrementing_payload]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/xfcp/tb/taxi_xfcp_mod_apb/test_taxi_xfcp_mod_apb.py b/src/xfcp/tb/taxi_xfcp_mod_apb/test_taxi_xfcp_mod_apb.py index 0b4f6eb..4df14f6 100644 --- a/src/xfcp/tb/taxi_xfcp_mod_apb/test_taxi_xfcp_mod_apb.py +++ b/src/xfcp/tb/taxi_xfcp_mod_apb/test_taxi_xfcp_mod_apb.py @@ -21,7 +21,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, AxiStreamSource, AxiStreamSink from cocotbext.axi import ApbBus, ApbRam @@ -72,6 +71,15 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -117,6 +125,11 @@ async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -155,6 +168,11 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -183,20 +201,6 @@ async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - for test in [run_test_write, run_test_read, run_test_id]: - - factory = TestFactory(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/xfcp/tb/taxi_xfcp_mod_axi/test_taxi_xfcp_mod_axi.py b/src/xfcp/tb/taxi_xfcp_mod_axi/test_taxi_xfcp_mod_axi.py index a425249..4df02d5 100644 --- a/src/xfcp/tb/taxi_xfcp_mod_axi/test_taxi_xfcp_mod_axi.py +++ b/src/xfcp/tb/taxi_xfcp_mod_axi/test_taxi_xfcp_mod_axi.py @@ -21,7 +21,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, AxiStreamSource, AxiStreamSink from cocotbext.axi import AxiBus, AxiRam @@ -76,6 +75,15 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -121,6 +129,11 @@ async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -159,6 +172,11 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -187,20 +205,6 @@ async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - for test in [run_test_write, run_test_read, run_test_id]: - - factory = TestFactory(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/xfcp/tb/taxi_xfcp_mod_axil/test_taxi_xfcp_mod_axil.py b/src/xfcp/tb/taxi_xfcp_mod_axil/test_taxi_xfcp_mod_axil.py index 8146d56..2d1a5b1 100644 --- a/src/xfcp/tb/taxi_xfcp_mod_axil/test_taxi_xfcp_mod_axil.py +++ b/src/xfcp/tb/taxi_xfcp_mod_axil/test_taxi_xfcp_mod_axil.py @@ -21,7 +21,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, AxiStreamSource, AxiStreamSink from cocotbext.axi import AxiLiteBus, AxiLiteRam @@ -76,6 +75,15 @@ class TB(object): await RisingEdge(self.dut.clk) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -121,6 +129,11 @@ async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -159,6 +172,11 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -187,20 +205,6 @@ async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - for test in [run_test_write, run_test_read, run_test_id]: - - factory = TestFactory(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/xfcp/tb/taxi_xfcp_mod_i2c_master/test_taxi_xfcp_mod_i2c_master.py b/src/xfcp/tb/taxi_xfcp_mod_i2c_master/test_taxi_xfcp_mod_i2c_master.py index a29ffbf..dc14b17 100644 --- a/src/xfcp/tb/taxi_xfcp_mod_i2c_master/test_taxi_xfcp_mod_i2c_master.py +++ b/src/xfcp/tb/taxi_xfcp_mod_i2c_master/test_taxi_xfcp_mod_i2c_master.py @@ -20,7 +20,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, AxiStreamSource, AxiStreamSink from cocotbext.i2c import I2cMemory @@ -117,6 +116,15 @@ class TB(object): self.log.debug("RX packet: %s", rx_pkt) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -171,6 +179,11 @@ async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -225,6 +238,11 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_nack(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -265,6 +283,11 @@ async def run_test_nack(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -293,25 +316,6 @@ async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - for test in [ - run_test_write, - run_test_read, - run_test_nack, - run_test_id, - ]: - - factory = TestFactory(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/xfcp/tb/taxi_xfcp_switch/test_taxi_xfcp_switch.py b/src/xfcp/tb/taxi_xfcp_switch/test_taxi_xfcp_switch.py index 096cf1b..77773a4 100644 --- a/src/xfcp/tb/taxi_xfcp_switch/test_taxi_xfcp_switch.py +++ b/src/xfcp/tb/taxi_xfcp_switch/test_taxi_xfcp_switch.py @@ -12,7 +12,6 @@ Authors: import itertools import logging import os -import struct import sys import cocotb_test.simulator @@ -21,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, AxiStreamSource, AxiStreamSink @@ -75,6 +73,21 @@ class TB(object): await RisingEdge(self.dut.clk) +ports = 1 +if getattr(cocotb, 'top', None) is not None: + ports = len(cocotb.top.xfcp_dsp_us) + + +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), + ("port", list(range(ports))), +) async def run_test_downstream(dut, idle_inserter=None, backpressure_inserter=None, port=0): tb = TB(dut) @@ -106,6 +119,12 @@ async def run_test_downstream(dut, idle_inserter=None, backpressure_inserter=Non await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), + ("port", list(range(ports))), +) async def run_test_upstream(dut, idle_inserter=None, backpressure_inserter=None, port=0): tb = TB(dut) @@ -136,6 +155,11 @@ async def run_test_upstream(dut, idle_inserter=None, backpressure_inserter=None, await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + ("idle_inserter", [None, cycle_pause]), + ("backpressure_inserter", [None, cycle_pause]), +) async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -164,30 +188,6 @@ async def run_test_id(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - ports = len(cocotb.top.xfcp_dsp_us) - - for test in [run_test_downstream, run_test_upstream]: - - factory = TestFactory(test) - 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_id]: - - factory = TestFactory(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__)