Updated cocotb tests and added them to CI

Author: RiceShelley

.github/workflows/ci.yml

@@ -78,7 +78,6 @@ jobs:
 
   eda_ci:
     name: Run EDA
-    if: false
     runs-on: ubuntu-latest
     container: ghcr.io/siliconcompiler/sc_runner:latest
     steps:
@@ -90,10 +89,6 @@ jobs:
           . .venv/bin/activate
 
           pip install --upgrade pip
-          pip install .[dev]
-
-          # change running directory
-          mkdir testrun
-          cd testrun
+          pip install -e .[dev]
 
-          pytest $GITHUB_WORKSPACE -m "eda"
+          pytest -n logical -m "eda" --verbose

lambdalib/ramlib/la_asyncfifo/testbench/la_asyncfifo.py

@@ -2,7 +2,7 @@
 from cocotb.triggers import RisingEdge
 from cocotb_bus.bus import Bus
 from cocotb.queue import Queue
-from cocotb.binary import BinaryValue
+from cocotb.types import LogicArray
 
 
 class LaAsyncFifoWrBus(Bus):
@@ -133,7 +133,7 @@ def pause(self):
     def resume(self):
         self._pause = False
 
-    async def read(self) -> BinaryValue:
+    async def read(self) -> LogicArray:
         return await self.queue.get()
 
     async def _run(self):

lambdalib/ramlib/la_asyncfifo/testbench/tb_la_asyncfifo.py

@@ -1,21 +1,17 @@
 import random
+import string
 from decimal import Decimal
 
-import siliconcompiler
-
 import cocotb
 from cocotb.clock import Clock
 from cocotb.triggers import ClockCycles, Timer, Combine
-from cocotb.regression import TestFactory
-from cocotb import utils
 
-from lambdalib import ramlib
 from lambdalib.utils._tb_common import (
     run_cocotb,
-    drive_reset,
+    do_reset,
     random_bool_generator
 )
-from lambdalib.ramlib.la_asyncfifo.testbench.la_asyncfifo import (
+from .la_asyncfifo import (
     LaAsyncFifoWrBus,
     LaAsyncFifoRdBus,
     LaAsyncFifoSource,
@@ -30,7 +26,13 @@ def bursty_en_gen(burst_len=20):
             yield en_state
 
 
-@cocotb.test()
+def random_decimal(max: int, min: int, decimal_places=2) -> Decimal:
+    prefix = str(random.randint(min, max))
+    suffix = ''.join(random.choice(string.digits) for _ in range(decimal_places))
+    return Decimal(prefix + "." + suffix)
+
+
+@cocotb.test(timeout_time=100, timeout_unit="ms")
 async def test_almost_full(dut):
 
     wr_clk_period_ns = 10.0
@@ -51,14 +53,14 @@ async def test_almost_full(dut):
 
     # Reset DUT
     await Combine(
-        cocotb.start_soon(drive_reset(dut.wr_nreset)),
-        cocotb.start_soon(drive_reset(dut.rd_nreset))
+        cocotb.start_soon(do_reset(dut.wr_nreset, time_ns=wr_clk_period_ns*4)),
+        cocotb.start_soon(do_reset(dut.rd_nreset, time_ns=rd_clk_period_ns*4))
     )
 
-    await cocotb.start(Clock(dut.wr_clk, wr_clk_period_ns, units="ns").start())
+    Clock(dut.wr_clk, wr_clk_period_ns, units="ns").start()
     # Randomize phase shift between clocks
     await Timer(wr_clk_period_ns * random.random(), "ns", round_mode="round")
-    await cocotb.start(Clock(dut.rd_clk, rd_clk_period_ns, units="ns").start())
+    Clock(dut.rd_clk, rd_clk_period_ns, units="ns").start()
 
     almost_full_level = int(dut.AFULLFINAL.value)
 
@@ -85,6 +87,22 @@ async def test_almost_full(dut):
     assert dut.wr_almost_full.value == 0
 
 
+# Generate sets of tests based on the different permutations of the possible arguments to fifo_test
+MAX_PERIOD_NS = 10
+MIN_PERIOD_NS = 1
+# Generate random clk period to test between min and max
+RAND_WR_CLK_PERIOD_NS, RAND_RD_CLK_PERIOD_NS = [
+    random_decimal(MAX_PERIOD_NS, MIN_PERIOD_NS) for _ in range(2)
+]
+
+
+@cocotb.test(timeout_time=100, timeout_unit="ms")
+@cocotb.parametrize(
+    wr_clk_period_ns=[MIN_PERIOD_NS, RAND_WR_CLK_PERIOD_NS, MAX_PERIOD_NS],
+    rd_clk_period_ns=[MIN_PERIOD_NS, RAND_RD_CLK_PERIOD_NS, MAX_PERIOD_NS],
+    wr_en_generator=[None, random_bool_generator, bursty_en_gen],
+    rd_en_generator=[None, random_bool_generator, bursty_en_gen]
+)
 async def fifo_rd_wr_test(
     dut,
     wr_clk_period_ns=10.0,
@@ -98,25 +116,27 @@ async def fifo_rd_wr_test(
         clock=dut.wr_clk,
         reset=dut.wr_nreset
     )
-    fifo_source.set_wr_en_generator(random_bool_generator())
+    if wr_en_generator:
+        fifo_source.set_wr_en_generator(wr_en_generator())
 
     fifo_sink = LaAsyncFifoSink(
         bus=LaAsyncFifoRdBus.from_prefix(dut, ""),
         clock=dut.rd_clk,
         reset=dut.rd_nreset
     )
-    fifo_sink.set_rd_en_generator(random_bool_generator())
+    if rd_en_generator:
+        fifo_sink.set_rd_en_generator(rd_en_generator())
 
     # Reset DUT
     await Combine(
-        cocotb.start_soon(drive_reset(dut.wr_nreset)),
-        cocotb.start_soon(drive_reset(dut.rd_nreset))
+        cocotb.start_soon(do_reset(dut.wr_nreset, time_ns=wr_clk_period_ns*4)),
+        cocotb.start_soon(do_reset(dut.rd_nreset, time_ns=rd_clk_period_ns*4))
     )
 
-    await cocotb.start(Clock(dut.wr_clk, wr_clk_period_ns, units="ns").start())
+    Clock(dut.wr_clk, wr_clk_period_ns, units="ns").start()
     # Randomize phase shift between clocks
-    await Timer(wr_clk_period_ns * random.random(), "ns", round_mode="round")
-    await cocotb.start(Clock(dut.rd_clk, rd_clk_period_ns, units="ns").start())
+    await Timer(wr_clk_period_ns * Decimal(random.random()), "ns", round_mode="round")
+    Clock(dut.rd_clk, rd_clk_period_ns, units="ns").start()
 
     await ClockCycles(dut.wr_clk, 3)
 
@@ -133,55 +153,25 @@ async def fifo_rd_wr_test(
     await ClockCycles(dut.wr_clk, 10)
 
 
-# Generate sets of tests based on the different permutations of the possible arguments to fifo_test
-MAX_PERIOD_NS = 10.0
-MIN_PERIOD_NS = 1.0
-# Generate random clk period to test between min and max
-RAND_WR_CLK_PERIOD_NS, RAND_RD_CLK_PERIOD_NS = [utils.get_time_from_sim_steps(
-    # Time step must be even for cocotb clock driver
-    steps=utils.get_sim_steps(
-        time=Decimal(MIN_PERIOD_NS) + (
-            Decimal(MAX_PERIOD_NS - MIN_PERIOD_NS)
-            * Decimal(random.random()).quantize(Decimal("0.00"))
-        ),
-        units="ns",
-        round_mode="round"
-    ) & ~1,
-    units="ns"
-) for _ in range(0, 2)]
-
-# Factory to automatically generate a set of tests based on the different permutations
-# of the provided test arguments
-tf = TestFactory(fifo_rd_wr_test)
-tf.add_option('wr_clk_period_ns', [MIN_PERIOD_NS, RAND_WR_CLK_PERIOD_NS, MAX_PERIOD_NS])
-tf.add_option('rd_clk_period_ns', [MIN_PERIOD_NS, RAND_RD_CLK_PERIOD_NS, MAX_PERIOD_NS])
-tf.add_option('wr_en_generator', [None, random_bool_generator, bursty_en_gen])
-tf.add_option('rd_en_generator', [None, random_bool_generator, bursty_en_gen])
-tf.generate_tests()
-
-
-def test_la_asyncfifo():
-    chip = siliconcompiler.Chip("la_asyncfifo")
-
-    # TODO: should not be needed?
-    chip.input("ramlib/la_asyncfifo/rtl/la_asyncfifo.v", package='lambdalib')
-    chip.use(ramlib)
+def load_cocotb_test():
+    from siliconcompiler import Sim
+    from lambdalib.ramlib import Asyncfifo
+
+    project = Sim(Asyncfifo())
+    project.add_fileset("rtl")
 
     for depth in [2, 4, 8]:
-        test_module_name = "lambdalib.ramlib.tests.tb_la_asyncfifo"
+        test_module_name = __name__
         test_name = f"{test_module_name}_depth_{depth}"
         tests_failed = run_cocotb(
-            chip=chip,
+            project=project,
             test_module_name=test_module_name,
             timescale=("1ns", "1ps"),
             parameters={
                 "DW": 32,
                 "DEPTH": depth
             },
-            output_dir_name=test_name
+            output_dir_name=test_name,
+            waves=False
         )
         assert (tests_failed == 0), f"Error test {test_name} failed!"
-
-
-if __name__ == "__main__":
-    test_la_asyncfifo()

lambdalib/utils/_tb_common.py

@@ -1,76 +1,115 @@
 import os
-
 import random
 from pathlib import Path
+from typing import List, Tuple, Optional, Mapping, Union
+
+from siliconcompiler import Sim
 
 from cocotb.triggers import Timer
-from cocotb.runner import get_runner, get_results
+from cocotb.handle import SimHandleBase
+
+from cocotb_tools.runner import get_runner
+from cocotb_tools.check_results import get_results
 
-import siliconcompiler
-from siliconcompiler.tools.slang import elaborate
+
+async def do_reset(
+        reset: SimHandleBase,
+        time_ns: int,
+        active_level: bool = False):
+    """Perform a async reset"""
+    reset.value = not active_level
+    await Timer(1, unit="step")
+    reset.value = active_level
+    await Timer(time_ns, "ns")
+    reset.value = not active_level
+    await Timer(1, unit="step")
 
 
 def run_cocotb(
-        chip: siliconcompiler.Chip,
-        test_module_name,
-        output_dir_name=None,
-        simulator_name="icarus",
-        timescale=None,
-        parameters=None):
-
-    # Use surelog to pickle Verilog sources
-    flow = "cocotb_flow"
-    chip.node(flow, "import", elaborate)
-    chip.set("option", "flow", flow)
-    assert chip.run()
-
-    pickled_verilog = chip.find_result("v", "import")
-    assert pickled_verilog, "Could not locate pickled verilog"
+        project: Sim,
+        test_module_name: str,
+        output_dir_name: Optional[str] = None,
+        simulator_name: str = "icarus",
+        build_args: Optional[List] = None,
+        timescale: Optional[Tuple[str, str]] = None,
+        parameters: Optional[Mapping[str, object]] = None,
+        seed: Optional[Union[str, int]] = None,
+        waves: bool = True):
+    """Launch cocotb given a SC Project"""
+
+    if parameters is None:
+        parameters = {}
+
+    if build_args is None:
+        build_args = []
 
     if output_dir_name is None:
         output_dir_name = test_module_name
 
-    top_level_dir = os.getcwd()
-    build_dir = Path(chip.getbuilddir()) / output_dir_name
+    pytest_current_test = os.getenv("PYTEST_CURRENT_TEST", None)
+
+    rootpath = Path(__file__).resolve().parent.parent
+    top_level_dir = rootpath
+    build_dir = rootpath / "build" / output_dir_name
     test_dir = None
 
     results_xml = None
-    # Need to check if we are running inside of pytest. See link below.
-    # https://github.com/cocotb/cocotb/blob/d883ce914063c3601455d10a40f459fffa22d8f2/cocotb/runner.py#L313
-    if not os.getenv("PYTEST_CURRENT_TEST", None):
+    if not pytest_current_test:
         results_xml = build_dir / "results.xml"
         test_dir = top_level_dir
 
+    # Get top level module name
+    top_lvl_module_name = None
+    main_filesets = project.get("option", "fileset")
+    if main_filesets and len(main_filesets) != 0:
+        main_fileset = main_filesets[0]
+        top_lvl_module_name = project.design.get_topmodule(
+            fileset=main_fileset
+        )
+
+    filesets = project.get_filesets()
+    idirs = []
+    defines = []
+    for lib, fileset in filesets:
+        idirs.extend(lib.find_files("fileset", fileset, "idir"))
+        defines.extend(lib.get("fileset", fileset, "define"))
+
+    sources = []
+    for lib, fileset in filesets:
+        for value in lib.get_file(fileset=fileset, filetype="systemverilog"):
+            sources.append(value)
+    for lib, fileset in filesets:
+        for value in lib.get_file(fileset=fileset, filetype="verilog"):
+            sources.append(value)
+
     # Build HDL in chosen simulator
     runner = get_runner(simulator_name)
     runner.build(
-        sources=[pickled_verilog],
-        hdl_toplevel=chip.design,
-        waves=True,
+        sources=sources,
+        includes=idirs,
+        hdl_toplevel=top_lvl_module_name,
+        build_args=build_args,
+        waves=waves,
         timescale=timescale,
         build_dir=build_dir,
         parameters=parameters
     )
+
     # Run test
     _, tests_failed = get_results(runner.test(
-        hdl_toplevel=chip.top(),
+        hdl_toplevel=top_lvl_module_name,
         test_module=test_module_name,
         test_dir=test_dir,
+        test_args=build_args,
         results_xml=results_xml,
         build_dir=build_dir,
-        timescale=timescale,
-        waves=True
+        seed=seed,
+        waves=waves
     ))
 
     return tests_failed
 
 
-async def drive_reset(reset, reset_time_in_ns=100, active_level=False):
-    reset.value = 1 if active_level else 0
-    await Timer(reset_time_in_ns, units="ns")
-    reset.value = 0 if active_level else 1
-
-
 def random_bool_generator():
     while True:
         yield (random.randint(0, 1) == 1)

pyproject.toml

@@ -31,8 +31,9 @@ version = {attr = "lambdalib.__version__"}
 dev = [
     "flake8 == 7.3.0",
     "pytest == 8.4.2",
+    "pytest-xdist==3.8.0",
     "pytest-timeout == 2.4.0",
-    "cocotb == 2.0.0",
+    "cocotb @ git+https://github.com/cocotb/cocotb.git@770d72214362cb366097267873839f50c69e417e",
     "cocotb-bus @ git+https://github.com/cocotb/cocotb-bus.git@c53c6024391fa02263969b02a0e0de5a22974cfc"
 ]
 
@@ -43,4 +44,4 @@ markers = [
 testpaths = [
     "tests"
 ]
-timeout = "60"
+timeout = "300"

tests/test_cocotb_runner.py

@@ -0,0 +1,7 @@
+import pytest
+from lambdalib.ramlib.la_asyncfifo.testbench import tb_la_asyncfifo
+
+
+@pytest.mark.eda
+def test_la_asyncfifo():
+    tb_la_asyncfifo.load_cocotb_test()