gram/test/test_core_multiplexer.py

   1 from nmigen import *
   2
   3 from gram.core.multiplexer import _AntiStarvation, _CommandChooser
   4 from gram.common import cmd_request_rw_layout
   5 import gram.stream as stream
   6 from gram.test.utils import *
   7
   8 class CommandChooserTestCase(FHDLTestCase):
   9     def prepare_testbench(self):
  10         a = 16
  11         ba = 3
  12         requests = []
  13         for i in range(10):
  14             requests += [stream.Endpoint(cmd_request_rw_layout(a, ba))]
  15         dut = _CommandChooser(requests)
  16
  17         return (requests, dut)
  18
  19     def test_wants(self):
  20         requests, dut = self.prepare_testbench()
  21
  22         def process():
  23             for i in range(10):
  24                 yield requests[i].a.eq(i)
  25
  26             # Fake requests, non valid requests shouldn't be picked
  27             yield requests[5].is_read.eq(1)
  28             yield requests[5].valid.eq(1)
  29             yield requests[6].is_read.eq(1)
  30             yield requests[6].valid.eq(0)
  31             yield requests[7].is_write.eq(1)
  32             yield requests[7].valid.eq(1)
  33             yield requests[8].is_write.eq(1)
  34             yield requests[8].valid.eq(0)
  35
  36             # want_writes
  37             yield dut.want_writes.eq(1)
  38             yield; yield Delay(1e-8)
  39             self.assertEqual((yield dut.cmd.a), 7)
  40
  41             # want_reads
  42             yield dut.want_writes.eq(0)
  43             yield dut.want_reads.eq(1)
  44             yield; yield Delay(1e-8)
  45             self.assertEqual((yield dut.cmd.a), 5)
  46
  47         runSimulation(dut, process, "test_core_multiplexer_commandchooser.vcd")
  48
  49     def test_helpers(self):
  50         requests, dut = self.prepare_testbench()
  51
  52         def process():
  53             for i in range(10):
  54                 yield requests[i].a.eq(i)
  55
  56             # Fake requests
  57             yield requests[5].is_read.eq(1)
  58             yield requests[5].valid.eq(1)
  59             yield requests[6].is_read.eq(1)
  60             yield requests[6].valid.eq(0)
  61             yield requests[7].is_write.eq(1)
  62             yield requests[7].valid.eq(1)
  63             yield requests[8].is_write.eq(1)
  64             yield requests[8].valid.eq(0)
  65
  66             # want_writes
  67             yield dut.want_writes.eq(1)
  68             yield; yield Delay(1e-8)
  69             self.assertTrue((yield dut.write()))
  70             self.assertFalse((yield dut.read()))
  71
  72             # want_reads
  73             yield dut.want_writes.eq(0)
  74             yield dut.want_reads.eq(1)
  75             yield; yield Delay(1e-8)
  76             self.assertTrue((yield dut.read()))
  77             self.assertFalse((yield dut.write()))
  78
  79         runSimulation(dut, process, "test_core_multiplexer_commandchooser.vcd")
  80
  81 class AntiStarvationTestCase(FHDLTestCase):
  82     def test_duration(self):
  83         def generic_test(timeout):
  84             dut = _AntiStarvation(timeout)
  85
  86             def process():
  87                 yield dut.en.eq(1)
  88                 yield
  89                 yield dut.en.eq(0)
  90                 yield
  91
  92                 for i in range(timeout):
  93                     self.assertFalse((yield dut.max_time))
  94                     yield
  95
  96                 self.assertTrue((yield dut.max_time))
  97
  98             runSimulation(dut, process, "test_core_multiplexer_antistarvation.vcd")
  99
 100     def test_formal(self):
 101         def generic_test(timeout):
 102             dut = _AntiStarvation(timeout)
 103             self.assertFormal(dut, mode="bmc", depth=timeout+1)
 104
 105         generic_test(0)
 106         generic_test(5)
 107         generic_test(10)
 108         generic_test(0x20)