Add overflow test, previous test is actually an oversize frame test

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

Author: alexforencich

tb/axis_async_fifo/test_axis_async_fifo.py

@@ -457,6 +457,44 @@ async def run_test_overflow(dut):
 
     tb.sink.pause = True
 
+    size = (16*byte_lanes)
+    count = depth*2 // size
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), size))
+    test_frame = AxiStreamFrame(test_data)
+    for k in range(count):
+        await tb.source.send(test_frame)
+
+    for k in range((depth//byte_lanes)*2):
+        await RisingEdge(dut.s_clk)
+
+    assert not tb.source.idle()
+
+    tb.sink.pause = False
+
+    for k in range(count):
+        rx_frame = await tb.sink.recv()
+
+        assert rx_frame.tdata == test_data
+        assert not rx_frame.tuser
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.s_clk)
+    await RisingEdge(dut.s_clk)
+
+
+async def run_test_oversize(dut):
+
+    tb = TB(dut)
+
+    depth = dut.DEPTH.value
+    byte_lanes = tb.source.byte_lanes
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
     test_data = bytearray(itertools.islice(itertools.cycle(range(256)), depth*2))
     test_frame = AxiStreamFrame(test_data)
     await tb.source.send(test_frame)
@@ -558,7 +596,8 @@ def incrementing_payload(length):
                 run_test_shift_out_source_reset,
                 run_test_shift_out_sink_reset,
                 run_test_pause,
-                run_test_overflow
+                run_test_overflow,
+                run_test_oversize
             ]:
 
         factory = TestFactory(test)

tb/axis_async_fifo_adapter/test_axis_async_fifo_adapter.py

@@ -454,6 +454,44 @@ async def run_test_overflow(dut):
 
     tb.sink.pause = True
 
+    size = (16*byte_lanes)
+    count = depth*2 // size
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), size))
+    test_frame = AxiStreamFrame(test_data)
+    for k in range(count):
+        await tb.source.send(test_frame)
+
+    for k in range((depth//byte_lanes)*2):
+        await RisingEdge(dut.s_clk)
+
+    assert not tb.source.idle()
+
+    tb.sink.pause = False
+
+    for k in range(count):
+        rx_frame = await tb.sink.recv()
+
+        assert rx_frame.tdata == test_data
+        assert not rx_frame.tuser
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.s_clk)
+    await RisingEdge(dut.s_clk)
+
+
+async def run_test_oversize(dut):
+
+    tb = TB(dut)
+
+    depth = dut.DEPTH.value
+    byte_lanes = min(tb.source.byte_lanes, tb.sink.byte_lanes)
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
     test_data = bytearray(itertools.islice(itertools.cycle(range(256)), depth*2))
     test_frame = AxiStreamFrame(test_data)
     await tb.source.send(test_frame)
@@ -555,7 +593,8 @@ def incrementing_payload(length):
                 run_test_shift_out_source_reset,
                 run_test_shift_out_sink_reset,
                 run_test_pause,
-                run_test_overflow
+                run_test_overflow,
+                run_test_oversize
             ]:
 
         factory = TestFactory(test)

tb/axis_fifo/test_axis_fifo.py

@@ -243,6 +243,44 @@ async def run_test_overflow(dut):
 
     tb.sink.pause = True
 
+    size = (16*byte_lanes)
+    count = depth*2 // size
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), size))
+    test_frame = AxiStreamFrame(test_data)
+    for k in range(count):
+        await tb.source.send(test_frame)
+
+    for k in range((depth//byte_lanes)*2):
+        await RisingEdge(dut.clk)
+
+    assert not tb.source.idle()
+
+    tb.sink.pause = False
+
+    for k in range(count):
+        rx_frame = await tb.sink.recv()
+
+        assert rx_frame.tdata == test_data
+        assert not rx_frame.tuser
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_oversize(dut):
+
+    tb = TB(dut)
+
+    depth = dut.DEPTH.value
+    byte_lanes = tb.source.byte_lanes
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
     test_data = bytearray(itertools.islice(itertools.cycle(range(256)), depth*2))
     test_frame = AxiStreamFrame(test_data)
     await tb.source.send(test_frame)
@@ -338,7 +376,8 @@ def incrementing_payload(length):
                 run_test_init_sink_pause,
                 run_test_init_sink_pause_reset,
                 run_test_pause,
-                run_test_overflow
+                run_test_overflow,
+                run_test_oversize
             ]:
 
         factory = TestFactory(test)

tb/axis_fifo_adapter/test_axis_fifo_adapter.py

@@ -243,6 +243,44 @@ async def run_test_overflow(dut):
 
     tb.sink.pause = True
 
+    size = (16*byte_lanes)
+    count = depth*2 // size
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), size))
+    test_frame = AxiStreamFrame(test_data)
+    for k in range(count):
+        await tb.source.send(test_frame)
+
+    for k in range((depth//byte_lanes)*2):
+        await RisingEdge(dut.clk)
+
+    assert not tb.source.idle()
+
+    tb.sink.pause = False
+
+    for k in range(count):
+        rx_frame = await tb.sink.recv()
+
+        assert rx_frame.tdata == test_data
+        assert not rx_frame.tuser
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_oversize(dut):
+
+    tb = TB(dut)
+
+    depth = dut.DEPTH.value
+    byte_lanes = min(tb.source.byte_lanes, tb.sink.byte_lanes)
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
     test_data = bytearray(itertools.islice(itertools.cycle(range(256)), depth*2))
     test_frame = AxiStreamFrame(test_data)
     await tb.source.send(test_frame)
@@ -338,7 +376,8 @@ def incrementing_payload(length):
                 run_test_init_sink_pause,
                 run_test_init_sink_pause_reset,
                 run_test_pause,
-                run_test_overflow
+                run_test_overflow,
+                run_test_oversize
             ]:
 
         factory = TestFactory(test)