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
147
148
149
150
151
152
153
154
155
156
157
158
159
|
import asyncio
import logging
import weakref
import pytest
from trame_server.controller import FunctionNotImplementedError
logger = logging.getLogger(__name__)
def test_define_later(controller):
f = controller.func
with pytest.raises(FunctionNotImplementedError):
f()
controller.func = lambda: 3
assert f() == 3
def test_trigger_name(controller):
def fn_1(x):
return x * 2
def fn_2(x):
return x * 3
a_name = controller.trigger_name(fn_1)
b_name = controller.trigger_name(fn_2)
a_name_next = controller.trigger_name(fn_1)
fn_1_r = controller.trigger_fn(a_name)
fn_2_r = controller.trigger_fn(b_name)
assert a_name != b_name
assert a_name == a_name_next
assert a_name == "trigger__1"
assert fn_1 is fn_1_r
assert fn_2 is fn_2_r
def test_composition(controller):
def fn():
return 1
@controller.add("func_attr")
def fn_1():
return 1.5
assert controller.func_attr() == [1.5]
@controller.add("func_attr", clear=True)
def fn_2():
return 2
@controller.once("func_attr")
def fn_3():
return 3
# get
f_attr = controller.func
f_item = controller["func"]
assert f_attr is f_item
# set
controller.func_attr = fn
controller["func_item"] = fn
assert controller.func_attr() == [1, 2, 3]
assert controller.func_attr() == [1, 2]
assert controller.func_attr() == [1, 2]
# invalid set
with pytest.raises(NameError):
controller.trigger = fn
def test_weakrefs(controller):
class Obj:
method_call_count = 0
destructor_call_count = 0
def __del__(self):
Obj.destructor_call_count += 1
def fn(self):
Obj.method_call_count += 1
print("Obj.fn called")
return 1
o = Obj()
controller.func.add(weakref.WeakMethod(o.fn))
@controller.add("func")
def fn_1():
return 1.5
controller.func()
assert Obj.method_call_count == 1
del o
assert Obj.destructor_call_count == 1
controller.func()
assert Obj.method_call_count == 1
@pytest.mark.asyncio
async def test_tasks(controller):
@controller.add("async_fn")
def sync_fn_add():
return 1
@controller.add_task("async_fn", clear=True)
async def async_fn():
await asyncio.sleep(0.01)
return 2
@controller.add("async_fn")
def sync_fn_add_2():
return 4
@controller.set("async_fn")
def set_fn():
return 5
result = controller.async_fn()
assert len(result) == 3
assert result[0] == 5
assert result[1] == 4
assert await result[2] == 2
result = controller.async_fn()
assert len(result) == 3
with pytest.raises(KeyError):
controller.async_fn.remove(async_fn)
controller.async_fn.remove_task(async_fn)
result = controller.async_fn()
assert len(result) == 2
controller.async_fn.discard(async_fn) # no error if missing
controller.async_fn.discard(sync_fn_add_2)
assert controller.async_fn() == 5
@controller.set("async_fn", clear=True)
def set_fn_2():
return 10
assert controller.async_fn() == 10
assert controller.async_fn.exists()
controller.async_fn.clear()
assert not controller.async_fn.exists()
|
