pysmartapp/tests/test_dispatch.py at master · pySmartThings/pysmartapp · GitHub

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
"""Define tests for the Dispatch module."""

import asyncio
import functools

import pytest

from pysmartapp.dispatch import Dispatcher


class TestDispatcher:
    """Define tests for the dispatcher class."""

    @staticmethod
    @pytest.mark.asyncio
    async def test_connect(handler):
        """Tests the connect function."""
        # Arrange
        dispatcher = Dispatcher()
        # Act
        dispatcher.connect('TEST', handler)
        # Assert
        assert handler in dispatcher.signals['TEST']

    @staticmethod
    @pytest.mark.asyncio
    async def test_disconnect(handler):
        """Tests the disconnect function."""
        # Arrange
        dispatcher = Dispatcher()
        disconnect = dispatcher.connect('TEST', handler)
        # Act
        disconnect()
        # Assert
        assert handler not in dispatcher.signals['TEST']

    @staticmethod
    @pytest.mark.asyncio
    async def test_disconnect_all(handler):
        """Tests the disconnect all function."""
        # Arrange
        dispatcher = Dispatcher()
        dispatcher.connect('TEST', handler)
        dispatcher.connect('TEST', handler)
        dispatcher.connect('TEST2', handler)
        dispatcher.connect('TEST3', handler)
        # Act
        dispatcher.disconnect_all()
        # Assert
        assert handler not in dispatcher.signals['TEST']
        assert handler not in dispatcher.signals['TEST2']
        assert handler not in dispatcher.signals['TEST3']

    @staticmethod
    @pytest.mark.asyncio
    async def test_already_disconnected(handler):
        """Tests that disconnect can be called more than once."""
        # Arrange
        dispatcher = Dispatcher()
        disconnect = dispatcher.connect('TEST', handler)
        disconnect()
        # Act
        disconnect()
        # Assert
        assert handler not in dispatcher.signals['TEST']

    @staticmethod
    @pytest.mark.asyncio
    async def test_send_async_handler(async_handler):
        """Tests sending to async handlers."""
        # Arrange
        dispatcher = Dispatcher()
        dispatcher.connect('TEST', async_handler)
        # Act
        await asyncio.gather(*dispatcher.send('TEST'))
        # Assert
        assert async_handler.fired

    @staticmethod
    @pytest.mark.asyncio
    async def test_send_async_partial_handler(async_handler):
        """Tests sending to async handlers."""
        # Arrange
        partial = functools.partial(async_handler)
        dispatcher = Dispatcher()
        dispatcher.connect('TEST', partial)
        # Act
        await asyncio.gather(*dispatcher.send('TEST'))
        # Assert
        assert async_handler.fired

    @staticmethod
    @pytest.mark.asyncio
    async def test_send(handler):
        """Tests sending to async handlers."""
        # Arrange
        dispatcher = Dispatcher()
        dispatcher.connect('TEST', handler)
        args = object()
        # Act
        await asyncio.gather(*dispatcher.send('TEST', args))
        # Assert
        assert handler.fired
        assert handler.args[0] == args

    @staticmethod
    @pytest.mark.asyncio
    async def test_custom_connect_and_send(handler):
        """Tests using the custom connect and send implementations."""
        # Arrange
        test_signal = 'PREFIX_TEST'
        stored_target = None

        def connect(signal, target):
            assert signal == test_signal
            nonlocal stored_target
            stored_target = target

            def disconnect():
                nonlocal stored_target
                stored_target = None
            return disconnect

        def send(signal, *args):
            assert signal == test_signal
            stored_target(*args)  # pylint:disable=not-callable

        dispatcher = Dispatcher(connect=connect, send=send,
                                signal_prefix='PREFIX_')

        # Act
        dispatcher.connect('TEST', handler)
        dispatcher.send('TEST')
        # Assert
        assert handler.fired