bslathi19/taxi
1
0
Fork 0
mirror of https://github.com/fpganinja/taxi.git synced 2026-08-29 08:05:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

dma: Update testbenches to use cocotb.parametrize

Browse Source 
Signed-off-by: Alex Forencich <alex@alexforencich.com>
This commit is contained in:
Alex Forencich
2026-08-15 13:13:14 -07:00
parent 1cf84f107d
commit ca26a48a28
12 changed files with 169 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/dma/tb/taxi_axi_cdma/test_taxi_axi_cdma.py
View File

@@ -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__)

33
src/dma/tb/taxi_axi_dma/test_taxi_axi_dma.py
View File

@@ -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__)

24
src/dma/tb/taxi_dma_client_axis_sink/test_taxi_dma_client_axis_sink.py
View File

@@ -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__)

24
src/dma/tb/taxi_dma_client_axis_source/test_taxi_dma_client_axis_source.py
View File

@@ -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__)

34
src/dma/tb/taxi_dma_if_axi/test_taxi_dma_if_axi.py
View File

@@ -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__)

22
src/dma/tb/taxi_dma_if_axi_rd/test_taxi_dma_if_axi_rd.py
View File

@@ -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__)

29
src/dma/tb/taxi_dma_if_axi_wr/test_taxi_dma_if_axi_wr.py
View File

@@ -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__)

37
src/dma/tb/taxi_dma_if_pcie_us/test_taxi_dma_if_pcie_us.py
View File

@@ -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__)

29
src/dma/tb/taxi_dma_if_pcie_us_rd/test_taxi_dma_if_pcie_us_rd.py
View File

@@ -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__)

20
src/dma/tb/taxi_dma_if_pcie_us_wr/test_taxi_dma_if_pcie_us_wr.py
View File

@@ -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__)

38
src/dma/tb/taxi_dma_psdpram/test_taxi_dma_psdpram.py
View File

@@ -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__)

38
src/dma/tb/taxi_dma_psdpram_async/test_taxi_dma_psdpram_async.py
View File

@@ -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__)