-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_runtime.py
More file actions
146 lines (100 loc) · 3.79 KB
/
test_runtime.py
File metadata and controls
146 lines (100 loc) · 3.79 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
"""Unit tests for the pn.runtime asyncio runtime."""
from __future__ import annotations
import asyncio
import threading
import time
import pytest
from pythonnative.runtime import (
call_on_main_thread,
call_threadsafe,
create_future,
get_loop,
reject_future,
resolve_future,
run_async,
)
def test_loop_is_singleton_and_running() -> None:
loop_a = get_loop()
loop_b = get_loop()
assert loop_a is loop_b
assert loop_a.is_running()
def test_loop_runs_on_dedicated_thread() -> None:
loop = get_loop()
main_thread = threading.current_thread()
async def grab_thread() -> str:
return threading.current_thread().name
name = asyncio.run_coroutine_threadsafe(grab_thread(), loop).result(timeout=2.0)
assert name == "pn-asyncio"
assert threading.current_thread() is main_thread # we never crossed onto the loop thread
def test_run_async_returns_thread_future_with_result() -> None:
async def work() -> int:
await asyncio.sleep(0)
return 7
future = run_async(work())
assert future.result(timeout=2.0) == 7
def test_run_async_propagates_exceptions() -> None:
async def explode() -> None:
raise ValueError("nope")
future = run_async(explode())
with pytest.raises(ValueError, match="nope"):
future.result(timeout=2.0)
def test_call_threadsafe_dispatches_callback() -> None:
received: list = []
done = threading.Event()
def callback(x: int) -> None:
received.append(x)
done.set()
call_threadsafe(callback, 42)
assert done.wait(2.0)
assert received == [42]
def test_resolve_future_from_any_thread() -> None:
future = create_future()
def deliver() -> None:
time.sleep(0.05)
resolve_future(future, "hi")
threading.Thread(target=deliver, daemon=True).start()
async def wait() -> str:
return await future
assert run_async(wait()).result(timeout=2.0) == "hi"
def test_reject_future_from_any_thread() -> None:
future = create_future()
def deliver() -> None:
time.sleep(0.05)
reject_future(future, RuntimeError("bad"))
threading.Thread(target=deliver, daemon=True).start()
async def wait() -> str:
await future
return "unreached"
with pytest.raises(RuntimeError, match="bad"):
run_async(wait()).result(timeout=2.0)
def test_resolve_after_done_is_noop() -> None:
future = create_future()
resolve_future(future, 1)
# Second resolve must not raise InvalidStateError.
resolve_future(future, 2)
async def wait() -> int:
return await future
# Race-free wait for the first resolution to land via call_soon_threadsafe.
assert run_async(wait()).result(timeout=2.0) == 1
def test_call_on_main_thread_runs_inline_off_device() -> None:
"""Off-device the helper has no platform main loop to marshal onto;
it should just invoke ``fn`` synchronously on the caller's thread."""
received: list = []
caller_thread = threading.current_thread()
def fn() -> None:
received.append(threading.current_thread())
call_on_main_thread(fn)
assert received == [caller_thread]
def test_call_on_main_thread_bridges_coroutine_to_caller_thread() -> None:
"""Off-device, ``call_on_main_thread`` from inside a coroutine runs
``fn`` on the asyncio loop thread (the only "main-like" thread we
have in tests). The future round-trip mirrors the iOS / Android
flow: coroutine → main → resolve_future → coroutine."""
async def confirm() -> str:
future = create_future()
def on_main() -> None:
resolve_future(future, threading.current_thread().name)
call_on_main_thread(on_main)
return await future
name = run_async(confirm()).result(timeout=2.0)
assert name == "pn-asyncio"