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