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
|
"""Request tests."""
import logging
from unittest.mock import AsyncMock, MagicMock, patch
import aiohttp
import pytest
from airos.airos8 import AirOS8
from airos.exceptions import (
AirOSConnectionAuthenticationError,
AirOSDataMissingError,
AirOSDeviceConnectionError,
)
# pylint: disable=redefined-outer-name
@pytest.fixture
def mock_session() -> MagicMock:
"""Return a mock aiohttp ClientSession."""
return MagicMock(spec=aiohttp.ClientSession)
@pytest.fixture
def mock_airos8_device(mock_session: MagicMock) -> AirOS8:
"""Return a mock AirOS instance with string host."""
return AirOS8(
host="192.168.1.3",
username="testuser",
password="testpassword",
session=mock_session,
)
@pytest.mark.asyncio
async def test_request_json_success(
mock_airos8_device: AirOS8,
mock_session: MagicMock,
) -> None:
"""Test successful JSON request."""
expected_response_data = {"key": "value"}
mock_response = AsyncMock()
mock_response.status = 200
mock_response.text = AsyncMock(return_value='{"key": "value"}')
mock_response.raise_for_status = MagicMock()
mock_session.request.return_value.__aenter__.return_value = mock_response
with patch.object(mock_airos8_device, "connected", True):
response_data = await mock_airos8_device._request_json("GET", "/test/path") # noqa: SLF001
assert response_data == expected_response_data
mock_session.request.assert_called_once()
mock_session.request.assert_called_once_with(
"GET",
"/test/path",
json=None,
data=None,
headers={},
)
@pytest.mark.asyncio
async def test_request_json_connection_error(
mock_airos8_device: AirOS8,
mock_session: MagicMock,
) -> None:
"""Test handling of a connection error."""
mock_session.request.return_value.__aenter__.side_effect = (
aiohttp.ClientConnectionError
)
with (
patch.object(mock_airos8_device, "connected", True),
pytest.raises(AirOSDeviceConnectionError),
):
await mock_airos8_device._request_json("GET", "/test/path") # noqa: SLF001
@pytest.mark.asyncio
async def test_request_json_http_error(
mock_airos8_device: AirOS8,
mock_session: MagicMock,
) -> None:
"""Test handling of a non-200 HTTP status code."""
mock_response = AsyncMock()
mock_response.status = 401
mock_response.raise_for_status = MagicMock(
side_effect=aiohttp.ClientResponseError(
request_info=MagicMock(), history=(), status=401, message="Unauthorized"
)
)
mock_response.text = AsyncMock(return_value="{}")
mock_session.request.return_value.__aenter__.return_value = mock_response
with (
patch.object(mock_airos8_device, "connected", True),
pytest.raises(AirOSConnectionAuthenticationError),
):
await mock_airos8_device._request_json("GET", "/test/path") # noqa: SLF001
mock_response.raise_for_status.assert_called_once()
@pytest.mark.asyncio
async def test_request_json_non_json_response(
mock_airos8_device: AirOS8,
mock_session: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test handling of a response that is not valid JSON."""
mock_response = AsyncMock()
mock_response.status = 200
mock_response.text = AsyncMock(return_value="NOT-A-JSON-STRING")
mock_response.raise_for_status = MagicMock()
mock_session.request.return_value.__aenter__.return_value = mock_response
with (
patch.object(mock_airos8_device, "connected", True),
pytest.raises(AirOSDataMissingError),
caplog.at_level(logging.DEBUG),
):
await mock_airos8_device._request_json("GET", "/test/path") # noqa: SLF001
assert "Failed to decode JSON from /test/path" in caplog.text
@pytest.mark.asyncio
async def test_request_json_with_params_and_data(
mock_airos8_device: AirOS8,
mock_session: MagicMock,
) -> None:
"""Test request with parameters and data."""
mock_response = AsyncMock()
mock_response.status = 200
mock_response.text = AsyncMock(return_value="{}")
mock_response.raise_for_status = MagicMock()
mock_session.request.return_value.__aenter__.return_value = mock_response
params = {"param1": "value1"}
data = {"key": "value"}
with patch.object(mock_airos8_device, "connected", True):
await mock_airos8_device._request_json( # noqa: SLF001
"POST", "/test/path", json_data=params, form_data=data
)
mock_session.request.assert_called_once_with(
"POST",
"/test/path",
json=params,
data=data,
headers={},
)
|
