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
|
# Copyright Tomer Figenblat.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Switcher integration UDP bridge module test cases."""
import socket
from asyncio import sleep
from binascii import unhexlify
from pathlib import Path
from unittest.mock import Mock, patch
from assertpy import assert_that
from pytest import fixture, mark
from aioswitcher.bridge import SwitcherBridge
pytestmark = mark.asyncio
@fixture
def mock_callback():
return Mock()
@fixture
def udp_broadcast_server():
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
server.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
server.settimeout(0.2)
return server
@patch("logging.Logger.info")
async def test_stopping_before_started_and_establishing_a_connection_should_write_to_the_info_output(mock_info, mock_callback):
bridge = SwitcherBridge(mock_callback, [1234])
assert_that(bridge.is_running).is_false()
await bridge.stop()
mock_info.assert_called_with("udp bridge on port %s not started", 1234)
async def test_bridge_operation_as_a_context_manager(unused_udp_port_factory, mock_callback):
port = unused_udp_port_factory()
async with SwitcherBridge(mock_callback, [port]) as bridge:
assert_that(bridge.is_running).is_true()
async def test_bridge_start_and_stop_operations(unused_udp_port_factory, mock_callback):
port = unused_udp_port_factory()
bridge = SwitcherBridge(mock_callback, [port])
assert_that(bridge.is_running).is_false()
await bridge.start()
assert_that(bridge.is_running).is_true()
await bridge.stop()
assert_that(bridge.is_running).is_false()
async def test_bridge_callback_loading(udp_broadcast_server, unused_udp_port_factory, mock_callback, resource_path):
port = unused_udp_port_factory()
sut_v2_off_datagram = Path(f'{resource_path}_v2_off.txt').read_text().replace('\n', '').encode()
sut_power_plug_off_datagram = Path(f'{resource_path}_power_plug_off.txt').read_text().replace('\n', '').encode()
async with SwitcherBridge(mock_callback, [port]):
udp_broadcast_server.sendto(unhexlify(sut_v2_off_datagram), ("localhost", port))
await sleep(0.2)
udp_broadcast_server.sendto(unhexlify(sut_power_plug_off_datagram), ("localhost", port))
await sleep(0.2)
assert_that(mock_callback.call_count).is_equal_to(2)
|
