aiosignal

1
import re
2
from unittest import mock
3

4
import pytest
5

6
from aiosignal import Signal
7

8

9
class Owner:
10
    def __repr__(self) -> str:
11
        return "<Owner 0xdeadbeef>"
12

13

14
@pytest.fixture
15
def owner() -> Owner:
16
    return Owner()
17

18

19
@pytest.mark.asyncio
20
async def test_add_signal_handler_not_a_callable(owner: Owner) -> None:
21
    callback = True
22
    signal = Signal(owner)
23
    signal.append(callback)
24
    signal.freeze()
25
    with pytest.raises(TypeError):
26
        await signal.send()
27

28

29
@pytest.mark.asyncio
30
async def test_function_signal_dispatch_kwargs(owner: Owner) -> None:
31
    signal = Signal(owner)
32
    kwargs = {"foo": 1, "bar": 2}
33

34
    callback_mock = mock.Mock()
35

36
    async def callback(**kwargs):
37
        callback_mock(**kwargs)
38

39
    signal.append(callback)
40
    signal.freeze()
41

42
    await signal.send(**kwargs)
43
    callback_mock.assert_called_once_with(**kwargs)
44

45

46
@pytest.mark.asyncio
47
async def test_function_signal_dispatch_args_kwargs(owner: Owner) -> None:
48
    signal = Signal(owner)
49
    args = {"a", "b"}
50
    kwargs = {"foo": 1, "bar": 2}
51

52
    callback_mock = mock.Mock()
53

54
    async def callback(*args, **kwargs):
55
        callback_mock(*args, **kwargs)
56

57
    signal.append(callback)
58
    signal.freeze()
59

60
    await signal.send(*args, **kwargs)
61
    callback_mock.assert_called_once_with(*args, **kwargs)
62

63

64
@pytest.mark.asyncio
65
async def test_non_coroutine(owner: Owner) -> None:
66
    signal = Signal(owner)
67
    kwargs = {"foo": 1, "bar": 2}
68

69
    callback = mock.Mock()
70

71
    signal.append(callback)
72
    signal.freeze()
73

74
    with pytest.raises(TypeError):
75
        await signal.send(**kwargs)
76

77

78
def test_setitem(owner: Owner) -> None:
79
    signal = Signal(owner)
80
    m1 = mock.Mock()
81
    signal.append(m1)
82
    assert signal[0] is m1
83
    m2 = mock.Mock()
84
    signal[0] = m2
85
    assert signal[0] is m2
86

87

88
def test_delitem(owner: Owner) -> None:
89
    signal = Signal(owner)
90
    m1 = mock.Mock()
91
    signal.append(m1)
92
    assert len(signal) == 1
93
    del signal[0]
94
    assert len(signal) == 0
95

96

97
def test_cannot_append_to_frozen_signal(owner: Owner) -> None:
98
    signal = Signal(owner)
99
    m1 = mock.Mock()
100
    m2 = mock.Mock()
101
    signal.append(m1)
102
    signal.freeze()
103
    with pytest.raises(RuntimeError):
104
        signal.append(m2)
105

106
    assert list(signal) == [m1]
107

108

109
def test_cannot_setitem_in_frozen_signal(owner: Owner) -> None:
110
    signal = Signal(owner)
111
    m1 = mock.Mock()
112
    m2 = mock.Mock()
113
    signal.append(m1)
114
    signal.freeze()
115
    with pytest.raises(RuntimeError):
116
        signal[0] = m2
117

118
    assert list(signal) == [m1]
119

120

121
def test_cannot_delitem_in_frozen_signal(owner: Owner) -> None:
122
    signal = Signal(owner)
123
    m1 = mock.Mock()
124
    signal.append(m1)
125
    signal.freeze()
126
    with pytest.raises(RuntimeError):
127
        del signal[0]
128

129
    assert list(signal) == [m1]
130

131

132
@pytest.mark.asyncio
133
async def test_cannot_send_non_frozen_signal(owner: Owner) -> None:
134
    signal = Signal(owner)
135

136
    callback_mock = mock.Mock()
137

138
    async def callback(**kwargs):
139
        callback_mock(**kwargs)
140

141
    signal.append(callback)
142

143
    with pytest.raises(RuntimeError):
144
        await signal.send()
145

146
    assert not callback_mock.called
147

148

149
def test_repr(owner: Owner) -> None:
150
    signal = Signal(owner)
151

152
    signal.append(mock.Mock(__repr__=lambda *a: "<callback>"))
153

154
    assert (
155
        re.match(
156
            r"<Signal owner=<Owner 0xdeadbeef>, frozen=False, " r"\[<callback>\]>",
157
            repr(signal),
158
        )
159
        is not None
160
    )
161