Getting Started with Verifications Using cocotb/cocotbext-ndk

In this section, you will learn how to create a basic test for a flow/storage hardware component (such as a pipe, FIFO, etc.).

To get started, first create a cocotb folder in the directory where the tested component is located, and put all the scripts implemented in this tutorial into it.

Note

For automatic generation of a test template use the generate_test_template script in ndk-fpga/build/scripts/cocotb.

Creating a Test

As an example, a simple test of an MVB FIFOX component will be used. It can be found at ndk-fpga/comp/mvb_tools/storage/fifox/cocotb/cocotb_test.py.

# SPDX-License-Identifier: BSD-3-Clause
# Copyright (C) 2025 CESNET z. s. p. o.
# Author(s): Ondřej Schwarz <ondrejschwarz@cesnet.cz>
#            Daniel Kondys <kondys@cesnet.cz>


# importing required modules
import itertools

import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge, ClockCycles
from cocotbext.ofm.mvb.drivers import MVBDriver
from cocotbext.ofm.mvb.monitors import MVBMonitor
from cocotbext.ofm.ver.generators import random_integers
from cocotb_bus.drivers import BitDriver
from cocotb_bus.scoreboard import Scoreboard
from cocotbext.ofm.utils.throughput_probe import ThroughputProbe, ThroughputProbeMvbInterface
from cocotbext.ofm.base.generators import ItemRateLimiter
from cocotbext.ofm.mvb.transaction import MvbTrClassic


# definition of the class encapsulating components of the test
class testbench():
    # dut = device tree of the tested component
    def __init__(self, dut, debug=False):
        self.dut = dut

        # setting up the input driver and connecting it to signals beginning with "RX"
        self.stream_in = MVBDriver(dut, "RX", dut.CLK)

        # setting up the output monitor and connecting it to signals beginning with "TX"
        self.stream_out = MVBMonitor(dut, "TX", dut.CLK, tr_type=MvbTrClassic)

        # setting up driver of the DST_RDY so it randomly fluctuates between 0 and 1
        self.backpressure = BitDriver(dut.TX_DST_RDY, dut.CLK)

        # setting up the probe measuring throughput
        self.throughput_probe = ThroughputProbe(ThroughputProbeMvbInterface(self.stream_out), throughput_units="items")
        self.throughput_probe.set_log_period(10)
        self.throughput_probe.add_log_interval(0, None)

        # counter of sent transactions
        self.pkts_sent = 0

        # list of the transactions that are expected to be received
        self.expected_output = []

        # setting up a scoreboard which compares received transactions with the expected transactions
        self.scoreboard = Scoreboard(dut)

        # linking a monitor with its expected output
        self.scoreboard.add_interface(self.stream_out, self.expected_output)

        # setting up the logging level
        if debug:
            self.stream_in.log.setLevel(cocotb.logging.DEBUG)
            self.stream_out.log.setLevel(cocotb.logging.DEBUG)

    # method for adding transactions to the expected output
    def model(self, transaction):
        """Model the DUT based on the input transaction"""
        self.expected_output.append(transaction)
        self.pkts_sent += 1

    # method preforming a hardware reset
    async def reset(self):
        self.dut.RESET.value = 1
        await ClockCycles(self.dut.CLK, 10)
        self.dut.RESET.value = 0
        await RisingEdge(self.dut.CLK)


# defining a test - functions with "@cocotb.test()" decorator will be automatically found and run
@cocotb.test()
async def run_test(dut, pkt_count=10000):
    # start a clock generator
    cocotb.start_soon(Clock(dut.CLK, 5, units="ns").start())

    # initialization of the test bench
    tb = testbench(dut, debug=False)

    # change MVB driver's IdleGenerator to ItemRateLimiter
    # note: the RateLimiter's rate is affected by backpressure (DST_RDY).
    # Even though it takes into account cycles with DST_RDY=0, the desired rate might not be achievable.
    idle_gen_conf = dict(random_idles=True, max_idles=5, zero_idles_chance=50)
    tb.stream_in.set_idle_generator(ItemRateLimiter(rate_percentage=30, **idle_gen_conf))

    # running simulated reset
    await tb.reset()

    # starting the BitDriver (randomized DST_RDY)
    tb.backpressure.start((1, i % 5) for i in itertools.count())

    # dynamically getting the width of the data signal that will be set (useful if the width of the signal may change)
    data_width = tb.stream_in.item_widths["data"]

    # generating MVB items as random integers between the minimum and maximum unsigned value of the item
    for transaction in random_integers(0, 2**data_width-1, pkt_count):

        # logging the generated transaction
        cocotb.log.debug(f"generated transaction: {hex(transaction)}")

        # initializing MVB transaction object
        mvb_tr = MvbTrClassic()

        # setting data of MVB transaction to the generated integer
        mvb_tr.data = transaction

        # appending the transaction to tb.expected_output
        tb.model(mvb_tr)

        # passing the transaction to the driver which then writes it to the bus
        tb.stream_in.append(mvb_tr)

    last_num = 0

    # checking if all the expected packets have been received
    while (tb.stream_out.item_cnt < pkt_count):

        # logging number of received packets after every 1000 packets
        if (tb.stream_out.item_cnt // 1000) > last_num:
            last_num = tb.stream_out.item_cnt // 1000
            cocotb.log.info(f"Number of transactions processed: {tb.stream_out.item_cnt}/{pkt_count}")

        # if not all packets have been received yet, waiting 100 cycles so the simulation doesn't stop prematurely
        await ClockCycles(dut.CLK, 100)

    # logging values measured by throughput probe
    tb.throughput_probe.log_max_throughput()
    tb.throughput_probe.log_average_throughput()

    # displaying result of the test
    raise tb.scoreboard.result

This test can be used as a template for tests of basic flow and storage components, and can be easily adapted for most other verifications.

It consists of two basic parts: the testbench class and the test itself.

The testbench is the more reusable of the two and usually looks basically the same, so it can be copied and adapted. Its purpose is to initialize and encapsulate objects that drive the test. It sets up drivers, monitors, a scoreboard, expected outputs, and other optional objects, such as a bit driver for ready signals, adds probes, and so on. It also includes a simulated reset.

The second part is the test part. It can consist of one test (typical for simple components) or multiple tests (more common for larger designs, such as the whole firmware of a card). Every test must have the @cocotb.test() decorator and be async.

A test begins with the clock starting, testbench initialization, and a reset. After the reset, a bit driver is started to test the component’s reaction to backpressure (dst_rdy). Random data is then generated, which can either be done using random_transactions or random data that is then inserted into transaction objects (this is the case in the test above). The generated transaction is then passed to the model method of the testbench, which inserts it into the expected_output list. The generated transaction is also inserted into the driver’s send queue using the append method, from where it is then written onto the bus.

The data is then read from the bus by a monitor, which should pack it into a transaction of the same type as was modeled and pass it to the test’s scoreboard via a callback. The scoreboard pops the transaction from the front of the expected output queue that the monitor is connected to and compares this transaction with the transaction it received from the monitor. If they are not the same, a test failure is raised.

A waiting loop is implemented to ensure that the test doesn’t report scoreboard results prematurely before all the transactions have been received. Otherwise, the scoreboard may receive a different number of transactions than it expected, which will lead to an error.

After all the packets are received, tb.scoreboard.result is raised, and the test results are shown.

Running the Test

To successfully build and run the simulation, it’s necessary to implement a couple more files. Examples of these can again be found in the ndk-fpga/comp/mvb_tools/storage/fifox/cocotb/ folder.

First, it is necessary to implement a pyproject.toml with all test dependencies listed:

[project]
name = "cocotb-hash-table-simple-test"
version = "0.1.0"
dependencies = [
    "cocotbext-ofm @ ${NDK_FPGA_COCOTBEXT_OFM_URL}",
    "setuptools",
]

[build-system]
requires = ["pdm-backend"]
build-backend = "pdm.backend"

Use a special cocotb_test_sig.fdo file to define the signals that will be displayed in the simulator’s waveform.

# coctb_test_sig.fdo : Include file with signals
# Copyright (C) 2024 CESNET z. s. p. o.
# Author(s): Jakub Cabal <cabal@cesnet.cz>
#
# SPDX-License-Identifier: BSD-3-Clause

view wave
delete wave *

add_wave -group {ALL} -noupdate -hex /mvb_fifox/*

config wave -signalnamewidth 1

Finally, create a Makefile that will run the simulation:

# Makefile: Makefile to compile module
# Copyright (C) 2024 CESNET z. s. p. o.
# Author(s): Ondřej Schwarz <Ondrej.Schwarz@cesnet.cz>
#
# SPDX-License-Identifier: BSD-3-Clause

TOP_LEVEL_ENT=mvb_fifox
TARGET=cocotb

.PHONY: all
all: comp

include ../../../../../build/Makefile

Note

Don’t forget to adjust the values that are component-specific and the relative paths if needed.

You can run the simulation by creating a python virtual environment using make cocotb-venv, entering the created virtual environment, and running the Makefile:

make cocotb-venv
source venv-xxx/bin/activate
make