test_check_axi_filtering: Mark payload done

Allow reading from indirect FIFO by marking
the payload done first.

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

Author: wkkuna

verification/cocotb/common/bus2csr.py

@@ -80,27 +80,33 @@ async def register_test_interfaces(self, fclk=500.0):
         await cocotb.start(setup_dut(self.clk, self.rst_n, (tclk, "ps")))
 
     async def read_csr(
-        self, addr: int, size: int = 4, timeout: int = 1, units: str = "us"
+        self, addr: int, size: int = 4, arid=None, timeout: int = 1, units: str = "us"
     ) -> List[int]:
         """Send a read request & await the response."""
         raise NotImplementedError
 
     async def write_csr(
-        self, addr: int, data: List[int], size: int = 4, timeout: int = 1, units: str = "us"
+        self,
+        addr: int,
+        data: List[int],
+        size: int = 4,
+        awid=None,
+        timeout: int = 1,
+        units: str = "us",
     ) -> None:
         """Send a write request & await transfer to finish."""
         raise NotImplementedError
 
-    async def write_csr_field(self, reg_addr, field, data) -> None:
+    async def write_csr_field(self, reg_addr, field, data, awid=None) -> None:
         """Read -> modify -> write CSR"""
-        value = bytes2int(await self.read_csr(reg_addr))
+        value = bytes2int(await self.read_csr(reg_addr, arid=awid))
         value = value & ~field.mask
         value = value | (data << field.low)
-        await self.write_csr(reg_addr, int2bytes(value))
+        await self.write_csr(reg_addr, int2bytes(value), awid=awid)
 
-    async def read_csr_field(self, reg_addr, field) -> int:
+    async def read_csr_field(self, reg_addr, field, arid=None) -> int:
         """Read -> modify -> write CSR"""
-        value = bytes2int(await self.read_csr(reg_addr))
+        value = bytes2int(await self.read_csr(reg_addr, arid=arid))
         value = value & field.mask
         value = value >> field.low
         return value
@@ -149,18 +155,28 @@ async def register_test_interfaces(self, *args, **kw):
         await super().register_test_interfaces(*args, **kw)
 
     async def read_csr(
-        self, addr: int, size: int = 4, timeout: int = 1, units: str = "us"
+        self, addr: int, size: int = 4, arid=None, timeout: int = 1, units: str = "us"
     ) -> List[int]:
         """Send a read request & await the response for 'timeout' in 'units'."""
+        if arid:
+            self.dut._log.debug(f"AHB doesn't support transaction IDs, ignoring arid={arid}")
         self.AHBManager.read(addr, size)
         await with_timeout(self.AHBManager.transfer_done(), timeout, units)
         read = self.AHBManager.get_rsp(addr, self.data_byte_width)
         return read
 
     async def write_csr(
-        self, addr: int, data: List[int], size: int = 4, timeout: int = 1, units: str = "us"
+        self,
+        addr: int,
+        data: List[int],
+        size: int = 4,
+        awid=None,
+        timeout: int = 1,
+        units: str = "us",
     ) -> None:
         """Send a write request & await transfer to finish for 'timeout' in 'units'."""
+        if awid:
+            self.dut._log.debug(f"AHB doesn't support transaction IDs, ignoring awid={awid}")
         data_len = len(data)
         # Extend bytes to size if there's less than that
         if data_len <= size:

verification/cocotb/top/lib_i3c_top/test_bypass.py

@@ -505,7 +505,7 @@ async def target_bus_traffic(addr, run_condition):
 
         for _ in range(10):  # Arbitrary number of repetitions
             payload_data = [random.randint(0, 2**32 - 1) for _ in range(fifo_size)]
-            delay_cycles = random.randint(0, 1000)
+            delay_cycles = random.randint(1, 1000)
             dut._log.info(f"Randomized delay is {delay_cycles} clock cycles")
 
             # Start I3C traffic
@@ -805,7 +805,7 @@ def csr_access_test_data(tb, rd_acc=Access.Priv, wr_acc=Access.Priv):
 
 
 @cocotb.test(skip=os.getenv("FrontendBusInterface") != "AXI")
-async def test_check_axi_filtering(dut):
+async def test_axi_filtering(dut):
     """
     Verifies AXI ID filtering in Secure Firmware Recovery registers access.
     """
@@ -851,6 +851,15 @@ async def test_check_axi_filtering(dut):
         payload_data = [random.randint(0, 2**32 - 1) for _ in range(2)]
         await write_to_indirect_fifo(tb, payload_data, format="dwords", awid=awid)
 
+        # Indicate that payload transfer is finished
+        for done in [1, 0]:
+            await tb.write_csr_field(
+                tb.reg_map.I3C_EC.SOCMGMTIF.REC_INTF_CFG.base_addr,
+                tb.reg_map.I3C_EC.SOCMGMTIF.REC_INTF_CFG.REC_PAYLOAD_DONE,
+                done,
+                awid=awid,
+            )
+
         resp = await tb.read_csr(addr, arid=get_axi_ids_seq(priv_ids, 1, rd_acc)[0])
         exp_rd = 0 if rd_acc == Access.Unpriv or wr_acc == Access.Unpriv else payload_data[0]
         compare_values(int2bytes(exp_rd), resp, addr)