diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/extmod/tls_noleak.py (renamed from tests/extmod/ssl_noleak.py) | 0 | ||||
| -rw-r--r-- | tests/extmod/tls_threads.py | 57 | ||||
| -rw-r--r-- | tests/ports/rp2/rp2_dma.py | 39 |
3 files changed, 80 insertions, 16 deletions
diff --git a/tests/extmod/ssl_noleak.py b/tests/extmod/tls_noleak.py index 870032d58e..870032d58e 100644 --- a/tests/extmod/ssl_noleak.py +++ b/tests/extmod/tls_noleak.py diff --git a/tests/extmod/tls_threads.py b/tests/extmod/tls_threads.py new file mode 100644 index 0000000000..4564abd3d8 --- /dev/null +++ b/tests/extmod/tls_threads.py @@ -0,0 +1,57 @@ +# Ensure that SSL sockets can be allocated from multiple +# threads without thread safety issues +import unittest + +try: + import _thread + import io + import tls + import time +except ImportError: + print("SKIP") + raise SystemExit + + +class TestSocket(io.IOBase): + def write(self, buf): + return len(buf) + + def readinto(self, buf): + return 0 + + def ioctl(self, cmd, arg): + return 0 + + def setblocking(self, value): + pass + + +ITERS = 256 + + +class TLSThreads(unittest.TestCase): + def test_sslsocket_threaded(self): + self.done = False + # only run in two threads: too much RAM demand otherwise, and rp2 only + # supports two anyhow + _thread.start_new_thread(self._alloc_many_sockets, (True,)) + self._alloc_many_sockets(False) + while not self.done: + time.sleep(0.1) + print("done") + + def _alloc_many_sockets(self, set_done_flag): + print("start", _thread.get_ident()) + ctx = tls.SSLContext(tls.PROTOCOL_TLS_CLIENT) + ctx.verify_mode = tls.CERT_NONE + for n in range(ITERS): + s = TestSocket() + s = ctx.wrap_socket(s, do_handshake_on_connect=False) + s.close() # Free associated resources now from thread, not in a GC pass + print("done", _thread.get_ident()) + if set_done_flag: + self.done = True + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/ports/rp2/rp2_dma.py b/tests/ports/rp2/rp2_dma.py index bd7f17913e..436e5ee48e 100644 --- a/tests/ports/rp2/rp2_dma.py +++ b/tests/ports/rp2/rp2_dma.py @@ -20,7 +20,7 @@ class Test(unittest.TestCase): def test_printing(self): dma = self.dma - self.assertEqual(str(dma), "DMA(0)") + self.assertEqual(str(dma), "DMA({})".format(dma.channel)) def test_pack_unpack_ctrl(self): dma = self.dma @@ -34,7 +34,7 @@ class Test(unittest.TestCase): self.assertEqual(ctrl_dict["ahb_err"], 0) self.assertEqual(ctrl_dict["bswap"], 0) self.assertEqual(ctrl_dict["busy"], 0) - self.assertEqual(ctrl_dict["chain_to"], 0) + self.assertEqual(ctrl_dict["chain_to"], dma.channel) self.assertEqual(ctrl_dict["enable"], 1) self.assertEqual(ctrl_dict["high_pri"], 0) self.assertEqual(ctrl_dict["inc_read"], 1) @@ -58,7 +58,7 @@ class Test(unittest.TestCase): self.assertEqual(dma.write, 0) self.assertEqual(dma.count, 0) self.assertEqual(dma.ctrl & 0x01F, 25) - self.assertEqual(dma.channel, 0) + self.assertIn(dma.channel, range(16)) self.assertIsInstance(dma.registers, memoryview) def test_close(self): @@ -86,26 +86,32 @@ class Test(unittest.TestCase): def test_time_taken_for_large_memory_copy(self): def run_and_time_dma(dma): ticks_us = time.ticks_us + active = dma.active irq_state = machine.disable_irq() t0 = ticks_us() - dma.active(True) - while dma.active(): + active(True) + while active(): pass t1 = ticks_us() machine.enable_irq(irq_state) return time.ticks_diff(t1, t0) - dma = self.dma - dest = bytearray(16 * 1024) - dma.read = SRC - dma.write = dest - dma.count = len(dest) // 4 - dma.ctrl = dma.pack_ctrl() - dt = run_and_time_dma(dma) - expected_dt_range = range(40, 70) if is_rp2350 else range(70, 125) - self.assertIn(dt, expected_dt_range) - self.assertEqual(dest[:8], SRC[:8]) - self.assertEqual(dest[-8:], SRC[-8:]) + num_average = 10 + dt_sum = 0 + for _ in range(num_average): + dma = self.dma + dest = bytearray(16 * 1024) + dma.read = SRC + dma.write = dest + dma.count = len(dest) // 4 + dma.ctrl = dma.pack_ctrl() + dt_sum += run_and_time_dma(dma) + self.assertEqual(dest[:8], SRC[:8]) + self.assertEqual(dest[-8:], SRC[-8:]) + self.tearDown() + self.setUp() + dt = dt_sum // num_average + self.assertIn(dt, range(30, 80)) def test_config_trigger(self): # Test using .config(trigger=True) to start DMA immediately. @@ -136,6 +142,7 @@ class Test(unittest.TestCase): ) while dma.active(): pass + time.sleep_ms(1) # when running as native code, give the scheduler a chance to run self.assertEqual(irq_flags, 1) self.assertEqual(dest[:8], SRC[:8]) self.assertEqual(dest[-8:], SRC[-8:]) |