diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ffeb20..8db5530 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ All notable changes to this project will be documented in this file. +## [0.3.0] - 2025-08-15 + +### Added + +- Implementation of `[AP|Sta]-[MODE]` to Enums. +- Added update check (non-forced) endpoint +- Added warnings fetch endpoint + ## [0.2.11] - 2025-08-14 ### Changed diff --git a/airos/airos8.py b/airos/airos8.py index 18beeae..3ff65f0 100644 --- a/airos/airos8.py +++ b/airos/airos8.py @@ -11,7 +11,12 @@ import aiohttp from mashumaro.exceptions import InvalidFieldValue, MissingField -from .data import AirOS8Data as AirOSData, redact_data_smart +from .data import ( + AirOS8Data as AirOSData, + DerivedWirelessMode, + DerivedWirelessRole, + redact_data_smart, +) from .exceptions import ( AirOSConnectionAuthenticationError, AirOSConnectionSetupError, @@ -54,6 +59,8 @@ def __init__( self._status_cgi_url = f"{self.base_url}/status.cgi" # AirOS 8 self._stakick_cgi_url = f"{self.base_url}/stakick.cgi" # AirOS 8 self._provmode_url = f"{self.base_url}/api/provmode" # AirOS 8 + self._warnings_url = f"{self.base_url}/api/warnings" # AirOS 8 + self._update_check_url = f"{self.base_url}/api/fw/update-check" # AirOS 8 self.current_csrf_token: str | None = None self._use_json_for_login_post = False @@ -201,6 +208,8 @@ def derived_data(response: dict[str, Any]) -> dict[str, Any]: "access_point": False, "ptp": False, "ptmp": False, + "role": DerivedWirelessRole.STATION, + "mode": DerivedWirelessMode.PTP, } # Access Point / Station vs PTP/PtMP @@ -209,12 +218,16 @@ def derived_data(response: dict[str, Any]) -> dict[str, Any]: case "ap-ptmp": derived["access_point"] = True derived["ptmp"] = True + derived["role"] = DerivedWirelessRole.ACCESS_POINT + derived["mode"] = DerivedWirelessMode.PTMP case "sta-ptmp": derived["station"] = True derived["ptmp"] = True + derived["mode"] = DerivedWirelessMode.PTMP case "ap-ptp": derived["access_point"] = True derived["ptp"] = True + derived["role"] = DerivedWirelessRole.ACCESS_POINT case "sta-ptp": derived["station"] = True derived["ptp"] = True @@ -384,3 +397,69 @@ async def provmode(self, active: bool = False) -> bool: except asyncio.CancelledError: _LOGGER.info("Provisioning mode change task was cancelled") raise + + async def warnings(self) -> dict[str, Any] | Any: + """Get warnings.""" + if not self.connected: + _LOGGER.error("Not connected, login first") + raise AirOSDeviceConnectionError from None + + request_headers = {**self._common_headers} + if self.current_csrf_token: + request_headers["X-CSRF-ID"] = self.current_csrf_token + + # Formal call is '/api/warnings?_=1755249683586' + try: + async with self.session.get( + self._warnings_url, + headers=request_headers, + ) as response: + response_text = await response.text() + if response.status == 200: + return json.loads(response_text) + log = f"Unable to fech warning status {response.status} with {response_text}" + _LOGGER.error(log) + raise AirOSDataMissingError from None + except json.JSONDecodeError: + _LOGGER.exception("JSON Decode Error in warning response") + raise AirOSDataMissingError from None + except (TimeoutError, aiohttp.ClientError) as err: + _LOGGER.exception("Error during call to retrieve warnings: %s", err) + raise AirOSDeviceConnectionError from err + except asyncio.CancelledError: + _LOGGER.info("Warning check task was cancelled") + raise + + async def update_check(self) -> dict[str, Any] | Any: + """Get warnings.""" + if not self.connected: + _LOGGER.error("Not connected, login first") + raise AirOSDeviceConnectionError from None + + request_headers = {**self._common_headers} + if self.current_csrf_token: + request_headers["X-CSRF-ID"] = self.current_csrf_token + request_headers["Content-type"] = "application/json" + + # Post without data + try: + async with self.session.post( + self._update_check_url, + headers=request_headers, + json={}, + ) as response: + response_text = await response.text() + if response.status == 200: + return json.loads(response_text) + log = f"Unable to fech update status {response.status} with {response_text}" + _LOGGER.error(log) + raise AirOSDataMissingError from None + except json.JSONDecodeError: + _LOGGER.exception("JSON Decode Error in warning response") + raise AirOSDataMissingError from None + except (TimeoutError, aiohttp.ClientError) as err: + _LOGGER.exception("Error during call to retrieve update status: %s", err) + raise AirOSDeviceConnectionError from err + except asyncio.CancelledError: + _LOGGER.info("Warning update status task was cancelled") + raise diff --git a/airos/data.py b/airos/data.py index 1e18da7..e840a25 100644 --- a/airos/data.py +++ b/airos/data.py @@ -141,6 +141,20 @@ class IeeeMode(Enum): # More to be added when known +class DerivedWirelessRole(Enum): + """Enum definition.""" + + STATION = "station" + ACCESS_POINT = "access_point" + + +class DerivedWirelessMode(Enum): + """Enum definition.""" + + PTP = "point_to_point" + PTMP = "point_to_multipoint" + + class WirelessMode(Enum): """Enum definition.""" @@ -350,7 +364,7 @@ class Remote(AirOSDataClass): rssi: int noisefloor: int tx_power: int - distance: int + distance: int # In meters rx_chainmask: int chainrssi: list[int] tx_ratedata: list[int] @@ -408,7 +422,7 @@ class Station(AirOSDataClass): tx_nss: int rx_nss: int tx_latency: int - distance: int + distance: int # In meters tx_packets: int tx_lretries: int tx_sretries: int @@ -446,7 +460,7 @@ class Wireless(AirOSDataClass): frequency: int center1_freq: int dfs: int - distance: int + distance: int # In meters security: Security noisef: int txpower: int @@ -554,6 +568,9 @@ class Derived(AirOSDataClass): ptp: bool ptmp: bool + role: DerivedWirelessRole + mode: DerivedWirelessMode + @dataclass class AirOS8Data(AirOSDataClass): diff --git a/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json b/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json index 50921e4..8bc4002 100644 --- a/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json +++ b/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json @@ -13,8 +13,10 @@ "access_point": true, "mac": "68:D7:9A:9A:08:BB", "mac_interface": "br0", + "mode": "point_to_point", "ptmp": false, "ptp": true, + "role": "access_point", "station": false }, "firewall": { diff --git a/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json b/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json index 6182815..e0a716e 100644 --- a/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json +++ b/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json @@ -13,8 +13,10 @@ "access_point": false, "mac": "68:D7:9A:98:FB:FF", "mac_interface": "br0", + "mode": "point_to_point", "ptmp": false, "ptp": true, + "role": "station", "station": true }, "firewall": { diff --git a/fixtures/airos_liteapgps_ap_ptmp_40mhz.json b/fixtures/airos_liteapgps_ap_ptmp_40mhz.json index d11415f..f052d24 100644 --- a/fixtures/airos_liteapgps_ap_ptmp_40mhz.json +++ b/fixtures/airos_liteapgps_ap_ptmp_40mhz.json @@ -13,8 +13,10 @@ "access_point": true, "mac": "04:11:22:33:19:7E", "mac_interface": "br0", + "mode": "point_to_multipoint", "ptmp": true, "ptp": false, + "role": "access_point", "station": false }, "firewall": { diff --git a/fixtures/airos_loco5ac_ap-ptp.json b/fixtures/airos_loco5ac_ap-ptp.json index 6f57d5d..a1d075d 100644 --- a/fixtures/airos_loco5ac_ap-ptp.json +++ b/fixtures/airos_loco5ac_ap-ptp.json @@ -13,8 +13,10 @@ "access_point": true, "mac": "01:23:45:67:89:AB", "mac_interface": "br0", + "mode": "point_to_point", "ptmp": false, "ptp": true, + "role": "access_point", "station": false }, "firewall": { diff --git a/fixtures/airos_loco5ac_sta-ptp.json b/fixtures/airos_loco5ac_sta-ptp.json index 932102e..41f9e44 100644 --- a/fixtures/airos_loco5ac_sta-ptp.json +++ b/fixtures/airos_loco5ac_sta-ptp.json @@ -13,8 +13,10 @@ "access_point": false, "mac": "01:23:45:67:89:CD", "mac_interface": "br0", + "mode": "point_to_point", "ptmp": false, "ptp": true, + "role": "station", "station": true }, "firewall": { diff --git a/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json b/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json index 5ec968b..ba9014b 100644 --- a/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json +++ b/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json @@ -13,8 +13,10 @@ "access_point": false, "mac": "22:22:33:44:31:38", "mac_interface": "br0", + "mode": "point_to_multipoint", "ptmp": true, "ptp": false, + "role": "station", "station": true }, "firewall": { diff --git a/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json b/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json index 948f514..05d8b09 100644 --- a/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json +++ b/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json @@ -13,8 +13,10 @@ "access_point": true, "mac": "00:11:22:33:34:66", "mac_interface": "br0", + "mode": "point_to_point", "ptmp": false, "ptp": true, + "role": "access_point", "station": false }, "firewall": { diff --git a/fixtures/update_check_available.json b/fixtures/update_check_available.json new file mode 100644 index 0000000..493ad5b --- /dev/null +++ b/fixtures/update_check_available.json @@ -0,0 +1 @@ +{"checksum": "b1bea879a9f518f714ce638172e3a860", "version": "v8.7.19", "security": "", "date": "250811", "url": "https://dl.ubnt.com/firmwares/XC-fw/v8.7.19/WA.v8.7.19.48279.250811.0636.bin", "update": true, "changelog": "https://dl.ubnt.com/firmwares/XC-fw/v8.7.19/changelog.txt"} diff --git a/fixtures/warnings.json b/fixtures/warnings.json new file mode 100644 index 0000000..29e4ab1 --- /dev/null +++ b/fixtures/warnings.json @@ -0,0 +1 @@ +{"isDefaultPasswd": false, "customScripts": false, "isWatchdogReset": 0, "label": 0, "chAvailable": false, "emergReasonCode": -1, "firmware": {"requirePasswd": false, "isThirdParty": false, "version": "", "uploaded": false}} diff --git a/pyproject.toml b/pyproject.toml index 57f657a..aa3c044 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "airos" -version = "0.2.11" +version = "0.3.0" license = "MIT" description = "Ubiquity airOS module(s) for Python 3." readme = "README.md" diff --git a/requirements-test.txt b/requirements-test.txt index 7312432..086e7d5 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -10,3 +10,4 @@ radon==6.0.1 types-aiofiles==24.1.0.20250809 mypy==1.17.1 pylint==3.3.7 +aiofiles==24.1.0 diff --git a/script/generate_ha_fixture.py b/script/generate_ha_fixture.py index 74efc5a..315cb80 100644 --- a/script/generate_ha_fixture.py +++ b/script/generate_ha_fixture.py @@ -55,8 +55,10 @@ def generate_airos_fixtures() -> None: except json.JSONDecodeError: _LOGGER.error("Skipping '%s': Not a valid JSON file.", filename) + raise except Exception as e: _LOGGER.error("Error processing '%s': %s", filename, e) + raise if __name__ == "__main__": diff --git a/tests/test_airos8.py b/tests/test_airos8.py index e88afab..90064dd 100644 --- a/tests/test_airos8.py +++ b/tests/test_airos8.py @@ -8,6 +8,7 @@ import airos.exceptions import pytest +import aiofiles import aiohttp from mashumaro.exceptions import MissingField @@ -259,3 +260,138 @@ async def test_status_missing_required_key_in_json(airos_device: AirOS) -> None: # --- MODIFICATION START --- # Assert that the cause of our exception is the correct type from mashumaro assert isinstance(excinfo.value.__cause__, MissingField) + + +# --- Tests for warnings() and update_check() --- +@pytest.mark.asyncio +async def test_warnings_correctly_parses_json() -> None: + """Test that the warnings() method correctly parses a valid JSON response.""" + mock_session = MagicMock() + airos_device = AirOS( + host="http://192.168.1.3", + username="test", + password="test", + session=mock_session, + ) + airos_device.connected = True + + mock_response = MagicMock() + mock_response.__aenter__.return_value = mock_response + mock_response.status = 200 + async with aiofiles.open("fixtures/warnings.json") as f: + content = await f.read() + mock_response_data = json.loads(content) + mock_response.text = AsyncMock(return_value=json.dumps(mock_response_data)) + + with patch.object(airos_device.session, "get", return_value=mock_response): + result = await airos_device.warnings() + assert result["isDefaultPasswd"] is False + assert result["chAvailable"] is False + + +@pytest.mark.asyncio +async def test_warnings_raises_exception_on_invalid_json() -> None: + """Test that warnings() raises an exception on invalid JSON response.""" + mock_session = MagicMock() + airos_device = AirOS( + host="http://192.168.1.3", + username="test", + password="test", + session=mock_session, + ) + airos_device.connected = True + + mock_response = MagicMock() + mock_response.__aenter__.return_value = mock_response + mock_response.status = 200 + mock_response.text = AsyncMock(return_value="This is not JSON") + + with ( + patch.object(airos_device.session, "get", return_value=mock_response), + pytest.raises(airos.exceptions.AirOSDataMissingError), + ): + await airos_device.warnings() + + +@pytest.mark.asyncio +async def test_update_check_correctly_parses_json() -> None: + """Test that update_check() method correctly parses a valid JSON response.""" + mock_session = MagicMock() + airos_device = AirOS( + host="http://192.168.1.3", + username="test", + password="test", + session=mock_session, + ) + airos_device.connected = True + airos_device.current_csrf_token = "mock-csrf-token" + + mock_response = MagicMock() + mock_response.__aenter__.return_value = mock_response + mock_response.status = 200 + async with aiofiles.open("fixtures/update_check_available.json") as f: + content = await f.read() + mock_response_data = json.loads(content) + mock_response.text = AsyncMock(return_value=json.dumps(mock_response_data)) + + with patch.object(airos_device.session, "post", return_value=mock_response): + result = await airos_device.update_check() + assert result["version"] == "v8.7.19" + assert result["update"] is True + + +@pytest.mark.asyncio +async def test_update_check_raises_exception_on_invalid_json() -> None: + """Test that update_check() raises an exception on invalid JSON response.""" + mock_session = MagicMock() + airos_device = AirOS( + host="http://192.168.1.3", + username="test", + password="test", + session=mock_session, + ) + airos_device.connected = True + airos_device.current_csrf_token = "mock-csrf-token" + + mock_response = MagicMock() + mock_response.__aenter__.return_value = mock_response + mock_response.status = 200 + mock_response.text = AsyncMock(return_value="This is not JSON") + + with ( + patch.object(airos_device.session, "post", return_value=mock_response), + pytest.raises(airos.exceptions.AirOSDataMissingError), + ): + await airos_device.update_check() + + +@pytest.mark.asyncio +async def test_warnings_when_not_connected() -> None: + """Test calling warnings() before a successful login.""" + mock_session = MagicMock() + airos_device = AirOS( + host="http://192.168.1.3", + username="test", + password="test", + session=mock_session, + ) + airos_device.connected = False # Explicitly set connected state to false + + with pytest.raises(airos.exceptions.AirOSDeviceConnectionError): + await airos_device.warnings() + + +@pytest.mark.asyncio +async def test_update_check_when_not_connected() -> None: + """Test calling update_check() before a successful login.""" + mock_session = MagicMock() + airos_device = AirOS( + host="http://192.168.1.3", + username="test", + password="test", + session=mock_session, + ) + airos_device.connected = False # Explicitly set connected state to false + + with pytest.raises(airos.exceptions.AirOSDeviceConnectionError): + await airos_device.update_check()