Cleanup AXI filtering-related verification

Move helpers common for both 'block' and 'top' tests to 'common'.
Lint AXI ID filtering tests.

Signed-off-by: Wiktoria Kuna <[email protected]>

Author: wkkuna

verification/cocotb/block/axi_adapter_id_filter/axi_utils.py

@@ -1,12 +1,8 @@
 # SPDX-License-Identifier: Apache-2.0
 import logging
-import random
-from enum import IntEnum
 
 from bus2csr import get_frontend_bus_if
 from cocotb_helpers import reset_n
-from cocotbext.axi import AxiResp
-from utils import rand_bits
 
 import cocotb
 from cocotb.triggers import ClockCycles, RisingEdge, Timer
@@ -34,93 +30,6 @@ async def error_monitor(dut):
         await RisingEdge(dut.aclk)
 
 
-def _report_axi_response(got, expected, is_read=False):
-    op = "read" if is_read else "write"
-    name = ["OKAY", "EXOKAY", "SLVERR", "DECERR"]
-    return (
-        f"{hex(expected)} != {hex(got)}."
-        f" Anticipated {op} response: {name[expected]} got: {name[got]}."
-    )
-
-
-async def _wait(dut, cond):
-    while not cond():
-        await RisingEdge(dut.aclk)
-
-
-async def axi_read_monitor(dut):
-    """
-    Ensures the AXI read response is set appropriately to
-    current filtering configuration and transaction ID.
-    """
-    await RisingEdge(dut.areset_n)
-
-    while True:
-        await _wait(dut, lambda: dut.arvalid.value and dut.arready.value)
-        priv_ids = dut.priv_ids_i.value
-        filter_off = dut.disable_id_filtering_i.value
-
-        await _wait(dut, lambda: dut.rvalid.value and dut.rready.value)
-        rid = dut.rid.value
-        rresp = dut.rresp.value
-        if filter_off or rid in priv_ids:
-            assert rresp == AxiResp.OKAY, _report_axi_response(rresp, AxiResp.OKAY, True)
-        else:
-            assert rresp == AxiResp.SLVERR, _report_axi_response(rresp, AxiResp.SLVERR, True)
-
-        await RisingEdge(dut.aclk)
-
-
-async def axi_write_monitor(dut):
-    """
-    Ensures the AXI write response is set appropriately to
-    current filtering configuration and transaction ID.
-    """
-    await RisingEdge(dut.areset_n)
-
-    while True:
-        await _wait(dut, lambda: dut.awvalid.value and dut.awready.value)
-        priv_ids = dut.priv_ids_i.value
-        filter_off = dut.disable_id_filtering_i.value
-
-        await _wait(dut, lambda: dut.bvalid.value and dut.bready.value)
-        bid = dut.bid.value
-        bresp = dut.bresp.value
-
-        if filter_off or bid in priv_ids:
-            assert bresp == AxiResp.OKAY, _report_axi_response(bresp, AxiResp.OKAY)
-        else:
-            assert bresp == AxiResp.SLVERR, _report_axi_response(bresp, AxiResp.SLVERR)
-
-        await RisingEdge(dut.aclk)
-
-
-class Access(IntEnum):
-    Priv = 0
-    Unpriv = 1
-    Mixed = 2
-
-
-def draw_ids(id_width=AxiIdWidth, num_priv_ids=NumPrivIds):
-    return [rand_bits(id_width) for _ in range(num_priv_ids)]
-
-
-def get_ids(priv_ids, count, priv=False, id_width=AxiIdWidth):
-    id_space = range(0, (1 << id_width))
-    unpriv = [x for x in id_space if x not in priv_ids]
-
-    is_priv = priv == Access.Priv
-    out = []
-    for _ in range(count):
-        if priv == Access.Mixed:
-            is_priv = random.randint(0, 1)
-
-        id_scope = priv_ids if is_priv else unpriv
-        out.append(random.choice(id_scope))
-
-    return out
-
-
 async def initialize_dut(dut, disable_id_filtering=False, priv_ids=None, timeout=50):
     """
     Common test initialization routine which sets up environment and starts a timeout coroutine
@@ -129,11 +38,15 @@ async def initialize_dut(dut, disable_id_filtering=False, priv_ids=None, timeout
 
     cocotb.log.setLevel(logging.DEBUG)
 
+    tb = get_frontend_bus_if()(dut)
+    tb.log = dut._log
+    await tb.register_test_interfaces()
+
     # Start the background timeout task
     cocotb.start_soon(timeout_task(timeout))
     cocotb.start_soon(error_monitor(dut))
-    cocotb.start_soon(axi_read_monitor(dut))
-    cocotb.start_soon(axi_write_monitor(dut))
+    cocotb.start_soon(tb.read_access_monitor())
+    cocotb.start_soon(tb.write_access_monitor())
 
     # Initialize inputs
     dut.araddr.value = 0
@@ -166,10 +79,6 @@ async def initialize_dut(dut, disable_id_filtering=False, priv_ids=None, timeout
     if hasattr(dut, "priv_ids_i"):
         dut.priv_ids_i.value = priv_ids
 
-    # Configure testbench
-    tb = get_frontend_bus_if()(dut)
-    tb.log = dut._log
-    await tb.register_test_interfaces()
     await ClockCycles(tb.clk, 20)
     await reset_n(tb.clk, tb.rst_n, cycles=5)
 

verification/cocotb/block/axi_adapter_id_filter/test_bus_stress.py

@@ -1,18 +1,17 @@
 # SPDX-License-Identifier: Apache-2.0
 import random
 
-from axi_utils import Access, draw_ids, get_ids, initialize_dut
+from axi_utils import initialize_dut
 from bus2csr import dword2int, int2bytes, int2dword
 from cocotbext.axi.constants import AxiBurstType
+from utils import Access, draw_axi_priv_ids, get_axi_ids_seq
 
 import cocotb
 from cocotb.triggers import ClockCycles, Combine, RisingEdge, with_timeout
 
 
 async def initialize(dut, filter_off=False, priv_ids=None, timeout=50):
-    tb = await initialize_dut(
-        dut, disable_id_filtering=filter_off, priv_ids=priv_ids, timeout=timeout
-    )
+    tb = await initialize_dut(dut, filter_off, priv_ids, timeout)
 
     # Generate test data
     data_len = random.randint(10, 64)
@@ -37,10 +36,10 @@ def verify_data(write_data, awids, read_data, arids, disable_id_filtering=False,
 
 
 async def collision_with_write(dut, filter_off=False, awid_priv=Access.Priv, arid_priv=Access.Priv):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off, priv_ids)
-    awids = get_ids(priv_ids, data_len, awid_priv)
-    arids = get_ids(priv_ids, data_len, arid_priv)
+    awids = get_axi_ids_seq(priv_ids, data_len, awid_priv)
+    arids = get_axi_ids_seq(priv_ids, data_len, arid_priv)
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
@@ -96,10 +95,10 @@ async def test_collision_with_write_id_filter_on_mixed(dut):
 
 
 async def collision_with_read(dut, filter_off=False, awid_priv=True, arid_priv=True):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off, priv_ids)
-    awids = get_ids(priv_ids, data_len, awid_priv)
-    arids = get_ids(priv_ids, data_len, arid_priv)
+    awids = get_axi_ids_seq(priv_ids, data_len, awid_priv)
+    arids = get_axi_ids_seq(priv_ids, data_len, arid_priv)
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
@@ -162,10 +161,10 @@ async def test_collision_with_read_id_filter_on_mixed(dut):
 
 
 async def write_read_burst(dut, filter_off=False, awid_priv=True, arid_priv=True):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off, priv_ids)
-    awids = get_ids(priv_ids, 1, awid_priv)
-    arids = get_ids(priv_ids, 1, arid_priv)
+    awids = get_axi_ids_seq(priv_ids, 1, awid_priv)
+    arids = get_axi_ids_seq(priv_ids, 1, arid_priv)
     awid, arid = awids[0], arids[0]
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
@@ -198,10 +197,10 @@ async def test_write_read_burst_id_filter_on_non_priv(dut):
 
 
 async def write_burst_collision_with_read(dut, filter_off=False, awid_priv=True, arid_priv=True):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off, priv_ids)
-    awids = get_ids(priv_ids, 1, awid_priv)
-    arids = get_ids(priv_ids, 1, arid_priv)
+    awids = get_axi_ids_seq(priv_ids, 1, awid_priv)
+    arids = get_axi_ids_seq(priv_ids, 1, arid_priv)
     awid, arid = awids[0], arids[0]
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
@@ -249,10 +248,10 @@ async def test_write_burst_collision_with_read_id_filter_non_priv(dut):
 
 
 async def read_burst_collision_with_write(dut, filter_off=False, awid_priv=True, arid_priv=True):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off, priv_ids)
-    awids = get_ids(priv_ids, 1, awid_priv)
-    arids = get_ids(priv_ids, 1, arid_priv)
+    awids = get_axi_ids_seq(priv_ids, 1, awid_priv)
+    arids = get_axi_ids_seq(priv_ids, 1, arid_priv)
     awid, arid = awids[0], arids[0]
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
@@ -311,10 +310,10 @@ async def test_read_burst_collision_with_write_id_filter_on_non_priv(dut):
 
 @cocotb.test()
 async def test_collision_with_write_mixed_priv(dut):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off=False, priv_ids=priv_ids)
-    awids = get_ids(priv_ids, data_len, Access.Mixed)
-    arids = get_ids(priv_ids, data_len, Access.Mixed)
+    awids = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
+    arids = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
@@ -348,10 +347,10 @@ async def reader():
 
 @cocotb.test()
 async def test_collision_with_read_mixed_priv(dut):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, filter_off=False, priv_ids=priv_ids)
-    awids = get_ids(priv_ids, data_len, Access.Mixed)
-    arids = get_ids(priv_ids, data_len, Access.Mixed)
+    awids = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
+    arids = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
 
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr

verification/cocotb/block/axi_adapter_id_filter/test_runtime_id_change.py

@@ -1,9 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 import random
 
-from axi_utils import Access, _wait, draw_ids, get_ids, initialize_dut
+from axi_utils import initialize_dut
 from bus2csr import dword2int, int2bytes
 from cocotbext.axi.constants import AxiBurstType
+from utils import Access, draw_axi_priv_ids, get_axi_ids_seq
 
 import cocotb
 from cocotb.triggers import Combine, RisingEdge
@@ -23,7 +24,8 @@ async def initialize(dut, filter_off=False, priv_ids=None, timeout=50):
 async def id_filter_disable_toggle(dut, cond):
     while True:
         filter_off = int(dut.disable_id_filtering_i.value)
-        await _wait(dut, cond)
+        while not cond:
+            await RisingEdge(dut.aclk)
         dut.disable_id_filtering_i.value = not filter_off
         await RisingEdge(dut.aclk)
 
@@ -32,7 +34,8 @@ async def priv_ids_swapper(dut, cond, new_priv_ids):
     assert isinstance(new_priv_ids, list)
     i, n = 0, len(new_priv_ids)
     while i < n:
-        await _wait(dut, cond)
+        while not cond:
+            await RisingEdge(dut.aclk)
         dut.priv_ids_i.value = new_priv_ids[i]
         i += 1
         await RisingEdge(dut.aclk)
@@ -64,7 +67,7 @@ async def reader(tb, addr, count, tids, return_data):
 
 
 async def toggle_filtering(dut, cond):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, True, priv_ids)
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
@@ -74,8 +77,8 @@ async def toggle_filtering(dut, cond):
 
     cocotb.start_soon(id_filter_disable_toggle(dut, cond))
 
-    awid = get_ids(priv_ids, data_len, Access.Mixed)
-    arid = get_ids(priv_ids, data_len, Access.Mixed)
+    awid = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
+    arid = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
 
     received_data = []
     w = cocotb.start_soon(writer(tb, waddr, test_data, awid))
@@ -95,19 +98,19 @@ async def test_toggle_filtering_mid_write(dut):
 
 
 async def swap_priv_ids(dut, cond):
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, True, priv_ids)
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
 
     # Fill fifo halfway to avoid reads when empty
     await tb.axi_m.write_dwords(waddr, range(64), burst=AxiBurstType.FIXED, awid=priv_ids[0])
 
-    priv_ids_seq = [draw_ids() for _ in range(data_len)]
+    priv_ids_seq = [draw_axi_priv_ids() for _ in range(data_len)]
     cocotb.start_soon(priv_ids_swapper(dut, cond, priv_ids_seq))
 
-    awid = get_ids(priv_ids, data_len, Access.Mixed)
-    arid = get_ids(priv_ids, data_len, Access.Mixed)
+    awid = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
+    arid = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
 
     received_data = []
     w = cocotb.start_soon(writer(tb, waddr, test_data, awid))
@@ -138,10 +141,10 @@ async def disable_random():
     async def priv_id_swap_random():
         while True:
             if abs(random.random()) < 0.2:
-                dut.priv_ids_i.value = draw_ids()
+                dut.priv_ids_i.value = draw_axi_priv_ids()
             await RisingEdge(dut.aclk)
 
-    priv_ids = draw_ids()
+    priv_ids = draw_axi_priv_ids()
     tb, data_len, test_data = await initialize(dut, True, priv_ids)
     waddr = tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr
     raddr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
@@ -152,8 +155,8 @@ async def priv_id_swap_random():
     cocotb.start_soon(disable_random())
     cocotb.start_soon(priv_id_swap_random())
 
-    awid = get_ids(priv_ids, data_len, Access.Mixed)
-    arid = get_ids(priv_ids, data_len, Access.Mixed)
+    awid = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
+    arid = get_axi_ids_seq(priv_ids, data_len, Access.Mixed)
 
     received_data = []
     w = cocotb.start_soon(writer(tb, waddr, test_data, awid))