gateware/test/serwb: update and cleanup test (v2...)

artiq/gateware/test/serwb/test_etherbone.py

@@ -1,70 +0,0 @@
-import unittest
-import random
-
-from migen import *
-
-from misoc.interconnect.wishbone import SRAM
-from misoc.interconnect.stream import Converter
-
-from artiq.gateware.serwb import packet
-from artiq.gateware.serwb import etherbone
-
-
-class DUT(Module):
-    def __init__(self):
-        # wishbone slave
-        slave_depacketizer = packet.Depacketizer(int(100e6))
-        slave_packetizer = packet.Packetizer()
-        self.submodules += slave_depacketizer, slave_packetizer
-        slave_etherbone = etherbone.Etherbone(mode="slave")
-        self.submodules += slave_etherbone
-        self.comb += [
-            slave_depacketizer.source.connect(slave_etherbone.sink),
-            slave_etherbone.source.connect(slave_packetizer.sink)
-        ]
-
-        # wishbone master
-        master_depacketizer = packet.Depacketizer(int(100e6))
-        master_packetizer = packet.Packetizer()
-        self.submodules += master_depacketizer, master_packetizer
-        master_etherbone = etherbone.Etherbone(mode="master")
-        master_sram = SRAM(64, bus=master_etherbone.wishbone.bus)
-        self.submodules += master_etherbone, master_sram
-        self.comb += [
-            master_depacketizer.source.connect(master_etherbone.sink),
-            master_etherbone.source.connect(master_packetizer.sink)
-        ]
-
-        # connect core directly with converters in the loop
-        s2m_downconverter = Converter(32, 16)
-        s2m_upconverter = Converter(16, 32)
-        self.submodules += s2m_downconverter, s2m_upconverter
-        m2s_downconverter = Converter(32, 16)
-        m2s_upconverter = Converter(16, 32)
-        self.submodules += m2s_upconverter, m2s_downconverter
-        self.comb += [
-        	slave_packetizer.source.connect(s2m_downconverter.sink),
-        	s2m_downconverter.source.connect(s2m_upconverter.sink),
-        	s2m_upconverter.source.connect(master_depacketizer.sink),
-
-        	master_packetizer.source.connect(m2s_downconverter.sink),
-        	m2s_downconverter.source.connect(m2s_upconverter.sink),
-        	m2s_upconverter.source.connect(slave_depacketizer.sink)
-        ]
-
-        # expose wishbone slave
-        self.wishbone = slave_etherbone.wishbone.bus
-
-
-class TestEtherbone(unittest.TestCase):
-    def test_write_read_sram(self):
-        dut = DUT()
-        prng = random.Random(1)
-        def generator(dut):
-            datas = [prng.randrange(0, 2**32-1) for i in range(16)]
-            for i in range(16):
-                yield from dut.wishbone.write(i, datas[i])
-            for i in range(16):
-                data = (yield from dut.wishbone.read(i))
-                self.assertEqual(data, datas[i])
-        run_simulation(dut, generator(dut))

artiq/gateware/test/serwb/test_serwb_core.py

@@ -0,0 +1,128 @@
+import unittest
+import random
+
+from migen import *
+
+from artiq.gateware.serwb import scrambler
+from artiq.gateware.serwb import core
+
+from misoc.interconnect.wishbone import SRAM
+
+
+class FakeInit(Module):
+    def __init__(self):
+        self.ready = 1
+
+
+class FakeSerdes(Module):
+    def __init__(self):
+        self.tx_k = Signal(4)
+        self.tx_d = Signal(32)
+        self.rx_k = Signal(4)
+        self.rx_d = Signal(32)
+
+
+class FakePHY(Module):
+    cd = "sys"
+    def __init__(self):
+        self.init = FakeInit()
+        self.serdes = FakeSerdes()
+
+
+class DUTScrambler(Module):
+    def __init__(self):
+        self.submodules.scrambler = scrambler.Scrambler(sync_interval=16)
+        self.submodules.descrambler = scrambler.Descrambler()
+        self.comb += [
+            self.scrambler.source.connect(self.descrambler.sink),
+            self.descrambler.source.ack.eq(1)
+        ]
+
+
+class DUTCore(Module):
+    def __init__(self, **kwargs):
+        # wishbone slave
+        phy_slave = FakePHY()
+        serwb_slave = core.SERWBCore(phy_slave, int(1e6), "slave", **kwargs)
+        self.submodules += phy_slave, serwb_slave
+
+        # wishbone master
+        phy_master = FakePHY()
+        serwb_master = core.SERWBCore(phy_master, int(1e6), "master", **kwargs)
+        self.submodules += phy_master, serwb_master
+
+        # connect phy
+        self.comb += [
+            phy_master.serdes.rx_k.eq(phy_slave.serdes.tx_k),
+            phy_master.serdes.rx_d.eq(phy_slave.serdes.tx_d),
+            phy_slave.serdes.rx_k.eq(phy_master.serdes.tx_k),
+            phy_slave.serdes.rx_d.eq(phy_master.serdes.tx_d)
+        ]
+
+        # add wishbone sram to wishbone master
+        sram = SRAM(1024, bus=serwb_master.etherbone.wishbone.bus)
+        self.submodules += sram
+
+        # expose wishbone slave
+        self.wishbone = serwb_slave.etherbone.wishbone.bus
+
+
+class TestSERWBCore(unittest.TestCase):
+    def test_scrambler(self):
+        def generator(dut):
+            i = 0
+            last_data = -1
+            while i != 256:
+                # stim
+                if (yield dut.scrambler.sink.ack):
+                    i += 1
+                yield dut.scrambler.sink.data.eq(i)
+
+                # check
+                if (yield dut.descrambler.source.stb):
+                    current_data = (yield dut.descrambler.source.data)
+                    if (current_data != (last_data + 1)):
+                        dut.errors += 1
+                    last_data = current_data
+
+                # cycle
+                yield
+
+        dut = DUTScrambler()
+        dut.errors = 0
+        run_simulation(dut, generator(dut))
+        self.assertEqual(dut.errors, 0)
+
+    def test_serwb(self):
+        def generator(dut):
+            # prepare test
+            prng = random.Random(42)
+            data_base = 0x100
+            data_length = 4
+            datas_w = [prng.randrange(2**32) for i in range(data_length)]
+            datas_r = []
+
+            # write
+            for i in range(data_length):
+                yield from dut.wishbone.write(data_base + i, datas_w[i])
+
+            # read
+            for i in range(data_length):
+                datas_r.append((yield from dut.wishbone.read(data_base + i)))
+
+            # check
+            for i in range(data_length):
+                if datas_r[i] != datas_w[i]:
+                    dut.errors += 1
+
+        # scrambling off
+        dut = DUTCore(with_scrambling=False)
+        dut.errors = 0
+        run_simulation(dut, generator(dut))
+        self.assertEqual(dut.errors, 0)
+
+        # scrambling on
+        dut = DUTCore(with_scrambling=True)
+        dut.errors = 0
+        run_simulation(dut, generator(dut))
+        self.assertEqual(dut.errors, 0)

artiq/gateware/test/serwb/test_serwb_init.py

@@ -4,7 +4,6 @@ import unittest
 from migen import *
 
 from artiq.gateware.serwb import packet
-from artiq.gateware.serwb import etherbone
 from artiq.gateware.serwb.phy import _SerdesMasterInit, _SerdesSlaveInit
 
 
@@ -116,7 +115,7 @@ def generator(test, dut, valid_bitslip, valid_delays, check_success):
         test.assertEqual(error, 1)
 
 
-class TestPHYInit(unittest.TestCase):
+class TestSERWBInit(unittest.TestCase):
     def test_master_init_success(self):
         dut = DUTMaster()
         valid_bitslip = 2