1from math import inf 2import time 3 4import pytest 5 6from trio import sleep 7from ... import _core 8from .. import wait_all_tasks_blocked 9from .._mock_clock import MockClock 10from .tutil import slow 11 12 13def test_mock_clock(): 14 REAL_NOW = 123.0 15 c = MockClock() 16 c._real_clock = lambda: REAL_NOW 17 repr(c) # smoke test 18 assert c.rate == 0 19 assert c.current_time() == 0 20 c.jump(1.2) 21 assert c.current_time() == 1.2 22 with pytest.raises(ValueError): 23 c.jump(-1) 24 assert c.current_time() == 1.2 25 assert c.deadline_to_sleep_time(1.1) == 0 26 assert c.deadline_to_sleep_time(1.2) == 0 27 assert c.deadline_to_sleep_time(1.3) > 999999 28 29 with pytest.raises(ValueError): 30 c.rate = -1 31 assert c.rate == 0 32 33 c.rate = 2 34 assert c.current_time() == 1.2 35 REAL_NOW += 1 36 assert c.current_time() == 3.2 37 assert c.deadline_to_sleep_time(3.1) == 0 38 assert c.deadline_to_sleep_time(3.2) == 0 39 assert c.deadline_to_sleep_time(4.2) == 0.5 40 41 c.rate = 0.5 42 assert c.current_time() == 3.2 43 assert c.deadline_to_sleep_time(3.1) == 0 44 assert c.deadline_to_sleep_time(3.2) == 0 45 assert c.deadline_to_sleep_time(4.2) == 2.0 46 47 c.jump(0.8) 48 assert c.current_time() == 4.0 49 REAL_NOW += 1 50 assert c.current_time() == 4.5 51 52 c2 = MockClock(rate=3) 53 assert c2.rate == 3 54 assert c2.current_time() < 10 55 56 57async def test_mock_clock_autojump(mock_clock): 58 assert mock_clock.autojump_threshold == inf 59 60 mock_clock.autojump_threshold = 0 61 assert mock_clock.autojump_threshold == 0 62 63 real_start = time.perf_counter() 64 65 virtual_start = _core.current_time() 66 for i in range(10): 67 print("sleeping {} seconds".format(10 * i)) 68 await sleep(10 * i) 69 print("woke up!") 70 assert virtual_start + 10 * i == _core.current_time() 71 virtual_start = _core.current_time() 72 73 real_duration = time.perf_counter() - real_start 74 print("Slept {} seconds in {} seconds".format(10 * sum(range(10)), real_duration)) 75 assert real_duration < 1 76 77 mock_clock.autojump_threshold = 0.02 78 t = _core.current_time() 79 # this should wake up before the autojump threshold triggers, so time 80 # shouldn't change 81 await wait_all_tasks_blocked() 82 assert t == _core.current_time() 83 # this should too 84 await wait_all_tasks_blocked(0.01) 85 assert t == _core.current_time() 86 87 # set up a situation where the autojump task is blocked for a long long 88 # time, to make sure that cancel-and-adjust-threshold logic is working 89 mock_clock.autojump_threshold = 10000 90 await wait_all_tasks_blocked() 91 mock_clock.autojump_threshold = 0 92 # if the above line didn't take affect immediately, then this would be 93 # bad: 94 await sleep(100000) 95 96 97async def test_mock_clock_autojump_interference(mock_clock): 98 mock_clock.autojump_threshold = 0.02 99 100 mock_clock2 = MockClock() 101 # messing with the autojump threshold of a clock that isn't actually 102 # installed in the run loop shouldn't do anything. 103 mock_clock2.autojump_threshold = 0.01 104 105 # if the autojump_threshold of 0.01 were in effect, then the next line 106 # would block forever, as the autojump task kept waking up to try to 107 # jump the clock. 108 await wait_all_tasks_blocked(0.015) 109 110 # but the 0.02 limit does apply 111 await sleep(100000) 112 113 114def test_mock_clock_autojump_preset(): 115 # Check that we can set the autojump_threshold before the clock is 116 # actually in use, and it gets picked up 117 mock_clock = MockClock(autojump_threshold=0.1) 118 mock_clock.autojump_threshold = 0.01 119 real_start = time.perf_counter() 120 _core.run(sleep, 10000, clock=mock_clock) 121 assert time.perf_counter() - real_start < 1 122 123 124async def test_mock_clock_autojump_0_and_wait_all_tasks_blocked_0(mock_clock): 125 # Checks that autojump_threshold=0 doesn't interfere with 126 # calling wait_all_tasks_blocked with the default cushion=0. 127 128 mock_clock.autojump_threshold = 0 129 130 record = [] 131 132 async def sleeper(): 133 await sleep(100) 134 record.append("yawn") 135 136 async def waiter(): 137 await wait_all_tasks_blocked() 138 record.append("waiter woke") 139 await sleep(1000) 140 record.append("waiter done") 141 142 async with _core.open_nursery() as nursery: 143 nursery.start_soon(sleeper) 144 nursery.start_soon(waiter) 145 146 assert record == ["waiter woke", "yawn", "waiter done"] 147 148 149@slow 150async def test_mock_clock_autojump_0_and_wait_all_tasks_blocked_nonzero(mock_clock): 151 # Checks that autojump_threshold=0 doesn't interfere with 152 # calling wait_all_tasks_blocked with a non-zero cushion. 153 154 mock_clock.autojump_threshold = 0 155 156 record = [] 157 158 async def sleeper(): 159 await sleep(100) 160 record.append("yawn") 161 162 async def waiter(): 163 await wait_all_tasks_blocked(1) 164 record.append("waiter done") 165 166 async with _core.open_nursery() as nursery: 167 nursery.start_soon(sleeper) 168 nursery.start_soon(waiter) 169 170 assert record == ["waiter done", "yawn"] 171