drtio: add switching input test

This commit is contained in:
Sebastien Bourdeauducq 2018-09-19 17:50:29 +08:00
parent 08be176369
commit b86b6dcc09

View File

@ -69,11 +69,12 @@ class Testbench:
self.dut = DUT(2)
self.now = 0
def init(self):
def init(self, with_buffer_space=True):
yield from self.dut.master.rt_controller.csrs.underflow_margin.write(100)
while not (yield from self.dut.master.link_layer.rx_up.read()):
yield
yield from self.get_buffer_space()
if with_buffer_space:
yield from self.get_buffer_space()
def get_buffer_space(self):
csrs = self.dut.master.rt_controller.csrs
@ -106,13 +107,35 @@ class Testbench:
if status & 0x4:
return "destination unreachable"
def read(self, channel, timeout):
mcri = self.dut.master.cri
yield mcri.chan_sel.eq(channel)
yield mcri.timestamp.eq(timeout)
yield
yield mcri.cmd.eq(cri.commands["read"])
yield
yield mcri.cmd.eq(cri.commands["nop"])
yield
status = yield mcri.i_status
while status & 0x4:
yield
status = yield mcri.i_status
if status & 0x1:
return "timeout"
if status & 0x2:
return "overflow"
if status & 0x8:
return "destination unreachable"
return ((yield mcri.i_timestamp),
(yield mcri.i_data))
class TestSwitching(unittest.TestCase):
clocks = {"sys": 8, "rtio": 5, "rtio_rx": 5,
"rio": 5, "rio_phy": 5,
"sys_with_rst": 8, "rtio_with_rst": 5}
def test_switching(self):
def test_outputs(self):
tb = Testbench()
def test():
@ -163,3 +186,64 @@ class TestSwitching(unittest.TestCase):
run_simulation(tb.dut,
{"sys": test(), "rtio": tb.dut.pr.receive(receive), "rtio_rx": send_replies()}, self.clocks)
def test_inputs(self):
tb = Testbench()
def test():
yield from tb.init(with_buffer_space=False)
reply = yield from tb.read(19, 145)
self.assertEqual(reply, (333, 23))
reply = yield from tb.read(20, 146)
self.assertEqual(reply, (334, 24))
reply = yield from tb.read(10, 34)
self.assertEqual(reply, "timeout")
reply = yield from tb.read(1, 20)
self.assertEqual(reply, "overflow")
reply = yield from tb.read(21, 147)
self.assertEqual(reply, (335, 25))
for _ in range(40):
yield
current_request = None
def get_request():
nonlocal current_request
while current_request is None:
yield
r = current_request
current_request = None
return r
def expect_read(chan_sel, timeout, reply):
packet_type, field_dict, trailer = yield from get_request()
self.assertEqual(packet_type, "read_request")
self.assertEqual(trailer, [])
self.assertEqual(field_dict["chan_sel"], chan_sel)
self.assertEqual(field_dict["timeout"], timeout)
if reply == "timeout":
yield from tb.dut.pt.send("read_reply_noevent", overflow=0)
elif reply == "overflow":
yield from tb.dut.pt.send("read_reply_noevent", overflow=1)
else:
timestamp, data = reply
yield from tb.dut.pt.send("read_reply", timestamp=timestamp, data=data)
@passive
def send_replies():
yield from expect_read(19, 145, (333, 23))
yield from expect_read(20, 146, (334, 24))
yield from expect_read(10, 34, "timeout")
yield from expect_read(1, 20, "overflow")
yield from expect_read(21, 147, (335, 25))
unexpected = yield from get_request()
self.fail("unexpected packet: {}".format(unexpected))
def receive(packet_type, field_dict, trailer):
nonlocal current_request
self.assertEqual(current_request, None)
current_request = (packet_type, field_dict, trailer)
run_simulation(tb.dut,
{"sys": test(), "rtio": tb.dut.pr.receive(receive), "rtio_rx": send_replies()}, self.clocks)