test/test_stream.py

   1 # This file is Copyright (c) 2020 Florent Kermarrec <florent@enjoy-digital.fr>
   2 # License: BSD
   3
   4 import unittest
   5 import random
   6
   7 from migen import *
   8
   9 from litex.soc.interconnect.stream import *
  10
  11
  12 class TestStream(unittest.TestCase):
  13     def pipe_test(self, dut):
  14         prng = random.Random(42)
  15         def generator(dut, valid_rand=90):
  16             for data in range(128):
  17                 yield dut.sink.valid.eq(1)
  18                 yield dut.sink.data.eq(data)
  19                 yield
  20                 while (yield dut.sink.ready) == 0:
  21                     yield
  22                 yield dut.sink.valid.eq(0)
  23                 while prng.randrange(100) < valid_rand:
  24                     yield
  25
  26         def checker(dut, ready_rand=90):
  27             dut.errors = 0
  28             for data in range(128):
  29                 yield dut.source.ready.eq(0)
  30                 yield
  31                 while (yield dut.source.valid) == 0:
  32                     yield
  33                 while prng.randrange(100) < ready_rand:
  34                     yield
  35                 yield dut.source.ready.eq(1)
  36                 yield
  37                 if ((yield dut.source.data) != data):
  38                     dut.errors += 1
  39             yield
  40         run_simulation(dut, [generator(dut), checker(dut)])
  41         self.assertEqual(dut.errors, 0)
  42
  43     def test_pipe_valid(self):
  44         dut = PipeValid([("data", 8)])
  45         self.pipe_test(dut)
  46
  47     def test_pipe_ready(self):
  48         dut = PipeReady([("data", 8)])
  49         self.pipe_test(dut)