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
|
from unittest.mock import Mock, patch
import pytest
from grpclib.events import _Event, _EventMeta, _ident, listen
from grpclib.events import _Dispatch, _DispatchMeta, _dispatches
def _event_eq(self, other):
if not isinstance(other, type(self)):
return False
for name in self.__slots__:
if getattr(self, name) != getattr(other, name):
return False
return True
def patch_event():
return patch.object(_Event, '__eq__', _event_eq)
def mock_callback(func):
return Mock(side_effect=func)
def test_empty_event():
class MyEvent(_Event, metaclass=_EventMeta):
pass
assert MyEvent.__slots__ == ()
assert MyEvent.__readonly__ == frozenset()
assert MyEvent.__payload__ == ()
event = MyEvent()
with pytest.raises(AttributeError, match='has no attribute'):
event.foo = 5
def test_frozen_event():
class MyEvent(_Event, metaclass=_EventMeta):
a: int
b: str
assert set(MyEvent.__slots__) == {'a', 'b'}
assert MyEvent.__readonly__ == frozenset(('a', 'b'))
assert MyEvent.__payload__ == ()
event = MyEvent(a=1, b='2')
assert event.a == 1
assert event.b == '2'
with pytest.raises(AttributeError, match='^Read-only'):
event.a = 5
def test_mixed_event():
class MyEvent(_Event, metaclass=_EventMeta):
__payload__ = ('a', 'b')
a: int
b: str
c: int
d: str
assert set(MyEvent.__slots__) == {'a', 'b', 'c', 'd'}
assert MyEvent.__readonly__ == frozenset(('c', 'd'))
assert MyEvent.__payload__ == ('a', 'b')
event = MyEvent(a=1, b='2', c=3, d='4')
assert event.a == 1
assert event.b == '2'
assert event.c == 3
assert event.d == '4'
event.a = 5
event.b = '6'
assert event.a == 5
assert event.b == '6'
with pytest.raises(AttributeError, match='^Read-only'):
event.c = 7
with pytest.raises(AttributeError, match='^Read-only'):
event.d = '8'
@pytest.mark.asyncio
async def test_dispatch():
class MyEvent(_Event, metaclass=_EventMeta):
__payload__ = ('a', 'b')
a: int
b: str
c: int
d: str
class MyDispatch(_Dispatch, metaclass=_DispatchMeta):
@_dispatches(MyEvent)
async def my_event(self, a, b, *, c, d):
return await self.__dispatch__(MyEvent(a=a, b=b, c=c, d=d))
class Target:
__dispatch__ = MyDispatch()
assert Target.__dispatch__.my_event is _ident
assert await Target.__dispatch__.my_event(1, 3, c=6, d=9) == (1, 3)
@mock_callback
async def callback(event: MyEvent):
assert event.a == 2
assert event.b == 4
assert event.c == 8
assert event.d == 16
listen(Target, MyEvent, callback)
assert Target.__dispatch__.my_event is not _ident
assert await Target.__dispatch__.my_event(2, 4, c=8, d=16) == (2, 4)
with patch_event():
callback.assert_called_once_with(MyEvent(a=2, b=4, c=8, d=16))
@pytest.mark.asyncio
async def test_interrupt():
class MyEvent(_Event, metaclass=_EventMeta):
payload: int
class MyDispatch(_Dispatch, metaclass=_DispatchMeta):
@_dispatches(MyEvent)
async def my_event(self, *, payload):
return await self.__dispatch__(MyEvent(payload=payload))
class Target:
__dispatch__ = MyDispatch()
@mock_callback
async def cb1(_: MyEvent):
pass
@mock_callback
async def cb2(event: MyEvent):
event.interrupt()
@mock_callback
async def cb3(_: MyEvent):
pass
listen(Target, MyEvent, cb1)
listen(Target, MyEvent, cb2)
listen(Target, MyEvent, cb3)
assert await Target.__dispatch__.my_event(payload=42) == ()
cb1.assert_called_once()
cb2.assert_called_once()
assert not cb3.called
|
