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
|
import asyncio
from unittest.mock import call
import pytest
import zigpy_znp.types as t
import zigpy_znp.commands as c
from zigpy_znp.api import OneShotResponseListener, CallbackResponseListener
async def test_resolve(mocker):
callback = mocker.Mock()
callback_listener = CallbackResponseListener(
[c.SYS.Ping.Rsp(partial=True)], callback
)
future = asyncio.get_running_loop().create_future()
one_shot_listener = OneShotResponseListener([c.SYS.Ping.Rsp(partial=True)], future)
match = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
no_match = c.SYS.OSALNVWrite.Rsp(Status=t.Status.SUCCESS)
assert callback_listener.resolve(match)
assert not callback_listener.resolve(no_match)
assert callback_listener.resolve(match)
assert not callback_listener.resolve(no_match)
assert one_shot_listener.resolve(match)
assert not one_shot_listener.resolve(no_match)
callback.assert_has_calls([call(match), call(match)])
assert callback.call_count == 2
assert (await future) == match
# Cancelling a callback will have no effect
assert not callback_listener.cancel()
# Cancelling a one-shot listener does not throw any errors
assert one_shot_listener.cancel()
assert one_shot_listener.cancel()
assert one_shot_listener.cancel()
async def test_cancel():
# Cancelling a one-shot listener prevents it from being fired
future = asyncio.get_running_loop().create_future()
one_shot_listener = OneShotResponseListener([c.SYS.Ping.Rsp(partial=True)], future)
one_shot_listener.cancel()
match = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
assert not one_shot_listener.resolve(match)
with pytest.raises(asyncio.CancelledError):
await future
async def test_multi_cancel(mocker):
callback = mocker.Mock()
callback_listener = CallbackResponseListener(
[c.SYS.Ping.Rsp(partial=True)], callback
)
future = asyncio.get_running_loop().create_future()
one_shot_listener = OneShotResponseListener([c.SYS.Ping.Rsp(partial=True)], future)
match = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
no_match = c.SYS.OSALNVWrite.Rsp(Status=t.Status.SUCCESS)
assert callback_listener.resolve(match)
assert not callback_listener.resolve(no_match)
assert one_shot_listener.resolve(match)
assert not one_shot_listener.resolve(no_match)
callback.assert_called_once_with(match)
assert (await future) == match
async def test_api_cancel_listeners(connected_znp, mocker):
znp, znp_server = connected_znp
callback = mocker.Mock()
znp.callback_for_response(
c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS), callback
)
future = znp.wait_for_responses(
[
c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS),
c.SYS.OSALNVWrite.Rsp(Status=t.Status.SUCCESS),
]
)
assert not future.done()
await znp.disconnect()
with pytest.raises(asyncio.CancelledError):
await future
# add_done_callback won't be executed immediately
await asyncio.sleep(0.1)
assert len(znp._listeners) == 0
|
