xfcp: Update testbenches to use cocotb.parametrize

Signed-off-by: Alex Forencich <alex@alexforencich.com>

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

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

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

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

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

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