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
|
import time
from inspect import Parameter, signature
from typing import Callable
from unittest.mock import Mock
import pytest
from psygnal import SignalInstance, _compiled, debounced, throttled
def test_debounced() -> None:
mock1 = Mock()
f1 = debounced(mock1, timeout=10, leading=False)
f2 = Mock()
for _ in range(10):
f1()
f2()
time.sleep(0.1)
mock1.assert_called_once()
assert f2.call_count == 10
@pytest.mark.skip(reason="flaky")
def test_debounced_leading() -> None:
mock1 = Mock()
f1 = debounced(mock1, timeout=10, leading=True)
f2 = Mock()
for _ in range(10):
f1()
f2()
time.sleep(0.1)
assert mock1.call_count == 2
assert f2.call_count == 10
@pytest.mark.skip(reason="flaky")
def test_throttled() -> None:
mock1 = Mock()
f1 = throttled(mock1, timeout=10, leading=True)
f2 = Mock()
for _ in range(10):
f1()
f2()
time.sleep(0.1)
assert mock1.call_count == 2
assert f2.call_count == 10
@pytest.mark.skip(reason="flaky")
def test_throttled_trailing() -> None:
mock1 = Mock()
f1 = throttled(mock1, timeout=10, leading=False)
f2 = Mock()
for _ in range(10):
f1()
f2()
time.sleep(0.1)
assert mock1.call_count == 1
assert f2.call_count == 10
def test_cancel() -> None:
mock1 = Mock()
f1 = debounced(mock1, timeout=50, leading=False)
f1()
f1()
f1.cancel()
time.sleep(0.2)
mock1.assert_not_called()
def test_flush() -> None:
mock1 = Mock()
f1 = debounced(mock1, timeout=50, leading=False)
f1()
f1()
f1.flush()
time.sleep(0.2)
mock1.assert_called_once()
@pytest.mark.parametrize("deco", [debounced, throttled])
@pytest.mark.skip(reason="flaky")
def test_throttled_debounced_signature(deco: Callable) -> None:
mock = Mock()
@deco(timeout=0, leading=True)
def f1(x: int) -> None:
"""Doc."""
mock(x)
# make sure we can still inspect the signature
assert signature(f1).parameters["x"] == Parameter(
"x", Parameter.POSITIONAL_OR_KEYWORD, annotation=int
)
# make sure these are connectable
sig = SignalInstance((int, int, int))
sig.connect(f1)
sig.emit(1, 2, 3)
mock.assert_called_once_with(1)
if not _compiled:
# unfortunately, dynamic assignment of __doc__ and stuff isn't possible in mypyc
assert f1.__doc__ == "Doc."
assert f1.__name__ == "f1"
|
