Adapter cocotb tests and changes

pyproject.toml

@@ -36,7 +36,8 @@ test = [
     "cocotb==2.0.1",
     "cocotb-bus==0.3.0",
     "cocotbext-axi==0.1.26",
-    "cocotbext-umi==0.0.3"
+    "cocotbext-umi==0.0.3",
+    "cocotbext-apb==0.11.0"
 ]
 
 [tool.check-wheel-contents]

tests/adapters/tl2umi/env.py

@@ -0,0 +1,143 @@
+# Owns the driver, monitor, and scoreboard for TL to UMI adapter tests,
+# and provides common functionality for the tests.
+#
+# Uses cocotbext-umi UmiMemoryDevice as a pure-Python replacement for
+# the Verilog umi_memagent. The DUT is tl2umi directly (no wrapper needed).
+
+from cocotb.clock import Clock
+from cocotb.triggers import ClockCycles, Timer
+
+from cocotb_bus.scoreboard import Scoreboard
+
+from cocotbext.umi.drivers.sumi_driver import SumiDriver
+from cocotbext.umi.monitors.sumi_monitor import SumiMonitor
+from cocotbext.umi.models.umi_memory_device import UmiMemoryDevice
+
+from tl_driver import TLDriver
+from tl_monitor import TLMonitor, TLDResponse, TLDOpcode
+
+
+async def do_reset(reset, time_ns, active_level=False):
+    """Perform an 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")
+
+
+class TL2UMIEnv:
+    """Test environment for tl2umi adapter with Python UmiMemoryDevice backend"""
+
+    def __init__(self, dut, clk_period_ns=10):
+        self.dut = dut
+        self.clk_period_ns = clk_period_ns
+
+        # Extract parameters from DUT
+        self.cw = int(dut.CW.value)  # UMI command width (32)
+        self.aw = int(dut.AW.value)  # Address width (64)
+        self.dw = int(dut.DW.value)  # Data width (64)
+
+        self.data_size = self.dw // 8  # 8 bytes
+
+        self.expected_responses = []
+
+        self.clk = dut.clk
+        self.nreset = dut.nreset
+
+        self._build()
+
+    def _build(self):
+        dut = self.dut
+
+        # TileLink A-channel driver (sends requests)
+        self.tl_driver = TLDriver(
+            entity=dut,
+            name="tl_a",
+            clock=self.clk,
+            bus_separator="_",
+        )
+
+        # TileLink D-channel monitor (receives responses)
+        self.tl_monitor = TLMonitor(
+            entity=dut,
+            name="tl_d",
+            clock=self.clk,
+            bus_separator="_",
+        )
+
+        # UMI request monitor (observes requests from tl2umi)
+        self.sumi_req_monitor = SumiMonitor(
+            entity=dut, name="uhost_req", clock=self.clk
+        )
+
+        # Drive UMI request ready (accept all requests immediately)
+        dut.uhost_req_ready.value = 1
+
+        # UMI response driver (sends responses back to tl2umi)
+        self.sumi_resp_driver = SumiDriver(
+            entity=dut, name="uhost_resp", clock=self.clk
+        )
+
+        # Python UMI memory device (replaces Verilog umi_memagent)
+        self.mem_device = UmiMemoryDevice(
+            monitor=self.sumi_req_monitor,
+            driver=self.sumi_resp_driver,
+            log=dut._log
+        )
+
+        # Scoreboard for response checking
+        self.scoreboard = Scoreboard(dut, fail_immediately=True)
+        self.scoreboard.add_interface(
+            monitor=self.tl_monitor,
+            expected_output=self.expected_responses,
+        )
+
+    async def start(self):
+        """Start clocks and perform reset"""
+        Clock(self.clk, self.clk_period_ns, unit="ns").start()
+        await do_reset(self.nreset, self.clk_period_ns)
+
+        # Initialize DUT configuration signals
+        self.dut.srcaddr.value = 0xAE510000
+
+    async def wait_for_responses(self, max_cycles=1000):
+        """Wait for all expected responses to be received"""
+        cycles = 0
+        while self.expected_responses:
+            await ClockCycles(self.clk, 1)
+            cycles += 1
+            if cycles > max_cycles:
+                raise TimeoutError(
+                    f"Timeout waiting for responses "
+                    f"({len(self.expected_responses)} remaining)"
+                )
+
+
+def create_expected_read_response(address, size, data, source=0):
+    """Create expected TileLink D-channel read response"""
+    return TLDResponse(
+        opcode=TLDOpcode.AccessAckData,
+        param=0,
+        size=size,
+        source=source,
+        sink=0,
+        denied=False,
+        data=data,
+        corrupt=False,
+    )
+
+
+def create_expected_write_response(size, source=0):
+    """Create expected TileLink D-channel write response"""
+    return TLDResponse(
+        opcode=TLDOpcode.AccessAck,
+        param=0,
+        size=size,
+        source=source,
+        sink=0,
+        denied=False,
+        data=0,
+        corrupt=False,
+    )