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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
"""Tests for common module."""
import re
from unittest.mock import MagicMock, patch
import pytest
from aiohttp import BasicAuth, ClientError, ClientSession
from aioresponses import aioresponses
from bleak.backends.device import BLEDevice
from yarl import URL
from aioshelly.common import (
ConnectionOptions,
get_info,
is_firmware_supported,
process_ip_or_options,
)
from aioshelly.const import DEFAULT_HTTP_PORT
from aioshelly.exceptions import (
DeviceConnectionError,
DeviceConnectionTimeoutError,
InvalidHostError,
MacAddressMismatchError,
)
from .rpc_device import load_device_fixture
@pytest.mark.parametrize(
("gen", "model", "firmware_version", "expected"),
[
(5, "XYZ-G5", "20250913-112054/v1.0.0-gcb84623", False),
(4, "XYZ-G4", "20240913-112054/v1.0.0-gcb84623", True),
(1, "SHSW-44", "20230913-112054/v1.14.0-gcb84623", False),
(1, "SHSW-1", "20230913-112054/v1.14.0-gcb84623", True),
(2, "SNDC-0D4P10WW", "20230703-112054/0.99.0-gcb84623", False),
(3, "UNKNOWN", "20240819-074343/1.4.20-gc2639da", True),
(3, "S3SW-002P16EU", "strange-firmware-version", False),
],
)
def test_is_firmware_supported(
gen: int, model: str, firmware_version: str, expected: bool
) -> None:
"""Test is_firmware_supported function."""
assert is_firmware_supported(gen, model, firmware_version) is expected
@pytest.mark.asyncio
async def test_process_ip_or_options() -> None:
"""Test process_ip_or_options function."""
ip = "192.168.20.11"
# Test string numeric IP address
assert await process_ip_or_options(ip) == ConnectionOptions(ip)
# Test string hostname IP address
with patch("aioshelly.common.gethostbyname", return_value=ip):
assert await process_ip_or_options("some_host") == ConnectionOptions(ip)
# Test ConnectionOptions
options = ConnectionOptions(ip, "user", "pass")
assert await process_ip_or_options(options) == options
assert options.auth == BasicAuth("user", "pass")
# Test missing password
with pytest.raises(ValueError, match="Supply both username and password"):
options = ConnectionOptions(ip, "user")
@pytest.mark.asyncio
async def test_get_info() -> None:
"""Test get_info function."""
mock_response = await load_device_fixture("shellyplus2pm", "shelly.json")
ip_address = "10.10.10.10"
session = ClientSession()
with aioresponses() as session_mock:
session_mock.get(
URL.build(
scheme="http", host=ip_address, port=DEFAULT_HTTP_PORT, path="/shelly"
),
payload=mock_response,
)
result = await get_info(session, ip_address, "AABBCCDDEEFF")
await session.close()
assert result == mock_response
@pytest.mark.asyncio
async def test_get_info_mac_mismatch() -> None:
"""Test get_info function with MAC mismatch."""
mock_response = await load_device_fixture("shellyplus2pm", "shelly.json")
ip_address = "10.10.10.10"
session = ClientSession()
with aioresponses() as session_mock:
session_mock.get(
URL.build(
scheme="http", host=ip_address, port=DEFAULT_HTTP_PORT, path="/shelly"
),
payload=mock_response,
)
with pytest.raises(
MacAddressMismatchError,
match="Input MAC: 112233445566, Shelly MAC: AABBCCDDEEFF",
):
await get_info(session, ip_address, "112233445566")
await session.close()
@pytest.mark.parametrize(
("exc", "expected_exc"),
[
(TimeoutError, DeviceConnectionTimeoutError),
(ClientError, DeviceConnectionError),
(OSError, DeviceConnectionError),
],
)
@pytest.mark.asyncio
async def test_get_info_exc(exc: Exception, expected_exc: Exception) -> None:
"""Test get_info function with exception."""
ip_address = "10.10.10.10"
session = ClientSession()
with aioresponses() as session_mock:
session_mock.get(
URL.build(
scheme="http", host=ip_address, port=DEFAULT_HTTP_PORT, path="/shelly"
),
exception=exc,
)
with pytest.raises(expected_exc):
await get_info(session, ip_address, "AABBCCDDEEFF")
await session.close()
@pytest.mark.asyncio
async def test_get_info_invalid_error() -> None:
"""Test get_info function with an invalid host exception."""
session = ClientSession()
with pytest.raises(
InvalidHostError,
match=re.escape("Host 'http://10.10.10.10' cannot contain ':'"),
):
await get_info(session, "http://10.10.10.10", "AABBCCDDEEFF")
await session.close()
def test_connection_options_missing_both() -> None:
"""Test ConnectionOptions with neither ip_address nor ble_device."""
with pytest.raises(
ValueError, match="Must provide either ip_address or ble_device"
):
ConnectionOptions()
def test_connection_options_both_provided() -> None:
"""Test ConnectionOptions with both ip_address and ble_device."""
ble_device = MagicMock(spec=BLEDevice)
with pytest.raises(
ValueError, match="Cannot provide both ip_address and ble_device"
):
ConnectionOptions(ip_address="192.168.1.1", ble_device=ble_device)
|
