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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
"""Ubiquiti AirOS8 tests."""
from http.cookies import SimpleCookie
import json
import os
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import aiofiles
from mashumaro.exceptions import MissingField
import pytest
from airos.airos8 import AirOS8
from airos.data import AirOS8Data, Wireless
from airos.exceptions import AirOSDeviceConnectionError, AirOSKeyDataMissingError
async def _read_fixture(fixture: str = "loco5ac_ap-ptp") -> Any:
"""Read fixture file per device type."""
fixture_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures", "userdata")
path = os.path.join(fixture_dir, f"{fixture}.json")
try:
async with aiofiles.open(path, encoding="utf-8") as f:
return json.loads(await f.read())
except FileNotFoundError:
pytest.fail(f"Fixture file not found: {path}")
except json.JSONDecodeError as e:
pytest.fail(f"Invalid JSON in fixture file {path}: {e}")
@pytest.mark.skip(reason="broken, needs investigation")
@patch("airos.airos8._LOGGER")
@pytest.mark.asyncio
async def test_status_logs_redacted_data_on_invalid_value(
mock_logger: MagicMock, airos8_device: AirOS8
) -> None:
"""Test that the status method correctly logs redacted data when it encounters an InvalidFieldValue during deserialization."""
# --- Prepare fake POST /api/auth response with cookies ---
cookie = SimpleCookie()
cookie["session_id"] = "test-cookie"
cookie["AIROS_TOKEN"] = "abc123"
mock_login_response = MagicMock()
mock_login_response.__aenter__.return_value = mock_login_response
mock_login_response.text = AsyncMock(return_value="{}")
mock_login_response.status = 200
mock_login_response.cookies = cookie
mock_login_response.headers = {"X-CSRF-ID": "test-csrf-token"}
# --- Prepare a response with data that would be redacted ---
fixture_data = await _read_fixture("mocked_invalid_wireless_mode")
mock_status_response = MagicMock()
mock_status_response.__aenter__.return_value = mock_status_response
mock_status_response.text = AsyncMock(return_value=json.dumps(fixture_data))
mock_status_response.status = 200
mock_status_response.json = AsyncMock(return_value=fixture_data)
# --- Patch `from_dict` to force the desired exception ---
# We use a valid fixture response, but force the exception to be a MissingField
with (
patch.object(airos8_device.session, "post", return_value=mock_login_response),
patch.object(airos8_device.session, "get", return_value=mock_status_response),
patch(
"airos.airos8.AirOSData.from_dict",
side_effect=MissingField(
field_name="wireless", field_type=Wireless, holder_class=AirOS8Data
),
),
):
await airos8_device.login()
with pytest.raises(AirOSKeyDataMissingError):
await airos8_device.status()
# --- Assertions for the logging and redaction ---
assert mock_logger.exception.called
assert mock_logger.exception.call_count == 1
assert mock_logger.error.called is False
# Get the dictionary that was passed as the second argument to the logger
logged_data = mock_logger.exception.call_args[0][1]
# Assert that the dictionary has been redacted
assert "wireless" in logged_data
assert "essid" in logged_data["wireless"]
assert logged_data["wireless"]["essid"] == "REDACTED"
assert "host" in logged_data
assert "hostname" in logged_data["host"]
assert logged_data["host"]["hostname"] == "REDACTED"
assert "apmac" in logged_data["wireless"]
assert logged_data["wireless"]["apmac"] == "00:11:22:33:89:AB"
assert "interfaces" in logged_data
assert len(logged_data["interfaces"]) > 2
assert "status" in logged_data["interfaces"][2]
assert "ipaddr" in logged_data["interfaces"][2]["status"]
assert logged_data["interfaces"][2]["status"]["ipaddr"] == "127.0.0.3"
@pytest.mark.skip(reason="broken, needs investigation")
@patch("airos.airos8._LOGGER")
@pytest.mark.asyncio
async def test_status_logs_exception_on_missing_field(
mock_logger: MagicMock, airos8_device: AirOS8
) -> None:
"""Test that the status method correctly logs a full exception when it encounters a MissingField during deserialization."""
# --- Prepare fake POST /api/auth response with cookies ---
cookie = SimpleCookie()
cookie["session_id"] = "test-cookie"
cookie["AIROS_TOKEN"] = "abc123"
mock_login_response = MagicMock()
mock_login_response.__aenter__.return_value = mock_login_response
mock_login_response.text = AsyncMock(return_value="{}")
mock_login_response.status = 200
mock_login_response.cookies = cookie
mock_login_response.headers = {"X-CSRF-ID": "test-csrf-token"}
# --- Prepare fake GET /api/status response with the missing field fixture ---
mock_status_response = MagicMock()
mock_status_response.__aenter__.return_value = mock_status_response
mock_status_response.status = 500 # Non-200 status
mock_status_response.text = AsyncMock(return_value="Error")
mock_status_response.json = AsyncMock(return_value={})
with (
patch.object(
airos8_device.session,
"request",
side_effect=[mock_login_response, mock_status_response],
),
):
await airos8_device.login()
with pytest.raises(AirOSDeviceConnectionError):
await airos8_device.status()
# Assert the logger was called correctly
assert mock_logger.error.called
assert mock_logger.error.call_count == 1
log_args = mock_logger.error.call_args[0]
assert log_args[0] == "API call to %s failed with status %d: %s"
assert log_args[2] == 500
assert log_args[3] == "Error"
@pytest.mark.parametrize(
("mode", "fixture", "sku"),
[
("ap-ptp", "loco5ac_ap-ptp", "Loco5AC"),
("ap-ptp", "nanostation_ap-ptp_8718_missing_gps", "Loco5AC"),
("sta-ptp", "loco5ac_sta-ptp", "Loco5AC"),
("sta-ptmp", "mocked_sta-ptmp", "UNKNOWN"),
("ap-ptmp", "liteapgps_ap_ptmp_40mhz", "LAP-GPS"),
("sta-ptmp", "nanobeam5ac_sta_ptmp_40mhz", "NBE-5AC-GEN2"),
("ap-ptmp", "NanoBeam_5AC_ap-ptmp_v8.7.18", "NBE-5AC-GEN2"),
],
)
@pytest.mark.asyncio
async def test_ap_object(
airos8_device: AirOS8, base_url: str, mode: str, fixture: str, sku: str
) -> None:
"""Test device operation using the new _request_json method."""
fixture_data = await _read_fixture(fixture)
# Create an async mock that can return different values for different calls
mock_request_json = AsyncMock(
side_effect=[
{}, # First call for login()
fixture_data, # Second call for status()
]
)
with (
# Patch the internal method, not the session object
patch.object(airos8_device, "_request_json", new=mock_request_json),
# You need to manually set the connected state since login() is mocked
patch.object(airos8_device, "connected", True),
):
# We don't need to patch the session directly anymore
await airos8_device.login()
status: AirOS8Data = await airos8_device.status()
# Assertions remain the same as they check the final result
assert status.wireless.mode
assert status.wireless.mode.value == mode
assert status.derived.sku == sku
assert status.derived.mac_interface == "br0"
cookie = SimpleCookie()
cookie["session_id"] = "test-cookie"
cookie["AIROS_TOKEN"] = "abc123"
@pytest.mark.skip(reason="broken, needs investigation")
@pytest.mark.asyncio
async def test_reconnect(airos8_device: AirOS8, base_url: str) -> None:
"""Test reconnect client."""
# --- Prepare fake POST /api/stakick response ---
mock_stakick_response = MagicMock()
mock_stakick_response.__aenter__.return_value = mock_stakick_response
mock_stakick_response.status = 200
mock_stakick_response.text = AsyncMock()
mock_stakick_response.text.return_value = ""
with (
patch.object(
airos8_device.session, "request", return_value=mock_stakick_response
),
patch.object(airos8_device, "connected", True),
):
assert await airos8_device.stakick("01:23:45:67:89:aB")
|
