Add i3c_target_read_to_multiple_targets test

This test performs multiple random read operations to the device under
test. The transactions are interleaved with accesses to other random I3C
devices.

Signed-off-by: Maciej Dudek <[email protected]>

Author: mtdudek

third_party/cocotbext-i3c

@@ -14,6 +14,16 @@
 import cocotb
 from cocotb.triggers import ClockCycles, RisingEdge, Timer
 
+VALID_I3C_ADDRESSES = (
+    [i for i in range(0x03, 0x3E)]
+    + [i for i in range(0x3F, 0x5B)]
+    + [i for i in range(0x5C, 0x5E)]
+    + [i for i in range(0x5F, 0x6E)]
+    + [i for i in range(0x5F, 0x6E)]
+    + [i for i in range(0x6F, 0x76)]
+    + [i for i in range(0x77, 0x7A)]
+    + [0x7B, 0x7D]
+)
 TARGET_ADDRESS = 0x5A
 
 
@@ -32,7 +42,7 @@ async def test_setup(dut, fclk=100.0, fbus=12.5):
     """
 
     cocotb.log.setLevel(logging.INFO)
-    cocotb.start_soon(timeout_task(200))
+    cocotb.start_soon(timeout_task(500000))
 
     dut._log.info(f"fclk = {fclk:.3f} MHz")
     dut._log.info(f"fbus = {fbus:.3f} MHz")
@@ -338,6 +348,78 @@ def compare(expected, received, lnt=None):
             assert response.nack
 
 
+@cocotb.test()
+async def test_i3c_target_read_to_multiple_targets(dut):
+
+    # Setup
+    i3c_controller, i3c_target, tb = await test_setup(dut, fclk=400)
+
+    # Generates a randomized transfer and puts it into the TTI TX queue
+    async def make_transfer(min_len=1, max_len=16):
+
+        length = random.randint(min_len, max_len)
+        data = [random.randint(0, 255) for _ in range(length)]
+
+        dut._log.info(f"Enqueueing transfer of length {length}")
+
+        # Write data to TTI TX FIFO
+        for i in range((length + 3) // 4):
+            word = data[4 * i]
+            if 4 * i + 1 < length:
+                word |= data[4 * i + 1] << 8
+            if 4 * i + 2 < length:
+                word |= data[4 * i + 2] << 16
+            if 4 * i + 3 < length:
+                word |= data[4 * i + 3] << 24
+
+            await tb.write_csr(tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr, int2dword(word), 4)
+
+        # Write the TX descriptor
+        await tb.write_csr(tb.reg_map.I3C_EC.TTI.TX_DESC_QUEUE_PORT.base_addr, int2dword(length), 4)
+
+        return data
+
+    def compare(expected, received, lnt=None):
+        if lnt is None or lnt == len(expected):
+            sfx = ""
+        else:
+            sfx = " ([" + " ".join([f"{d:02X}" for d in expected[lnt:]]) + "] skipped)"
+            expected = expected[:lnt]
+
+        dut._log.info("Expected: [" + " ".join([f"{d:02X}" for d in expected]) + "]" + sfx)
+        dut._log.info("Received: [" + " ".join([f"{d:02X}" for d in received]) + "]")
+        assert expected == received
+
+    # issue 40 random read transactions
+    # randomly choose to inicialize the FIFO or not
+    # if FIFO is not initialized the transation should be NACKed
+    for _ in range(40):
+        num_transfers = random.randint(3, 10)
+        addresses = []
+        num_transfers_to_our_target = random.randint(1, num_transfers - 1)
+        for _ in range(num_transfers_to_our_target):
+            addresses.append(TARGET_ADDRESS)
+        while len(addresses) < num_transfers:
+            addresses.append(random.choice(VALID_I3C_ADDRESSES))
+        random.shuffle(addresses)
+        data_len_rsvd_stop_nack = []
+        for i, addr in enumerate(addresses):
+            send_rsvd = random.choice([True, False]) if i == 0 else False
+            stop = i == num_transfers - 1
+            if addr == TARGET_ADDRESS:
+                tx_data = await make_transfer()
+                data_len_rsvd_stop_nack.append((tx_data, len(tx_data), send_rsvd, stop, False))
+            else:
+                data_len_rsvd_stop_nack.append((None, random.randint(1, 16), send_rsvd, stop, True))
+
+        for address, (tx_data, length, rsvd, stop, nack) in zip(addresses, data_len_rsvd_stop_nack):
+            response = await i3c_controller.i3c_read(address, length, send_rsvd=rsvd, stop=stop)
+            assert nack == response.nack
+            if not nack:
+                rx_data = list(response.data)
+                compare(tx_data, rx_data)
+
+
 @cocotb.test()
 async def test_i3c_target_ibi(dut):
     """