From ca26a48a2863dba3e388f1d918318ce0f1466645 Mon Sep 17 00:00:00 2001 From: Alex Forencich Date: Sat, 15 Aug 2026 13:13:14 -0700 Subject: [PATCH] dma: Update testbenches to use cocotb.parametrize Signed-off-by: Alex Forencich --- .../tb/taxi_axi_cdma/test_taxi_axi_cdma.py | 26 +++++-------- src/dma/tb/taxi_axi_dma/test_taxi_axi_dma.py | 33 ++++++++-------- .../test_taxi_dma_client_axis_sink.py | 24 +++++------- .../test_taxi_dma_client_axis_source.py | 24 +++++------- .../taxi_dma_if_axi/test_taxi_dma_if_axi.py | 34 +++++++++-------- .../test_taxi_dma_if_axi_rd.py | 22 +++++------ .../test_taxi_dma_if_axi_wr.py | 29 +++++++------- .../test_taxi_dma_if_pcie_us.py | 37 +++++++++--------- .../test_taxi_dma_if_pcie_us_rd.py | 29 ++++++-------- .../test_taxi_dma_if_pcie_us_wr.py | 20 ++++------ .../taxi_dma_psdpram/test_taxi_dma_psdpram.py | 38 ++++++++++--------- .../test_taxi_dma_psdpram_async.py | 38 ++++++++++--------- 12 files changed, 169 insertions(+), 185 deletions(-) diff --git a/src/dma/tb/taxi_axi_cdma/test_taxi_axi_cdma.py b/src/dma/tb/taxi_axi_cdma/test_taxi_axi_cdma.py index 421c2ce..570c0f8 100644 --- a/src/dma/tb/taxi_axi_cdma/test_taxi_axi_cdma.py +++ b/src/dma/tb/taxi_axi_cdma/test_taxi_axi_cdma.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 AxiBus, AxiRam from cocotbext.axi.stream import define_stream @@ -77,7 +76,16 @@ class TB(object): await RisingEdge(self.dut.clk) -async def run_test(dut, data_in=None, idle_inserter=None, backpressure_inserter=None): +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(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -124,20 +132,6 @@ async def run_test(dut, data_in=None, idle_inserter=None, backpressure_inserter= 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]: - - 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/dma/tb/taxi_axi_dma/test_taxi_axi_dma.py b/src/dma/tb/taxi_axi_dma/test_taxi_axi_dma.py index 03c0c40..89d3246 100644 --- a/src/dma/tb/taxi_axi_dma/test_taxi_axi_dma.py +++ b/src/dma/tb/taxi_axi_dma/test_taxi_axi_dma.py @@ -20,7 +20,6 @@ import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge -from cocotb.regression import TestFactory from cocotbext.axi import AxiBus, AxiRam from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink @@ -90,7 +89,16 @@ class TB(object): await RisingEdge(self.dut.clk) -async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_inserter=None): +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) @@ -146,7 +154,12 @@ async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_ins await RisingEdge(dut.clk) -async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inserter=None): +@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) @@ -195,20 +208,6 @@ async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inse 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]: - - 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/dma/tb/taxi_dma_client_axis_sink/test_taxi_dma_client_axis_sink.py b/src/dma/tb/taxi_dma_client_axis_sink/test_taxi_dma_client_axis_sink.py index 0ce1f2e..160b6de 100644 --- a/src/dma/tb/taxi_dma_client_axis_sink/test_taxi_dma_client_axis_sink.py +++ b/src/dma/tb/taxi_dma_client_axis_sink/test_taxi_dma_client_axis_sink.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 from cocotbext.axi.stream import define_stream @@ -89,7 +88,16 @@ class TB(object): await RisingEdge(self.dut.clk) -async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_inserter=None): +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) @@ -144,18 +152,6 @@ async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_ins await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test_write) - 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/dma/tb/taxi_dma_client_axis_source/test_taxi_dma_client_axis_source.py b/src/dma/tb/taxi_dma_client_axis_source/test_taxi_dma_client_axis_source.py index 3ea2344..8b610f4 100644 --- a/src/dma/tb/taxi_dma_client_axis_source/test_taxi_dma_client_axis_source.py +++ b/src/dma/tb/taxi_dma_client_axis_source/test_taxi_dma_client_axis_source.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, AxiStreamSink from cocotbext.axi.stream import define_stream @@ -87,7 +86,16 @@ class TB(object): await RisingEdge(self.dut.clk) -async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inserter=None): +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_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -135,18 +143,6 @@ async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inse await RisingEdge(dut.clk) -def cycle_pause(): - return itertools.cycle([1, 1, 1, 0]) - - -if getattr(cocotb, 'top', None) is not None: - - factory = TestFactory(run_test_read) - 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/dma/tb/taxi_dma_if_axi/test_taxi_dma_if_axi.py b/src/dma/tb/taxi_dma_if_axi/test_taxi_dma_if_axi.py index bc80a83..09c432a 100644 --- a/src/dma/tb/taxi_dma_if_axi/test_taxi_dma_if_axi.py +++ b/src/dma/tb/taxi_dma_if_axi/test_taxi_dma_if_axi.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 AxiBus, AxiRam from cocotbext.axi.stream import define_stream @@ -96,6 +95,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) @@ -152,6 +160,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) @@ -208,6 +221,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_write_imm(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -260,20 +278,6 @@ async def run_test_write_imm(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_write_imm]: - - 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/dma/tb/taxi_dma_if_axi_rd/test_taxi_dma_if_axi_rd.py b/src/dma/tb/taxi_dma_if_axi_rd/test_taxi_dma_if_axi_rd.py index 1844306..648e9e4 100644 --- a/src/dma/tb/taxi_dma_if_axi_rd/test_taxi_dma_if_axi_rd.py +++ b/src/dma/tb/taxi_dma_if_axi_rd/test_taxi_dma_if_axi_rd.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 AxiReadBus, AxiRamRead from cocotbext.axi.stream import define_stream @@ -88,6 +87,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_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -144,18 +152,6 @@ async def run_test_read(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: - - factory = TestFactory(run_test_read) - 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/dma/tb/taxi_dma_if_axi_wr/test_taxi_dma_if_axi_wr.py b/src/dma/tb/taxi_dma_if_axi_wr/test_taxi_dma_if_axi_wr.py index 01b075a..892879c 100644 --- a/src/dma/tb/taxi_dma_if_axi_wr/test_taxi_dma_if_axi_wr.py +++ b/src/dma/tb/taxi_dma_if_axi_wr/test_taxi_dma_if_axi_wr.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 AxiWriteBus, AxiRamWrite from cocotbext.axi.stream import define_stream @@ -89,6 +88,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) @@ -145,6 +153,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_write_imm(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -197,20 +210,6 @@ async def run_test_write_imm(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_write_imm]: - - 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/dma/tb/taxi_dma_if_pcie_us/test_taxi_dma_if_pcie_us.py b/src/dma/tb/taxi_dma_if_pcie_us/test_taxi_dma_if_pcie_us.py index 6ae981f..8af48c9 100644 --- a/src/dma/tb/taxi_dma_if_pcie_us/test_taxi_dma_if_pcie_us.py +++ b/src/dma/tb/taxi_dma_if_pcie_us/test_taxi_dma_if_pcie_us.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.triggers import RisingEdge, FallingEdge, Timer -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus from cocotbext.pcie.core import RootComplex @@ -183,6 +182,15 @@ class TB(object): self.stat_err_uncor_asserted = True +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) @@ -245,6 +253,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) @@ -305,6 +318,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_read_errors(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -376,23 +394,6 @@ async def run_test_read_errors(dut, idle_inserter=None, backpressure_inserter=No 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_read_errors, - ]: - - factory = TestFactory(test) - factory.add_option(("idle_inserter", "backpressure_inserter"), [(None, None), (cycle_pause, cycle_pause)]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/dma/tb/taxi_dma_if_pcie_us_rd/test_taxi_dma_if_pcie_us_rd.py b/src/dma/tb/taxi_dma_if_pcie_us_rd/test_taxi_dma_if_pcie_us_rd.py index c3d0ab8..b3ebb6b 100644 --- a/src/dma/tb/taxi_dma_if_pcie_us_rd/test_taxi_dma_if_pcie_us_rd.py +++ b/src/dma/tb/taxi_dma_if_pcie_us_rd/test_taxi_dma_if_pcie_us_rd.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.triggers import RisingEdge, FallingEdge, Timer -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus from cocotbext.pcie.core import RootComplex @@ -175,6 +174,14 @@ class TB(object): self.stat_err_uncor_asserted = True +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + (("idle_inserter", "backpressure_inserter"), [(None, None), (cycle_pause, cycle_pause)]), +) async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -240,6 +247,10 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk) +@cocotb.test() +@cocotb.parametrize( + (("idle_inserter", "backpressure_inserter"), [(None, None), (cycle_pause, cycle_pause)]), +) async def run_test_read_errors(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -311,22 +322,6 @@ async def run_test_read_errors(dut, idle_inserter=None, backpressure_inserter=No 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_read, - run_test_read_errors, - ]: - - factory = TestFactory(test) - factory.add_option(("idle_inserter", "backpressure_inserter"), [(None, None), (cycle_pause, cycle_pause)]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/dma/tb/taxi_dma_if_pcie_us_wr/test_taxi_dma_if_pcie_us_wr.py b/src/dma/tb/taxi_dma_if_pcie_us_wr/test_taxi_dma_if_pcie_us_wr.py index 373b5d5..41c4ce6 100644 --- a/src/dma/tb/taxi_dma_if_pcie_us_wr/test_taxi_dma_if_pcie_us_wr.py +++ b/src/dma/tb/taxi_dma_if_pcie_us_wr/test_taxi_dma_if_pcie_us_wr.py @@ -19,7 +19,6 @@ import pytest import cocotb from cocotb.triggers import RisingEdge, FallingEdge, Timer -from cocotb.regression import TestFactory from cocotbext.axi import AxiStreamBus from cocotbext.pcie.core import RootComplex @@ -162,6 +161,14 @@ class TB(object): self.dma_ram.set_pause_generator(generator()) +def cycle_pause(): + return itertools.cycle([1, 1, 1, 0]) + + +@cocotb.test() +@cocotb.parametrize( + (("idle_inserter", "backpressure_inserter"), [(None, None), (cycle_pause, cycle_pause)]), +) async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None): tb = TB(dut) @@ -228,17 +235,6 @@ async def run_test_write(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: - - factory = TestFactory(run_test_write) - factory.add_option(("idle_inserter", "backpressure_inserter"), [(None, None), (cycle_pause, cycle_pause)]) - factory.generate_tests() - - # cocotb-test tests_dir = os.path.dirname(__file__) diff --git a/src/dma/tb/taxi_dma_psdpram/test_taxi_dma_psdpram.py b/src/dma/tb/taxi_dma_psdpram/test_taxi_dma_psdpram.py index 94e973b..b60462e 100644 --- a/src/dma/tb/taxi_dma_psdpram/test_taxi_dma_psdpram.py +++ b/src/dma/tb/taxi_dma_psdpram/test_taxi_dma_psdpram.py @@ -21,7 +21,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge, Timer -from cocotb.regression import TestFactory try: from dma_psdp_ram import PsdpRamMaster, PsdpRamBus @@ -67,7 +66,16 @@ class TB(object): await RisingEdge(self.dut.clk) -async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_inserter=None, size=None): +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) @@ -96,7 +104,12 @@ async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_ins await RisingEdge(dut.clk) -async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inserter=None, size=None): +@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) @@ -123,6 +136,11 @@ async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inse 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,20 +177,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]) - - -if getattr(cocotb, 'top', None) is not None: - - for test in [run_test_write, run_test_read, run_stress_test]: - - 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/dma/tb/taxi_dma_psdpram_async/test_taxi_dma_psdpram_async.py b/src/dma/tb/taxi_dma_psdpram_async/test_taxi_dma_psdpram_async.py index 27516a3..807c7c8 100644 --- a/src/dma/tb/taxi_dma_psdpram_async/test_taxi_dma_psdpram_async.py +++ b/src/dma/tb/taxi_dma_psdpram_async/test_taxi_dma_psdpram_async.py @@ -21,7 +21,6 @@ import pytest import cocotb from cocotb.clock import Clock from cocotb.triggers import RisingEdge, Timer -from cocotb.regression import TestFactory try: from dma_psdp_ram import PsdpRamMasterWrite, PsdpRamMasterRead, PsdpRamWriteBus, PsdpRamReadBus @@ -72,7 +71,16 @@ class TB(object): await RisingEdge(self.dut.clk_wr) -async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_inserter=None, size=None): +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) @@ -101,7 +109,12 @@ async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_ins await RisingEdge(dut.clk_wr) -async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inserter=None, size=None): +@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) @@ -128,6 +141,11 @@ async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inse await RisingEdge(dut.clk_wr) +@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) @@ -164,20 +182,6 @@ async def run_stress_test(dut, idle_inserter=None, backpressure_inserter=None): await RisingEdge(dut.clk_wr) -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_stress_test]: - - 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__)