1 files changed, 73 insertions, 0 deletions
diff --git a/tests/gadget-zero/test_gadget0.py b/tests/gadget-zero/test_gadget0.py
index 830c631a..c27c10f7 100644
--- a/tests/gadget-zero/test_gadget0.py
+++ b/tests/gadget-zero/test_gadget0.py
@@ -1,5 +1,6 @@
 import array
 import datetime
+import random
 import usb.core
 import usb.util as uu
 import sys
@@ -181,6 +182,78 @@ class TestConfigSourceSink(unittest.TestCase):
             self.assertIn("Pipe", e.strerror)
 
 
+class TestConfigLoopBack(unittest.TestCase):
+    """
+    We could inherit, but it doesn't save much, and this saves me from remembering how to call super.
+    """
+
+    def setUp(self):
+        self.dev = usb.core.find(idVendor=VENDOR_ID, idProduct=PRODUCT_ID, custom_match=find_by_serial(DUT_SERIAL))
+        self.assertIsNotNone(self.dev, "Couldn't find locm3 gadget0 device")
+
+        self.cfg = uu.find_descriptor(self.dev, bConfigurationValue=3)
+        self.assertIsNotNone(self.cfg, "Config 3 should exist")
+        self.dev.set_configuration(self.cfg)
+        self.intf = self.cfg[(0, 0)]
+        # heh, kinda gross...
+        self.eps_out = [ep for ep in self.intf if uu.endpoint_direction(ep.bEndpointAddress) == uu.ENDPOINT_OUT]
+        self.eps_in = [ep for ep in self.intf if uu.endpoint_direction(ep.bEndpointAddress) == uu.ENDPOINT_IN]
+
+    def tearDown(self):
+        uu.dispose_resources(self.dev)
+
+    def _inner_basic(self, ep_out, ep_in, data):
+        written = self.dev.write(ep_out, data)
+        self.assertEqual(written, len(data), "Should have written all bytes plz")
+        read = self.dev.read(ep_in, len(data))
+        self.assertEqual(len(data), len(read))
+        expected = array.array('B', [x for x in data])
+        self.assertEqual(expected, read, "should have read back what we wrote")
+
+
+    def test_simple_loop(self):
+        """Plain simple loopback, does it work at all"""
+        eout = self.eps_out[0]
+        ein = self.eps_in[0]
+        data = [random.randrange(255) for _ in range(eout.wMaxPacketSize)]
+        self._inner_basic(eout, ein, data)
+
+    def test_dual_loop(self):
+        """Testing that we don't mix our data up, just plain and simple"""
+        dlen = self.eps_out[0].wMaxPacketSize
+        data = [
+            [0xaa for _ in range(dlen)],
+            [0xbb for _ in range(dlen)],
+        ]
+        for epo, epi, data in zip(self.eps_out, self.eps_in, data):
+            self._inner_basic(epo, epi, data)
+
+    def test_dual_loop_back_to_back(self):
+        """
+        write to both, _before_ we read back...
+        This can expose problems with buffer management
+        """
+        dlen = self.eps_out[0].wMaxPacketSize
+        data = [
+            [0xaa for _ in range(dlen)],
+            [0xbb for _ in range(dlen)],
+        ]
+        written = [
+            self.dev.write(self.eps_out[0], data[0]),
+            self.dev.write(self.eps_out[1], data[1]),
+        ]
+        read = [
+            self.dev.read(self.eps_in[0], dlen),
+            self.dev.read(self.eps_in[1], dlen),
+        ]
+
+        for w, r, dat in zip(written, read, data):
+            self.assertEqual(w, len(dat), "Should have written all bytes plz")
+            self.assertEqual(len(dat), len(r), "Should have read back same size")
+            expected = array.array('B', [x for x in dat])
+            self.assertEqual(expected, r, "should have read back what we wrote")
+
+
 @unittest.skip("Perf tests only on demand (comment this line!)")
 class TestConfigSourceSinkPerformance(unittest.TestCase):
     """