Add TX underrun and error tests

Signed-off-by: Alex Forencich <[email protected]>

tb/axis_baser_tx_64/test_axis_baser_tx_64.py

@@ -216,6 +216,82 @@ async def run_test_alignment(dut, payload_data=None, ifg=12):
     await RisingEdge(dut.clk)
 
 
+async def run_test_underrun(dut, ifg=12):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        await tb.source.send(test_frame)
+
+    for k in range(16):
+        await RisingEdge(dut.clk)
+
+    tb.source.pause = True
+
+    for k in range(4):
+        await RisingEdge(dut.clk)
+
+    tb.source.pause = False
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.data[-1] == 0xFE
+            assert rx_frame.ctrl[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.ctrl is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_error(dut, ifg=12):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        if k == 1:
+            test_frame.tuser = 1
+        await tb.source.send(test_frame)
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.data[-1] == 0xFE
+            assert rx_frame.ctrl[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.ctrl is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
 def size_list():
     return list(range(60, 128)) + [512, 1514, 9214] + [60]*10
 
@@ -241,6 +317,12 @@ def cycle_en():
     factory.add_option("ifg", [12])
     factory.generate_tests()
 
+    for test in [run_test_underrun, run_test_error]:
+
+        factory = TestFactory(test)
+        factory.add_option("ifg", [12])
+        factory.generate_tests()
+
 
 # cocotb-test
 

tb/axis_gmii_tx/test_axis_gmii_tx.py

@@ -141,6 +141,94 @@ async def run_test(dut, payload_lengths=None, payload_data=None, ifg=12, enable_
     await RisingEdge(dut.clk)
 
 
+async def run_test_underrun(dut, ifg=12, enable_gen=None, mii_sel=False):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+    tb.dut.mii_select.value = mii_sel
+
+    if enable_gen is not None:
+        tb.set_enable_generator(enable_gen())
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        await tb.source.send(test_frame)
+
+    for k in range(200 if mii_sel else 100):
+        while True:
+            await RisingEdge(dut.clk)
+            if dut.clk_enable.value.integer:
+                break
+
+    tb.source.pause = True
+
+    for k in range(10):
+        while True:
+            await RisingEdge(dut.clk)
+            if dut.clk_enable.value.integer:
+                break
+
+    tb.source.pause = False
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.error[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.error is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_error(dut, ifg=12, enable_gen=None, mii_sel=False):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+    tb.dut.mii_select.value = mii_sel
+
+    if enable_gen is not None:
+        tb.set_enable_generator(enable_gen())
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        if k == 1:
+            test_frame.tuser = 1
+        await tb.source.send(test_frame)
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.error[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.error is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
 def size_list():
     return list(range(60, 128)) + [512, 1514] + [60]*10
 
@@ -163,6 +251,14 @@ def cycle_en():
     factory.add_option("mii_sel", [False, True])
     factory.generate_tests()
 
+    for test in [run_test_underrun, run_test_error]:
+
+        factory = TestFactory(test)
+        factory.add_option("ifg", [12])
+        factory.add_option("enable_gen", [None, cycle_en])
+        factory.add_option("mii_sel", [False, True])
+        factory.generate_tests()
+
 
 # cocotb-test
 

tb/axis_xgmii_tx_32/test_axis_xgmii_tx_32.py

@@ -197,6 +197,82 @@ async def run_test_alignment(dut, payload_data=None, ifg=12):
     await RisingEdge(dut.clk)
 
 
+async def run_test_underrun(dut, ifg=12):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        await tb.source.send(test_frame)
+
+    for k in range(32):
+        await RisingEdge(dut.clk)
+
+    tb.source.pause = True
+
+    for k in range(4):
+        await RisingEdge(dut.clk)
+
+    tb.source.pause = False
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.data[-1] == 0xFE
+            assert rx_frame.ctrl[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.ctrl is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_error(dut, ifg=12):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        if k == 1:
+            test_frame.tuser = 1
+        await tb.source.send(test_frame)
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.data[-1] == 0xFE
+            assert rx_frame.ctrl[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.ctrl is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
 def size_list():
     return list(range(60, 128)) + [512, 1514, 9214] + [60]*10
 
@@ -222,6 +298,12 @@ def cycle_en():
     factory.add_option("ifg", [12])
     factory.generate_tests()
 
+    for test in [run_test_underrun, run_test_error]:
+
+        factory = TestFactory(test)
+        factory.add_option("ifg", [12])
+        factory.generate_tests()
+
 
 # cocotb-test
 

tb/axis_xgmii_tx_64/test_axis_xgmii_tx_64.py

@@ -205,6 +205,82 @@ async def run_test_alignment(dut, payload_data=None, ifg=12):
     await RisingEdge(dut.clk)
 
 
+async def run_test_underrun(dut, ifg=12):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        await tb.source.send(test_frame)
+
+    for k in range(16):
+        await RisingEdge(dut.clk)
+
+    tb.source.pause = True
+
+    for k in range(4):
+        await RisingEdge(dut.clk)
+
+    tb.source.pause = False
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.data[-1] == 0xFE
+            assert rx_frame.ctrl[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.ctrl is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_error(dut, ifg=12):
+
+    tb = TB(dut)
+
+    tb.dut.cfg_ifg.value = ifg
+    tb.dut.cfg_tx_enable.value = 1
+
+    await tb.reset()
+
+    test_data = bytes(x for x in range(60))
+
+    for k in range(3):
+        test_frame = AxiStreamFrame(test_data)
+        if k == 1:
+            test_frame.tuser = 1
+        await tb.source.send(test_frame)
+
+    for k in range(3):
+        rx_frame = await tb.sink.recv()
+
+        if k == 1:
+            assert rx_frame.data[-1] == 0xFE
+            assert rx_frame.ctrl[-1] == 1
+        else:
+            assert rx_frame.get_payload() == test_data
+            assert rx_frame.check_fcs()
+            assert rx_frame.ctrl is None
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
 def size_list():
     return list(range(60, 128)) + [512, 1514, 9214] + [60]*10
 
@@ -230,6 +306,12 @@ def cycle_en():
     factory.add_option("ifg", [12])
     factory.generate_tests()
 
+    for test in [run_test_underrun, run_test_error]:
+
+        factory = TestFactory(test)
+        factory.add_option("ifg", [12])
+        factory.generate_tests()
+
 
 # cocotb-test