From b073fc8efbda1faf771eccebcbed8395730aac74 Mon Sep 17 00:00:00 2001 From: Alex Forencich Date: Mon, 7 Apr 2025 13:25:53 -0700 Subject: [PATCH] eth: Check stats outputs in AXI stream GMII TX module testbench Signed-off-by: Alex Forencich --- .../test_taxi_axis_gmii_tx.py | 94 +++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/tb/eth/taxi_axis_gmii_tx/test_taxi_axis_gmii_tx.py b/tb/eth/taxi_axis_gmii_tx/test_taxi_axis_gmii_tx.py index b926ae3..583c612 100644 --- a/tb/eth/taxi_axis_gmii_tx/test_taxi_axis_gmii_tx.py +++ b/tb/eth/taxi_axis_gmii_tx/test_taxi_axis_gmii_tx.py @@ -50,6 +50,22 @@ class TB: dut.cfg_tx_ifg.setimmediatevalue(0) dut.cfg_tx_enable.setimmediatevalue(0) + self.stats = {} + self.stats["tx_start_packet"] = 0 + self.stats["stat_tx_byte"] = 0 + self.stats["stat_tx_pkt_len"] = 0 + self.stats["stat_tx_pkt_ucast"] = 0 + self.stats["stat_tx_pkt_mcast"] = 0 + self.stats["stat_tx_pkt_bcast"] = 0 + self.stats["stat_tx_pkt_vlan"] = 0 + self.stats["stat_tx_pkt_good"] = 0 + self.stats["stat_tx_pkt_bad"] = 0 + self.stats["stat_tx_err_oversize"] = 0 + self.stats["stat_tx_err_user"] = 0 + self.stats["stat_tx_err_underflow"] = 0 + + cocotb.start_soon(self._run_stats_counters()) + async def reset(self): self.dut.rst.setimmediatevalue(0) await RisingEdge(self.dut.clk) @@ -61,6 +77,15 @@ class TB: await RisingEdge(self.dut.clk) await RisingEdge(self.dut.clk) + for stat in self.stats: + self.stats[stat] = 0 + + async def _run_stats_counters(self): + while True: + await RisingEdge(self.dut.clk) + for stat in self.stats: + self.stats[stat] += int(getattr(self.dut, stat).value) + def set_enable_generator(self, generator=None): if self._enable_cr is not None: self._enable_cr.kill() @@ -96,8 +121,13 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_ test_frames = [payload_data(x) for x in payload_lengths()] + total_bytes = 0 + total_pkts = 0 + for test_data in test_frames: await tb.source.send(AxiStreamFrame(test_data, tid=0, tuser=0)) + total_bytes += max(len(test_data), 60)+4 + total_pkts += 1 for test_data in test_frames: rx_frame = await tb.sink.recv() @@ -118,6 +148,22 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_ assert tb.sink.empty() + for stat, val in tb.stats.items(): + tb.log.info("%s: %d", stat, val) + + assert tb.stats["tx_start_packet"] == total_pkts + assert tb.stats["stat_tx_byte"] == total_bytes + assert tb.stats["stat_tx_pkt_len"] == total_bytes + assert tb.stats["stat_tx_pkt_ucast"] == total_pkts + assert tb.stats["stat_tx_pkt_mcast"] == 0 + assert tb.stats["stat_tx_pkt_bcast"] == 0 + assert tb.stats["stat_tx_pkt_vlan"] == 0 + assert tb.stats["stat_tx_pkt_good"] == total_pkts + assert tb.stats["stat_tx_pkt_bad"] == 0 + assert tb.stats["stat_tx_err_oversize"] == 0 + assert tb.stats["stat_tx_err_user"] == 0 + assert tb.stats["stat_tx_err_underflow"] == 0 + await RisingEdge(dut.clk) await RisingEdge(dut.clk) @@ -170,6 +216,22 @@ async def run_test_underrun(dut, ifg=12, enable_gen=None, mii_sel=False): assert tb.sink.empty() + for stat, val in tb.stats.items(): + tb.log.info("%s: %d", stat, val) + + assert tb.stats["tx_start_packet"] == 3 + assert tb.stats["stat_tx_byte"] == 64*3 + assert tb.stats["stat_tx_pkt_len"] == 64*3 + assert tb.stats["stat_tx_pkt_ucast"] == 3 + assert tb.stats["stat_tx_pkt_mcast"] == 0 + assert tb.stats["stat_tx_pkt_bcast"] == 0 + assert tb.stats["stat_tx_pkt_vlan"] == 0 + assert tb.stats["stat_tx_pkt_good"] == 2 + assert tb.stats["stat_tx_pkt_bad"] == 1 + assert tb.stats["stat_tx_err_oversize"] == 0 + assert tb.stats["stat_tx_err_user"] == 0 + assert tb.stats["stat_tx_err_underflow"] == 1 + await RisingEdge(dut.clk) await RisingEdge(dut.clk) @@ -208,6 +270,22 @@ async def run_test_error(dut, ifg=12, enable_gen=None, mii_sel=False): assert tb.sink.empty() + for stat, val in tb.stats.items(): + tb.log.info("%s: %d", stat, val) + + assert tb.stats["tx_start_packet"] == 3 + assert tb.stats["stat_tx_byte"] == 64*3 + assert tb.stats["stat_tx_pkt_len"] == 64*3 + assert tb.stats["stat_tx_pkt_ucast"] == 3 + assert tb.stats["stat_tx_pkt_mcast"] == 0 + assert tb.stats["stat_tx_pkt_bcast"] == 0 + assert tb.stats["stat_tx_pkt_vlan"] == 0 + assert tb.stats["stat_tx_pkt_good"] == 2 + assert tb.stats["stat_tx_pkt_bad"] == 1 + assert tb.stats["stat_tx_err_oversize"] == 0 + assert tb.stats["stat_tx_err_user"] == 1 + assert tb.stats["stat_tx_err_underflow"] == 0 + await RisingEdge(dut.clk) await RisingEdge(dut.clk) @@ -246,6 +324,22 @@ async def run_test_oversize(dut, ifg=12, enable_gen=None, mii_sel=False): assert tb.sink.empty() + for stat, val in tb.stats.items(): + tb.log.info("%s: %d", stat, val) + + assert tb.stats["tx_start_packet"] == 3 + assert tb.stats["stat_tx_byte"] == 64*2+1519 + assert tb.stats["stat_tx_pkt_len"] == 64*2+1519 + assert tb.stats["stat_tx_pkt_ucast"] == 3 + assert tb.stats["stat_tx_pkt_mcast"] == 0 + assert tb.stats["stat_tx_pkt_bcast"] == 0 + assert tb.stats["stat_tx_pkt_vlan"] == 0 + assert tb.stats["stat_tx_pkt_good"] == 2 + assert tb.stats["stat_tx_pkt_bad"] == 1 + assert tb.stats["stat_tx_err_oversize"] == 1 + assert tb.stats["stat_tx_err_user"] == 0 + assert tb.stats["stat_tx_err_underflow"] == 0 + await RisingEdge(dut.clk) await RisingEdge(dut.clk)