primitives/tests/delay_test.py

   1 # delay_test.py Tests for Delay_ms class
   2
   3 # Copyright (c) 2020 Peter Hinch
   4 # Released under the MIT License (MIT) - see LICENSE file
   5
   6 import uasyncio as asyncio
   7 import micropython
   8 from primitives.delay_ms import Delay_ms
   9
  10 micropython.alloc_emergency_exception_buf(100)
  11
  12 def printexp(exp, runtime=0):
  13     print('Expected output:')
  14     print('\x1b[32m')
  15     print(exp)
  16     print('\x1b[39m')
  17     if runtime:
  18         print('Running (runtime = {}s):'.format(runtime))
  19     else:
  20         print('Running (runtime < 1s):')
  21
  22 async def ctor_test():  # Constructor arg
  23     s = '''
  24 Trigger 5 sec delay
  25 Retrigger 5 sec delay
  26 Callback should run
  27 cb callback
  28 Done
  29 '''
  30     printexp(s, 12)
  31     def cb(v):
  32         print('cb', v)
  33
  34     d = Delay_ms(cb, ('callback',), duration=5000)
  35
  36     print('Trigger 5 sec delay')
  37     d.trigger()
  38     await asyncio.sleep(4)
  39     print('Retrigger 5 sec delay')
  40     d.trigger()
  41     await asyncio.sleep(4)
  42     print('Callback should run')
  43     await asyncio.sleep(2)
  44     print('Done')
  45
  46 async def launch_test():
  47     s = '''
  48 Trigger 5 sec delay
  49 Coroutine should run: run to completion.
  50 Coroutine starts
  51 Coroutine ends
  52 Coroutine should run: test cancellation.
  53 Coroutine starts
  54 Coroutine should run: test awaiting.
  55 Coroutine starts
  56 Coroutine ends
  57 Done
  58 '''
  59     printexp(s, 20)
  60     async def cb(v, ms):
  61         print(v, 'starts')
  62         await asyncio.sleep_ms(ms)
  63         print(v, 'ends')
  64
  65     d = Delay_ms(cb, ('coroutine', 1000))
  66
  67     print('Trigger 5 sec delay')
  68     d.trigger(5000)  # Test extending time
  69     await asyncio.sleep(4)
  70     print('Coroutine should run: run to completion.')
  71     await asyncio.sleep(3)
  72     d = Delay_ms(cb, ('coroutine', 3000))
  73     d.trigger(5000)
  74     await asyncio.sleep(4)
  75     print('Coroutine should run: test cancellation.')
  76     await asyncio.sleep(2)
  77     coro = d.rvalue()
  78     coro.cancel()
  79     d.trigger(5000)
  80     await asyncio.sleep(4)
  81     print('Coroutine should run: test awaiting.')
  82     await asyncio.sleep(2)
  83     coro = d.rvalue()
  84     await coro
  85     print('Done')
  86
  87
  88 async def reduce_test():  # Test reducing a running delay
  89     s = '''
  90 Trigger 5 sec delay
  91 Callback should run
  92 cb callback
  93 Callback should run
  94 cb callback
  95 Done
  96 '''
  97     printexp(s, 11)
  98     def cb(v):
  99         print('cb', v)
 100
 101     d = Delay_ms(cb, ('callback',))
 102
 103     print('Trigger 5 sec delay')
 104     d.trigger(5000)  # Test extending time
 105     await asyncio.sleep(4)
 106     print('Callback should run')
 107     await asyncio.sleep(2)
 108     d.trigger(10000)
 109     await asyncio.sleep(1)
 110     d.trigger(3000)
 111     await asyncio.sleep(2)
 112     print('Callback should run')
 113     await asyncio.sleep(2)
 114     print('Done')
 115
 116
 117 async def stop_test():  # Test the .stop and .running methods
 118     s = '''
 119 Trigger 5 sec delay
 120 Running
 121 Callback should run
 122 cb callback
 123 Callback returned 42
 124 Callback should not run
 125 Done
 126     '''
 127     printexp(s, 12)
 128     def cb(v):
 129         print('cb', v)
 130         return 42
 131
 132     d = Delay_ms(cb, ('callback',))
 133
 134     print('Trigger 5 sec delay')
 135     d.trigger(5000)  # Test extending time
 136     await asyncio.sleep(4)
 137     if d():
 138         print('Running')
 139     print('Callback should run')
 140     await asyncio.sleep(2)
 141     print('Callback returned', d.rvalue())
 142     d.trigger(3000)
 143     await asyncio.sleep(1)
 144     d.stop()
 145     await asyncio.sleep(1)
 146     if d():
 147         print('Running')
 148     print('Callback should not run')
 149     await asyncio.sleep(4)
 150     print('Done')
 151
 152
 153 async def isr_test():  # Test trigger from hard ISR
 154     from pyb import Timer
 155     s = '''
 156 Timer holds off cb for 5 secs
 157 cb should now run
 158 cb callback
 159 Done
 160 '''
 161     printexp(s, 6)
 162     def cb(v):
 163         print('cb', v)
 164
 165     d = Delay_ms(cb, ('callback',))
 166
 167     def timer_cb(_):
 168         d.trigger(200)
 169     tim = Timer(1, freq=10, callback=timer_cb)
 170
 171     print('Timer holds off cb for 5 secs')
 172     await asyncio.sleep(5)
 173     tim.deinit()
 174     print('cb should now run')
 175     await asyncio.sleep(1)
 176     print('Done')
 177
 178 async def err_test():  # Test triggering de-initialised timer
 179     s = '''
 180 Running (runtime = 3s):
 181 Trigger 1 sec delay
 182 cb callback
 183 Success: error was raised.
 184 Done
 185     '''
 186     printexp(s, 3)
 187     def cb(v):
 188         print('cb', v)
 189         return 42
 190
 191     d = Delay_ms(cb, ('callback',))
 192
 193     print('Trigger 1 sec delay')
 194     d.trigger(1000)
 195     await asyncio.sleep(2)
 196     d.deinit()
 197     try:
 198         d.trigger(1000)
 199     except RuntimeError:
 200         print("Success: error was raised.")
 201     print('Done')
 202
 203 av = '''
 204 Run a test by issuing
 205 delay_test.test(n)
 206 where n is a test number. Avaliable tests:
 207 \x1b[32m
 208 0 Test triggering from a hard ISR (Pyboard only)
 209 1 Test the .stop method and callback return value.
 210 2 Test reducing the duration of a running timer
 211 3 Test delay defined by constructor arg
 212 4 Test triggering a Task
 213 5 Attempt to trigger de-initialised instance
 214 \x1b[39m
 215 '''
 216 print(av)
 217
 218 tests = (isr_test, stop_test, reduce_test, ctor_test, launch_test, err_test)
 219 def test(n=0):
 220     try:
 221         asyncio.run(tests[n]())
 222     finally:
 223         asyncio.new_event_loop()