bslathi19/taxi
1
0
Fork 0
mirror of https://github.com/fpganinja/taxi.git synced 2026-08-16 04:11:19 -07:00
Code Issues Packages Projects Releases Wiki Activity

eth: Consolidate example design testbench

Browse Source 
Signed-off-by: Alex Forencich <alex@alexforencich.com>
This commit is contained in:
Alex Forencich
2026-08-09 14:34:18 -07:00
parent ecc1558ab7
commit ec6723e90d
1 changed files with 6 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
src/eth/example/RK_XCKU5P_F/fpga/tb/fpga_core/test_fpga_core.py
View File

@@ -20,7 +20,7 @@ import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge, Timer, Combine
from cocotbext.eth import GmiiFrame, RgmiiPhy
from cocotbext.eth import RgmiiPhy
from cocotbext.eth import XgmiiFrame
from cocotbext.uart import UartSource, UartSink
@@ -132,52 +132,6 @@ class TB:
async def mac_test(tb, source, sink):
tb.log.info("Test MAC")
tb.log.info("Multiple small packets")
count = 64
pkts = [bytearray([(x+k) % 256 for x in range(60)]) for k in range(count)]
for p in pkts:
await source.send(GmiiFrame.from_payload(p))
for k in range(count):
rx_frame = await sink.recv()
tb.log.info("RX frame: %s", rx_frame)
assert rx_frame.get_payload() == pkts[k]
assert rx_frame.check_fcs()
assert rx_frame.error is None
tb.log.info("Multiple large packets")
count = 32
pkts = [bytearray([(x+k) % 256 for x in range(1514)]) for k in range(count)]
for p in pkts:
await source.send(GmiiFrame.from_payload(p))
for k in range(count):
rx_frame = await sink.recv()
tb.log.info("RX frame: %s", rx_frame)
assert rx_frame.get_payload() == pkts[k]
assert rx_frame.check_fcs()
assert rx_frame.error is None
tb.log.info("MAC test done")
async def mac_test_25g(tb, source, sink):
tb.log.info("Test MAC")
tb.log.info("Wait for block lock")
for k in range(1200):
await RisingEdge(tb.dut.clk)
sink.clear()
tb.log.info("Multiple small packets")
@@ -226,6 +180,10 @@ async def run_test(dut):
tests = []
tb.log.info("Wait for block lock")
for k in range(1200):
await RisingEdge(tb.dut.clk)
tb.log.info("Start BASE-T MAC loopback test")
tests.append(cocotb.start_soon(mac_test(tb, tb.baset_phy.rx, tb.baset_phy.tx)))
@@ -233,7 +191,7 @@ async def run_test(dut):
for k in range(len(tb.qsfp_sources)):
tb.log.info("Start QSFP %d MAC loopback test", k)
tests.append(cocotb.start_soon(mac_test_25g(tb, tb.qsfp_sources[k], tb.qsfp_sinks[k])))
tests.append(cocotb.start_soon(mac_test(tb, tb.qsfp_sources[k], tb.qsfp_sinks[k])))
await Combine(*tests)