More tests for pipeline FIFO

Author: alexforencich

tb/axis_pipeline_fifo/test_axis_pipeline_fifo.py

@@ -131,6 +131,90 @@ async def run_test_tuser_assert(dut):
     await RisingEdge(dut.clk)
 
 
+async def run_test_init_sink_pause(dut):
+
+    tb = TB(dut)
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), 32))
+    test_frame = AxiStreamFrame(test_data)
+    await tb.source.send(test_frame)
+
+    for k in range(64):
+        await RisingEdge(dut.clk)
+
+    tb.sink.pause = False
+
+    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_init_sink_pause_reset(dut):
+
+    tb = TB(dut)
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), 32))
+    test_frame = AxiStreamFrame(test_data)
+    await tb.source.send(test_frame)
+
+    for k in range(64):
+        await RisingEdge(dut.clk)
+
+    await tb.reset()
+
+    tb.sink.pause = False
+
+    for k in range(64):
+        await RisingEdge(dut.clk)
+
+    assert tb.sink.empty()
+
+    await RisingEdge(dut.clk)
+    await RisingEdge(dut.clk)
+
+
+async def run_test_overflow(dut):
+
+    tb = TB(dut)
+
+    await tb.reset()
+
+    tb.sink.pause = True
+
+    test_data = bytearray(itertools.islice(itertools.cycle(range(256)), 2048))
+    test_frame = AxiStreamFrame(test_data)
+    await tb.source.send(test_frame)
+
+    for k in range(2048):
+        await RisingEdge(dut.clk)
+
+    tb.sink.pause = False
+
+    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_stress_test(dut, idle_inserter=None, backpressure_inserter=None):
 
     tb = TB(dut)
@@ -196,7 +280,13 @@ def incrementing_payload(length):
     factory.add_option("backpressure_inserter", [None, cycle_pause])
     factory.generate_tests()
 
-    for test in [run_test_tuser_assert]:
+    for test in [
+                run_test_tuser_assert,
+                run_test_init_sink_pause,
+                run_test_init_sink_pause_reset,
+                run_test_overflow
+            ]:
+
         factory = TestFactory(test)
         factory.generate_tests()