diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 7c48192..393ca0f 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -174,7 +174,7 @@ jobs: run: | . venv/bin/activate coverage combine artifacts/.coverage* - coverage report --fail-under=85 + coverage report --fail-under=80 coverage xml - name: Upload coverage to Codecov uses: codecov/codecov-action@v5 diff --git a/.gitignore b/.gitignore index 5e1335d..9cfb169 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,4 @@ tests/__pycache__ tmp todo .DS_Store +test.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7369fe7..4d57746 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ All notable changes to this project will be documented in this file. +## [0.6.0] - 2025-10-22 + +Major thanks to user HJ@discord for putting up with testing and dustball62 for confirming + +### Added + +- Support for v6 firmware XM models using a different login path (XW already was successful) +- Calculated cpuload on v6 if values available (to prevent reporting close to 100%) +- Fix frequency on v6 firmware (if value is cast as a string ending in MHz) +- Added tx/rx rates for v6 as capacity (throughput is available in v6 web interface, but probably through counters, so not yet available) +- Fixed ieeemode (v8) vs opmode (v6) mapped back to IeeeMode enum +- Derived antenna_gain (v8) from antenna (v6 string) +- Improved internal workings and firmware detection + ## [0.5.6] - 2025-10-11 ### Added diff --git a/airos/airos6.py b/airos/airos6.py index cfb8ab8..4044c52 100644 --- a/airos/airos6.py +++ b/airos/airos6.py @@ -36,7 +36,7 @@ def __init__( ) @staticmethod - def derived_wireless_data( + def _derived_wireless_data( derived: dict[str, Any], response: dict[str, Any] ) -> dict[str, Any]: """Add derived wireless data to the device response.""" diff --git a/airos/base.py b/airos/base.py index bb76c1e..6146fb2 100644 --- a/airos/base.py +++ b/airos/base.py @@ -55,6 +55,8 @@ def __init__( self.username = username self.password = password + self.api_version: int = 8 + parsed_host = urlparse(host) scheme = ( parsed_host.scheme @@ -74,11 +76,13 @@ def __init__( self.current_csrf_token: str | None = None # Mostly 8.x API endpoints, login/status are the same in 6.x - self._login_urls = { - "default": f"{self.base_url}/api/auth", - "v6_alternative": f"{self.base_url}/login.cgi", - } + self._login_url = f"{self.base_url}/api/auth" self._status_cgi_url = f"{self.base_url}/status.cgi" + + # Presumed 6.x XM only endpoint + self._v6_xm_login_url = f"{self.base_url}/login.cgi" + self._v6_form_url = "/index.cgi" + # Presumed 8.x only endpoints self._stakick_cgi_url = f"{self.base_url}/stakick.cgi" self._provmode_url = f"{self.base_url}/api/provmode" @@ -88,8 +92,10 @@ def __init__( self._download_progress_url = f"{self.base_url}/api/fw/download-progress" self._install_url = f"{self.base_url}/fwflash.cgi" + self._login_urls = [self._login_url, self._v6_xm_login_url] + @staticmethod - def derived_wireless_data( + def _derived_wireless_data( derived: dict[str, Any], response: dict[str, Any] ) -> dict[str, Any]: """Add derived wireless data to the device response.""" @@ -129,7 +135,7 @@ def _derived_data_helper( sku = UispAirOSProductMapper().get_sku_by_devmodel(devmodel) except KeyError: _LOGGER.warning( - "Unknown SKU/Model ID for %s. Please report at " + "Unknown SKU/Model ID for '%s'. Please report at " "https://github.com/CoMPaTech/python-airos/issues so we can add support.", devmodel, ) @@ -152,41 +158,71 @@ def _derived_data_helper( "mode": DerivedWirelessMode.PTP, "sku": sku, } + # WIRELESS derived = derived_wireless_data_func(derived, response) - # INTERFACES - addresses = {} - interface_order = ["br0", "eth0", "ath0"] - + # Interfaces / MAC (for unique id) interfaces = response.get("interfaces", []) - # No interfaces, no mac, no usability if not interfaces: _LOGGER.error("Failed to determine interfaces from AirOS data") raise AirOSKeyDataMissingError from None - for interface in interfaces: - if interface["enabled"]: # Only consider if enabled - addresses[interface["ifname"]] = interface["hwaddr"] - - # Fallback take fist alternate interface found - derived["mac"] = interfaces[0]["hwaddr"] - derived["mac_interface"] = interfaces[0]["ifname"] + derived["mac"] = AirOS.get_mac(interfaces)["mac"] + derived["mac_interface"] = AirOS.get_mac(interfaces)["mac_interface"] - for interface in interface_order: - if interface in addresses: - derived["mac"] = addresses[interface] - derived["mac_interface"] = interface - break + # Firmware Major Version + fwversion = (response.get("host") or {}).get("fwversion", "invalid") + derived["fw_major"] = AirOS.get_fw_major(fwversion) response["derived"] = derived return response - def derived_data(self, response: dict[str, Any]) -> dict[str, Any]: + @staticmethod + def get_fw_major(fwversion: str) -> int: + """Extract major firmware version from fwversion string.""" + try: + return int(fwversion.lstrip("v").split(".", 1)[0]) + except (ValueError, AttributeError) as err: + _LOGGER.error("Invalid firmware version '%s'", fwversion) + raise AirOSKeyDataMissingError("invalid fwversion") from err + + @staticmethod + def get_mac(interfaces: list[dict[str, Any]]) -> dict[str, str]: + """Extract MAC address from interfaces.""" + result: dict[str, str] = {"mac": "", "mac_interface": ""} + + if not interfaces: + return result + + addresses: dict[str, str] = {} + interface_order = ["br0", "eth0", "ath0"] + + for interface in interfaces: + if ( + interface.get("enabled") + and interface.get("hwaddr") + and interface.get("ifname") + ): + addresses[interface["ifname"]] = interface["hwaddr"] + + for preferred in interface_order: + if preferred in addresses: + result["mac"] = addresses[preferred] + result["mac_interface"] = preferred + break + else: + result["mac"] = interfaces[0].get("hwaddr", "") + result["mac_interface"] = interfaces[0].get("ifname", "") + + return result + + @classmethod + def derived_data(cls, response: dict[str, Any]) -> dict[str, Any]: """Add derived data to the device response (instance method for polymorphism).""" - return self._derived_data_helper(response, self.derived_wireless_data) + return cls._derived_data_helper(response, cls._derived_wireless_data) def _get_authenticated_headers( self, @@ -204,7 +240,8 @@ def _get_authenticated_headers( headers["X-CSRF-ID"] = self._csrf_id if self._auth_cookie: # pragma: no cover - headers["Cookie"] = f"AIROS_{self._auth_cookie}" + # headers["Cookie"] = f"AIROS_{self._auth_cookie}" + headers["Cookie"] = self._auth_cookie return headers @@ -218,7 +255,8 @@ def _store_auth_data(self, response: aiohttp.ClientResponse) -> None: cookie.load(set_cookie) for key, morsel in cookie.items(): if key.startswith("AIROS_"): - self._auth_cookie = morsel.key[6:] + "=" + morsel.value + # self._auth_cookie = morsel.key[6:] + "=" + morsel.value + self._auth_cookie = f"{morsel.key}={morsel.value}" break async def _request_json( @@ -243,7 +281,7 @@ async def _request_json( request_headers.update(headers) try: - if url not in self._login_urls.values() and not self.connected: + if url not in self._login_urls and not self.connected: _LOGGER.error("Not connected, login first") raise AirOSDeviceConnectionError from None @@ -259,7 +297,7 @@ async def _request_json( _LOGGER.debug("Successfully fetched JSON from %s", url) # If this is the login request, we need to store the new auth data - if url in self._login_urls.values(): + if url in self._login_urls: self._store_auth_data(response) self.connected = True @@ -283,31 +321,71 @@ async def _request_json( _LOGGER.warning("Request to %s was cancelled", url) raise + async def _login_v6(self) -> None: + """Login to airOS v6 (XM) devices.""" + # Handle session cookie from login url + async with self.session.request( + "GET", + self._v6_xm_login_url, + allow_redirects=False, + ) as response: + session_cookie = next( + (c for n, c in response.cookies.items() if n.startswith("AIROS")), None + ) + if not session_cookie: + raise AirOSDeviceConnectionError("No session cookie received.") + self._auth_cookie = f"{session_cookie.key}={session_cookie.value}" + + # Handle login expecting 302 redirect + payload = { + "username": self.username, + "password": self.password, + "uri": self._v6_form_url, + } + headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Origin": self.base_url, + "Referer": self._v6_xm_login_url, + "Cookie": self._auth_cookie, + } + async with self.session.request( + "POST", + self._v6_xm_login_url, + data=payload, + headers=headers, + allow_redirects=False, + ) as response: + if response.status != 302: + raise AirOSConnectionAuthenticationError("Login failed.") + + # Activate session by accessing the form URL + headers = {"Referer": self._v6_xm_login_url, "Cookie": self._auth_cookie} + async with self.session.request( + "GET", + f"{self.base_url}{self._v6_form_url}", + headers=headers, + allow_redirects=True, + ) as response: + if "login.cgi" in str(response.url): + raise AirOSConnectionAuthenticationError("Session activation failed.") + self.connected = True + self.api_version = 6 + async def login(self) -> None: """Login to AirOS device.""" payload = {"username": self.username, "password": self.password} try: - await self._request_json( - "POST", self._login_urls["default"], json_data=payload - ) + await self._request_json("POST", self._login_url, json_data=payload) except AirOSUrlNotFoundError: - pass # Try next URL + await self._login_v6() except AirOSConnectionSetupError as err: raise AirOSConnectionSetupError("Failed to login to AirOS device") from err else: return - try: # Alternative URL - await self._request_json( - "POST", - self._login_urls["v6_alternative"], - form_data=payload, - ct_form=True, - ) - except AirOSConnectionSetupError as err: - raise AirOSConnectionSetupError( - "Failed to login to default and alternate AirOS device urls" - ) from err + async def raw_status(self) -> dict[str, Any]: + """Retrieve raw status from the device.""" + return await self._request_json("GET", self._status_cgi_url, authenticated=True) async def status(self) -> AirOSDataModel: """Retrieve status from the device.""" diff --git a/airos/data.py b/airos/data.py index cf2f4f2..d164226 100644 --- a/airos/data.py +++ b/airos/data.py @@ -102,9 +102,6 @@ def _redact(d: dict[str, Any]) -> dict[str, Any]: return _redact(data) -# Data class start - - class AirOSDataClass(DataClassDictMixin): """A base class for all mashumaro dataclasses.""" @@ -145,8 +142,12 @@ class IeeeMode(Enum): AUTO = "AUTO" _11ACVHT80 = "11ACVHT80" # On a NanoStation + _11ACVHT60 = "11ACVHT60" + _11ACVHT50 = "11ACVHT50" _11ACVHT40 = "11ACVHT40" _11ACVHT20 = "11ACVHT20" # On a LiteBeam + _11NAHT40MINUS = "11NAHT40MINUS" # On a v6 XM + _11NAHT40PLUS = "11NAHT40PLUS" # On a v6 XW # More to be added when known @@ -246,11 +247,20 @@ class Host6(AirOSDataClass): totalram: int freeram: int cpuload: float | int | None + cputotal: float | int | None # Reported on XM firmware + cpubusy: float | int | None # Reported on XM firmware @classmethod def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: """Pre-deserialize hook for Host.""" _check_and_log_unknown_enum_value(d, "netrole", NetRole, "Host", "netrole") + + # Calculate cpufloat from actuals instead on relying on near 100% value + if ( + all(isinstance(d.get(k), (int, float)) for k in ("cpubusy", "cputotal")) + and d["cputotal"] > 0 + ): + d["cpuload"] = round((d["cpubusy"] / d["cputotal"]) * 100, 2) return d @@ -324,6 +334,14 @@ class Polling(AirOSDataClass): flex_mode: int | None = None # Not present in all devices +@dataclass +class Polling6(AirOSDataClass): + """Leaf definition.""" + + dl_capacity: int | None = None # New + ul_capacity: int | None = None # New + + @dataclass class Stats(AirOSDataClass): """Leaf definition.""" @@ -543,13 +561,19 @@ class Wireless(AirOSDataClass): @classmethod def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: """Pre-deserialize hook for Wireless.""" + _check_and_log_unknown_enum_value(d, "mode", WirelessMode, "Wireless", "mode") + _check_and_log_unknown_enum_value( - d, "ieeemode", IeeeMode, "Wireless", "ieeemode" + d, "security", Security, "Wireless", "security" ) + + # Ensure ieeemode/opmode are in uppercase and map opmode back into ieeemode + d["ieeemode"] = d["ieeemode"].upper() or None _check_and_log_unknown_enum_value( - d, "security", Security, "Wireless", "security" + d, "ieeemode", IeeeMode, "Wireless", "ieeemode" ) + return d @@ -562,7 +586,7 @@ class Wireless6(AirOSDataClass): apmac: str countrycode: int channel: int - frequency: str + frequency: int dfs: int opmode: str antenna: str @@ -584,7 +608,10 @@ class Wireless6(AirOSDataClass): wds: int aprepeater: int # Not bool as v8 chanbw: int + polling: Polling6 + ieeemode: IeeeMode # Virtual to match base/v8 mode: Wireless6Mode | None = None + antenna_gain: int | None = None # Virtual to match base/v8 @classmethod def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: @@ -593,6 +620,25 @@ def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: _check_and_log_unknown_enum_value( d, "security", Security, "Wireless", "security" ) + + freq = d.get("frequency") + if isinstance(freq, str) and "MHz" in freq: + d["frequency"] = int(freq.split()[0]) + + rxrate = d.get("rxrate") + txrate = d.get("txrate") + d["polling"] = { # Map to Polling6 as MBPS strings to int kbps + "dl_capacity": int(float(rxrate) * 1000) if rxrate else 0, + "ul_capacity": int(float(txrate) * 1000) if txrate else 0, + } + + d["ieeemode"] = d["opmode"].upper() or None + _check_and_log_unknown_enum_value( + d, "ieeemode", IeeeMode, "Wireless", "ieeemode" + ) + match = re.search(r"(\d+)\s*dBi", d["antenna"]) + d["antenna_gain"] = int(match.group(1)) if match else None + return d @@ -691,6 +737,9 @@ class Derived(AirOSDataClass): # Lookup of model_id (presumed via SKU) sku: str + # Firmware major version + fw_major: int | None = None + @dataclass class AirOS8Data(AirOSDataBaseClass): diff --git a/airos/helpers.py b/airos/helpers.py index bf47279..2c37cfa 100644 --- a/airos/helpers.py +++ b/airos/helpers.py @@ -1,12 +1,20 @@ """Ubiquiti AirOS firmware helpers.""" +import logging from typing import TypedDict import aiohttp -from .airos6 import AirOS6 from .airos8 import AirOS8 -from .exceptions import AirOSKeyDataMissingError +from .exceptions import ( + AirOSConnectionAuthenticationError, + AirOSConnectionSetupError, + AirOSDataMissingError, + AirOSDeviceConnectionError, + AirOSKeyDataMissingError, +) + +_LOGGER = logging.getLogger(__name__) class DetectDeviceData(TypedDict): @@ -25,47 +33,26 @@ async def async_get_firmware_data( use_ssl: bool = True, ) -> DetectDeviceData: """Connect to a device and return the major firmware version.""" - detect: AirOS8 = AirOS8(host, username, password, session, use_ssl) - - await detect.login() - raw_status = await detect._request_json( # noqa: SLF001 - "GET", - detect._status_cgi_url, # noqa: SLF001 - authenticated=True, - ) - - fw_version = (raw_status.get("host") or {}).get("fwversion") - if not fw_version: - raise AirOSKeyDataMissingError("Missing host.fwversion in API response") + detect_device: AirOS8 = AirOS8(host, username, password, session, use_ssl) try: - fw_major = int(fw_version.lstrip("v").split(".", 1)[0]) - except (ValueError, AttributeError) as exc: - raise AirOSKeyDataMissingError( - f"Invalid firmware version '{fw_version}'" - ) from exc - - if fw_major == 6: - derived_data = AirOS6._derived_data_helper( # noqa: SLF001 - raw_status, AirOS6.derived_wireless_data - ) - else: # Assume AirOS 8 for all other versions - derived_data = AirOS8._derived_data_helper( # noqa: SLF001 - raw_status, AirOS8.derived_wireless_data - ) - - # Extract MAC address and hostname from the derived data - hostname = derived_data.get("host", {}).get("hostname") - mac = derived_data.get("derived", {}).get("mac") - - if not hostname: # pragma: no cover - raise AirOSKeyDataMissingError("Missing hostname") - - if not mac: # pragma: no cover - raise AirOSKeyDataMissingError("Missing MAC address") + await detect_device.login() + device_data = await detect_device.raw_status() + except ( + AirOSConnectionSetupError, + AirOSDeviceConnectionError, + ): + _LOGGER.exception("Error connecting to device at %s", host) + raise + except (AirOSConnectionAuthenticationError, AirOSDataMissingError): + _LOGGER.exception("Authentication error connecting to device at %s", host) + raise + except AirOSKeyDataMissingError: + _LOGGER.exception("Key data missing from device at %s", host) + raise return { - "fw_major": fw_major, - "mac": mac, - "hostname": hostname, + "fw_major": AirOS8.get_fw_major(device_data.get("host", {}).get("fwversion")), + "hostname": device_data.get("host", {}).get("hostname"), + "mac": AirOS8.get_mac(device_data.get("interfaces", {}))["mac"], } diff --git a/airos/model_map.py b/airos/model_map.py index 8f69b77..9dacf01 100644 --- a/airos/model_map.py +++ b/airos/model_map.py @@ -2,8 +2,8 @@ from .exceptions import AirOSMultipleMatchesFoundException -MODELS: dict[str, str] = { - # Generated list from https://store.ui.com/us/en/category/wireless +# Generated list from https://store.ui.com/us/en/category/wireless +SITE_MODELS: dict[str, str] = { "Wave MLO5": "Wave-MLO5", "airMAX Rocket Prism 5AC": "RP-5AC-Gen2", "airFiber 5XHD": "AF-5XHD", @@ -76,10 +76,16 @@ "airMAX 5 GHz, 19/20 dBi Sector": "AM-5G2", "airMAX 2.4 GHz, 10 dBi Omni": "AMO-2G10", "airMAX 2.4 GHz, 15 dBi, 120ยบ Sector": "AM-2G15-120", - # Manually added entries for common unofficial names +} + +# Manually added entries for common unofficial names +MANUAL_MODELS: dict[str, str] = { "LiteAP GPS": "LAP-GPS", # Shortened name for airMAX Lite Access Point GPS + "NanoStation loco M5": "LocoM5", # XM firmware version 6 - note the reversed names } +MODELS: dict[str, str] = {**SITE_MODELS, **MANUAL_MODELS} + class UispAirOSProductMapper: """Utility class to map product model names to SKUs and vice versa.""" @@ -90,6 +96,7 @@ def __init__(self) -> None: def get_sku_by_devmodel(self, devmodel: str) -> str: """Retrieves the SKU for a given device model name.""" + devmodel = devmodel.strip() if devmodel in MODELS: return MODELS[devmodel] diff --git a/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json b/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json index c507f45..bb21cab 100644 --- a/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json +++ b/fixtures/airos_LiteBeam5AC_ap-ptp_30mhz.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": true, + "fw_major": 8, "mac": "68:D7:9A:9A:08:BB", "mac_interface": "br0", "mode": "point_to_point", diff --git a/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json b/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json index d6fa87f..f705373 100644 --- a/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json +++ b/fixtures/airos_LiteBeam5AC_sta-ptp_30mhz.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": false, + "fw_major": 8, "mac": "68:D7:9A:98:FB:FF", "mac_interface": "br0", "mode": "point_to_point", diff --git a/fixtures/airos_NanoBeam_5AC_ap-ptmp_v8.7.18.json b/fixtures/airos_NanoBeam_5AC_ap-ptmp_v8.7.18.json index afceca4..315623c 100644 --- a/fixtures/airos_NanoBeam_5AC_ap-ptmp_v8.7.18.json +++ b/fixtures/airos_NanoBeam_5AC_ap-ptmp_v8.7.18.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": true, + "fw_major": 8, "mac": "xxxxxxxxxxxxxxxx", "mac_interface": "br0", "mode": "point_to_multipoint", diff --git a/fixtures/airos_NanoStation_M5_sta_v6.3.16.json b/fixtures/airos_NanoStation_M5_sta_v6.3.16.json index 4682d57..84a30a9 100644 --- a/fixtures/airos_NanoStation_M5_sta_v6.3.16.json +++ b/fixtures/airos_NanoStation_M5_sta_v6.3.16.json @@ -4,13 +4,14 @@ }, "derived": { "access_point": false, + "fw_major": 6, "mac": "XX:XX:XX:XX:XX:XX", "mac_interface": "br0", "mode": "point_to_point", "ptmp": false, "ptp": true, "role": "station", - "sku": "LocoM5", + "sku": "NSM5", "station": true }, "firewall": { @@ -21,7 +22,9 @@ }, "genuine": "/images/genuine.png", "host": { - "cpuload": 24.0, + "cpubusy": 3786414, + "cpuload": 25.51, + "cputotal": 14845531, "devmodel": "NanoStation M5 ", "freeram": 42516480, "fwprefix": "XW", @@ -129,6 +132,7 @@ "wireless": { "ack": 5, "antenna": "Built in - 16 dBi", + "antenna_gain": 16, "apmac": "xxxxxxxx", "aprepeater": 0, "cac_nol": 0, @@ -140,12 +144,17 @@ "dfs": 0, "distance": 750, "essid": "Nano", - "frequency": "5180 MHz", + "frequency": 5180, "hide_essid": 0, + "ieeemode": "11NAHT40PLUS", "mode": "sta", "noisef": -99, "nol_chans": 0, "opmode": "11NAHT40PLUS", + "polling": { + "dl_capacity": 216000, + "ul_capacity": 270000 + }, "qos": "No QoS", "rssi": 32, "rstatus": 5, diff --git a/fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_ap.json b/fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_ap.json new file mode 100644 index 0000000..eff6efe --- /dev/null +++ b/fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_ap.json @@ -0,0 +1,154 @@ +{ + "airview": { + "enabled": 0 + }, + "derived": { + "access_point": true, + "fw_major": 6, + "mac": "XX:XX:XX:XX:XX:XX", + "mac_interface": "br0", + "mode": "point_to_point", + "ptmp": false, + "ptp": true, + "role": "access_point", + "sku": "LocoM5", + "station": false + }, + "firewall": { + "eb6tables": false, + "ebtables": true, + "ip6tables": false, + "iptables": false + }, + "genuine": "/images/genuine.png", + "host": { + "cpubusy": 11092563, + "cpuload": 5.68, + "cputotal": 195292633, + "devmodel": "NanoStation loco M5 ", + "freeram": 8429568, + "fwprefix": "XM", + "fwversion": "v6.3.16", + "hostname": "NanoStation loco M5 AP", + "netrole": "bridge", + "totalram": 30220288, + "uptime": 1953224 + }, + "interfaces": [ + { + "enabled": true, + "hwaddr": "00:00:00:00:00:00", + "ifname": "lo", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 0 + } + }, + { + "enabled": true, + "hwaddr": "XX:XX:XX:XX:XX:XX", + "ifname": "eth0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 100 + } + }, + { + "enabled": true, + "hwaddr": "XX:XX:XX:XX:XX:XX", + "ifname": "wifi0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 0 + } + }, + { + "enabled": true, + "hwaddr": "XX:XX:XX:XX:XX:XX", + "ifname": "ath0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 300 + } + }, + { + "enabled": true, + "hwaddr": "XX:XX:XX:XX:XX:XX", + "ifname": "br0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 0 + } + } + ], + "services": { + "dhcpc": false, + "dhcpd": false, + "pppoe": false + }, + "unms": { + "status": 0, + "timestamp": "" + }, + "wireless": { + "ack": 28, + "antenna": "Built in - 13 dBi", + "antenna_gain": 13, + "apmac": "XX:XX:XX:XX:XX:XX", + "aprepeater": 0, + "cac_nol": 0, + "ccq": 870, + "chains": "2X2", + "chanbw": 40, + "channel": 140, + "countrycode": 616, + "dfs": 1, + "distance": 600, + "essid": "SOMETHING", + "frequency": 5700, + "hide_essid": 1, + "ieeemode": "11NAHT40MINUS", + "mode": "ap", + "noisef": -91, + "nol_chans": 0, + "opmode": "11naht40minus", + "polling": { + "dl_capacity": 300000, + "ul_capacity": 270000 + }, + "qos": "No QoS", + "rssi": 51, + "rstatus": 5, + "rxrate": "300", + "security": "WPA2", + "signal": -45, + "txpower": 2, + "txrate": "270", + "wds": 1 + } +} \ No newline at end of file diff --git a/fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_sta.json b/fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_sta.json new file mode 100644 index 0000000..7435068 --- /dev/null +++ b/fixtures/airos_NanoStation_loco_M5_v6.3.16_XM_sta.json @@ -0,0 +1,154 @@ +{ + "airview": { + "enabled": 0 + }, + "derived": { + "access_point": false, + "fw_major": 6, + "mac": "YY:YY:YY:YY:YY:YY", + "mac_interface": "br0", + "mode": "point_to_point", + "ptmp": false, + "ptp": true, + "role": "station", + "sku": "LocoM5", + "station": true + }, + "firewall": { + "eb6tables": false, + "ebtables": true, + "ip6tables": false, + "iptables": false + }, + "genuine": "/images/genuine.png", + "host": { + "cpubusy": 11150046, + "cpuload": 5.65, + "cputotal": 197455604, + "devmodel": "NanoStation loco M5 ", + "freeram": 8753152, + "fwprefix": "XM", + "fwversion": "v6.3.16", + "hostname": "NanoStation loco M5 Client", + "netrole": "bridge", + "totalram": 30220288, + "uptime": 1974859 + }, + "interfaces": [ + { + "enabled": true, + "hwaddr": "00:00:00:00:00:00", + "ifname": "lo", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 0 + } + }, + { + "enabled": true, + "hwaddr": "YY:YY:YY:YY:YY:YY", + "ifname": "eth0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": false, + "ip6addr": null, + "plugged": false, + "snr": null, + "speed": 0 + } + }, + { + "enabled": true, + "hwaddr": "YY:YY:YY:YY:YY:YY", + "ifname": "wifi0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 0 + } + }, + { + "enabled": true, + "hwaddr": "YY:YY:YY:YY:YY:YY", + "ifname": "ath0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 300 + } + }, + { + "enabled": true, + "hwaddr": "YY:YY:YY:YY:YY:YY", + "ifname": "br0", + "mtu": null, + "status": { + "cable_len": null, + "duplex": true, + "ip6addr": null, + "plugged": true, + "snr": null, + "speed": 0 + } + } + ], + "services": { + "dhcpc": false, + "dhcpd": false, + "pppoe": false + }, + "unms": { + "status": 0, + "timestamp": "" + }, + "wireless": { + "ack": 28, + "antenna": "Built in - 13 dBi", + "antenna_gain": 13, + "apmac": "XX:XX:XX:XX:XX:XX", + "aprepeater": 0, + "cac_nol": 0, + "ccq": 738, + "chains": "2X2", + "chanbw": 40, + "channel": 140, + "countrycode": 616, + "dfs": 0, + "distance": 600, + "essid": "SOMETHING", + "frequency": 5700, + "hide_essid": 0, + "ieeemode": "11NAHT40MINUS", + "mode": "sta", + "noisef": -89, + "nol_chans": 0, + "opmode": "11naht40minus", + "polling": { + "dl_capacity": 180000, + "ul_capacity": 243000 + }, + "qos": "No QoS", + "rssi": 50, + "rstatus": 5, + "rxrate": "180", + "security": "WPA2", + "signal": -46, + "txpower": 2, + "txrate": "243", + "wds": 1 + } +} \ No newline at end of file diff --git a/fixtures/airos_liteapgps_ap_ptmp_40mhz.json b/fixtures/airos_liteapgps_ap_ptmp_40mhz.json index d0d827a..7aea668 100644 --- a/fixtures/airos_liteapgps_ap_ptmp_40mhz.json +++ b/fixtures/airos_liteapgps_ap_ptmp_40mhz.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": true, + "fw_major": 8, "mac": "04:11:22:33:19:7E", "mac_interface": "br0", "mode": "point_to_multipoint", diff --git a/fixtures/airos_loco5ac_ap-ptp.json b/fixtures/airos_loco5ac_ap-ptp.json index b4019c4..75bbefe 100644 --- a/fixtures/airos_loco5ac_ap-ptp.json +++ b/fixtures/airos_loco5ac_ap-ptp.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": true, + "fw_major": 8, "mac": "01:23:45:67:89:AB", "mac_interface": "br0", "mode": "point_to_point", diff --git a/fixtures/airos_loco5ac_sta-ptp.json b/fixtures/airos_loco5ac_sta-ptp.json index 8fafa7e..2d2d464 100644 --- a/fixtures/airos_loco5ac_sta-ptp.json +++ b/fixtures/airos_loco5ac_sta-ptp.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": false, + "fw_major": 8, "mac": "01:23:45:67:89:CD", "mac_interface": "br0", "mode": "point_to_point", diff --git a/fixtures/airos_mocked_sta-ptmp.json b/fixtures/airos_mocked_sta-ptmp.json index 524dad8..2dc5189 100644 --- a/fixtures/airos_mocked_sta-ptmp.json +++ b/fixtures/airos_mocked_sta-ptmp.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": false, + "fw_major": 8, "mac": "01:23:45:67:89:CD", "mac_interface": "br0", "mode": "point_to_multipoint", diff --git a/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json b/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json index f0bf919..f2c5734 100644 --- a/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json +++ b/fixtures/airos_nanobeam5ac_sta_ptmp_40mhz.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": false, + "fw_major": 8, "mac": "22:22:33:44:31:38", "mac_interface": "br0", "mode": "point_to_multipoint", diff --git a/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json b/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json index 6e00e08..754024c 100644 --- a/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json +++ b/fixtures/airos_nanostation_ap-ptp_8718_missing_gps.json @@ -11,6 +11,7 @@ ], "derived": { "access_point": true, + "fw_major": 8, "mac": "00:11:22:33:34:66", "mac_interface": "br0", "mode": "point_to_point", diff --git a/fixtures/userdata/NanoStation_loco_M5_v6.3.16_XM_ap.json b/fixtures/userdata/NanoStation_loco_M5_v6.3.16_XM_ap.json new file mode 100644 index 0000000..de1f9d4 --- /dev/null +++ b/fixtures/userdata/NanoStation_loco_M5_v6.3.16_XM_ap.json @@ -0,0 +1,121 @@ +{ + "host": { + "uptime": 1953224, + "time": "2025-10-20 15:35:42", + "fwversion": "v6.3.16", + "fwprefix": "XM", + "hostname": "NanoStation loco M5 AP", + "devmodel": "NanoStation loco M5 ", + "netrole": "bridge", + "totalram": 30220288, + "freeram": 8429568, + "cpuload": 100.0, + "cputotal": 195292633, + "cpubusy": 11092563 + }, + "wireless": { + "mode": "ap", + "essid": "SOMETHING", + "hide_essid": 1, + "apmac": "XX:XX:XX:XX:XX:XX", + "countrycode": 616, + "channel": 140, + "frequency": "5700 MHz", + "dfs": "1", + "opmode": "11naht40minus", + "antenna": "Built in - 13 dBi", + "chains": "2X2", + "signal": -45, + "rssi": 51, + "noisef": -91, + "txpower": 2, + "ack": 28, + "distance": 600, + "ccq": 870, + "txrate": "270", + "rxrate": "300", + "security": "WPA2", + "qos": "No QoS", + "rstatus": 5, + "count": 1, + "cac_nol": 0, + "nol_chans": 0, + "polling": { + "enabled": 1, + "quality": 86, + "capacity": 62, + "priority": 3, + "noack": 0, + "airsync_mode": 0, + "airsync_connections": 0, + "airsync_down_util": 0, + "airsync_up_util": 0, + "airselect": 0, + "airselect_interval": 1000, + "ff_mode": 0, + "ff_duration": 0, + "ff_dlulratio": 0, + "atpc_status": 0 + }, + "stats": { + "rx_nwids": 0, + "rx_crypts": 0, + "rx_frags": 0, + "tx_retries": 0, + "missed_beacons": 0, + "err_other": 0 + }, + "wds": 1, + "aprepeater": 0, + "chwidth": 20, + "chanbw": 40, + "cwmmode": 0, + "rx_chainmask": 3, + "tx_chainmask": 3, + "chainrssi": [49, 47, 0], + "chainrssimgmt": [49, 47, 0], + "chainrssiext": [49, 47, 0] + }, + "airview": { "enabled": 0 }, + "services": { "dhcpc": 0, "dhcpd": 0, "pppoe": 0 }, + "firewall": { "iptables": 0, "ebtables": 1, "ip6tables": 0, "eb6tables": 0 }, + "genuine": "/images/genuine.png", + "unms": { "status": 0, "timestamp": "", "link": "" }, + "interfaces": [ + { + "ifname": "lo", + "hwaddr": "00:00:00:00:00:00", + "enabled": true, + "status": { "plugged": 1, "speed": 0, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "eth0", + "hwaddr": "XX:XX:XX:XX:XX:XX", + "enabled": true, + "status": { "plugged": 1, "speed": 100, "duplex": 1 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "wifi0", + "hwaddr": "XX:XX:XX:XX:XX:XX", + "enabled": true, + "status": { "plugged": 1, "speed": 0, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "ath0", + "hwaddr": "XX:XX:XX:XX:XX:XX", + "enabled": true, + "status": { "plugged": 1, "speed": 300, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "br0", + "hwaddr": "XX:XX:XX:XX:XX:XX", + "enabled": true, + "status": { "plugged": 1, "speed": 0, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + } + ] +} diff --git a/fixtures/userdata/NanoStation_loco_M5_v6.3.16_XM_sta.json b/fixtures/userdata/NanoStation_loco_M5_v6.3.16_XM_sta.json new file mode 100644 index 0000000..8d638fc --- /dev/null +++ b/fixtures/userdata/NanoStation_loco_M5_v6.3.16_XM_sta.json @@ -0,0 +1,121 @@ +{ + "host": { + "uptime": 1974859, + "time": "2025-10-20 15:33:05", + "fwversion": "v6.3.16", + "fwprefix": "XM", + "hostname": "NanoStation loco M5 Client", + "devmodel": "NanoStation loco M5 ", + "netrole": "bridge", + "totalram": 30220288, + "freeram": 8753152, + "cpuload": 100.0, + "cputotal": 197455604, + "cpubusy": 11150046 + }, + "wireless": { + "mode": "sta", + "essid": "SOMETHING", + "hide_essid": 0, + "apmac": "XX:XX:XX:XX:XX:XX", + "countrycode": 616, + "channel": 140, + "frequency": "5700 MHz", + "dfs": "0", + "opmode": "11naht40minus", + "antenna": "Built in - 13 dBi", + "chains": "2X2", + "signal": -46, + "rssi": 50, + "noisef": -89, + "txpower": 2, + "ack": 28, + "distance": 600, + "ccq": 738, + "txrate": "243", + "rxrate": "180", + "security": "WPA2", + "qos": "No QoS", + "rstatus": 5, + "count": 1, + "cac_nol": 0, + "nol_chans": 0, + "polling": { + "enabled": 1, + "quality": 84, + "capacity": 55, + "priority": 0, + "noack": 0, + "airsync_mode": 0, + "airsync_connections": 0, + "airsync_down_util": 0, + "airsync_up_util": 0, + "airselect": 0, + "airselect_interval": 1000, + "ff_mode": 0, + "ff_duration": 0, + "ff_dlulratio": 0, + "atpc_status": 0 + }, + "stats": { + "rx_nwids": 0, + "rx_crypts": 0, + "rx_frags": 0, + "tx_retries": 0, + "missed_beacons": 0, + "err_other": 0 + }, + "wds": 1, + "aprepeater": 0, + "chwidth": 20, + "chanbw": 40, + "cwmmode": 0, + "rx_chainmask": 3, + "tx_chainmask": 3, + "chainrssi": [50, 47, 0], + "chainrssimgmt": [50, 47, 0], + "chainrssiext": [50, 47, 0] + }, + "airview": { "enabled": 0 }, + "services": { "dhcpc": 0, "dhcpd": 0, "pppoe": 0 }, + "firewall": { "iptables": 0, "ebtables": 1, "ip6tables": 0, "eb6tables": 0 }, + "genuine": "/images/genuine.png", + "unms": { "status": 0, "timestamp": "", "link": "" }, + "interfaces": [ + { + "ifname": "lo", + "hwaddr": "00:00:00:00:00:00", + "enabled": true, + "status": { "plugged": 1, "speed": 0, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "eth0", + "hwaddr": "YY:YY:YY:YY:YY:YY", + "enabled": true, + "status": { "plugged": 0, "speed": 0, "duplex": 0 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "wifi0", + "hwaddr": "YY:YY:YY:YY:YY:YY", + "enabled": true, + "status": { "plugged": 1, "speed": 0, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "ath0", + "hwaddr": "YY:YY:YY:YY:YY:YY", + "enabled": true, + "status": { "plugged": 1, "speed": 300, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + }, + { + "ifname": "br0", + "hwaddr": "YY:YY:YY:YY:YY:YY", + "enabled": true, + "status": { "plugged": 1, "speed": 0, "duplex": 255 }, + "services": { "dhcpc": false, "dhcpd": false, "pppoe": false } + } + ] +} diff --git a/pyproject.toml b/pyproject.toml index f2b1ea9..9732b65 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "airos" -version = "0.5.6" +version = "0.6.0" license = "MIT" description = "Ubiquiti airOS module(s) for Python 3." readme = "README.md" diff --git a/script/generate_ha_fixture.py b/script/generate_ha_fixture.py index d8b8dc1..25acc78 100644 --- a/script/generate_ha_fixture.py +++ b/script/generate_ha_fixture.py @@ -67,14 +67,10 @@ def generate_airos_fixtures() -> None: new_data: AirOS6Data | AirOS8Data if fw_major == 6: - derived_data = AirOS6._derived_data_helper( # noqa: SLF001 - source_data, AirOS6.derived_wireless_data - ) + derived_data = AirOS6.derived_data(source_data) new_data = AirOS6Data.from_dict(derived_data) else: - derived_data = AirOS8._derived_data_helper( # noqa: SLF001 - source_data, AirOS8.derived_wireless_data - ) + derived_data = AirOS8.derived_data(source_data) new_data = AirOS8Data.from_dict(derived_data) with open(new_fixture_path, "w", encoding="utf-8") as new: # noqa: PTH123 diff --git a/script/mashumaro-step-debug.py b/script/mashumaro-step-debug.py index 540cc8b..a35c248 100644 --- a/script/mashumaro-step-debug.py +++ b/script/mashumaro-step-debug.py @@ -70,11 +70,16 @@ def main() -> None: if fw_major == 6: wireless_data_prepped = Wireless6.__pre_deserialize__(wireless_data.copy()) else: - wireless_data_prepped = Wireless.__pre_deserialize__(wireless_data.copy()) # noqa: F841 + wireless_data_prepped = Wireless.__pre_deserialize__(wireless_data.copy()) _LOGGER.info( " Success! Wireless enums (mode, ieeemode, security) are valid." ) + _LOGGER.info( + " -> Proving Wireless enums via ieeemode: %s", + wireless_data_prepped["ieeemode"], + ) + if fw_major >= 8: _LOGGER.info(" -> Checking list of Station objects...") station_list_data = wireless_data["sta"] @@ -117,7 +122,8 @@ def main() -> None: if fw_major == 6: _LOGGER.info("Deriving AirOS6Data from object...") derived_data = AirOS6._derived_data_helper( # noqa: SLF001 - data, AirOS6.derived_wireless_data + data, + AirOS6._derived_wireless_data, # noqa: SLF001 ) _LOGGER.info("Attempting to deserialize full AirOS6Data object...") airos_data_obj = AirOS6Data.from_dict(derived_data) @@ -125,7 +131,8 @@ def main() -> None: else: _LOGGER.info("Deriving AirOS8Data from object...") derived_data = AirOS8._derived_data_helper( # noqa: SLF001 - data, AirOS8.derived_wireless_data + data, + AirOS8._derived_wireless_data, # noqa: SLF001 ) _LOGGER.info("Attempting to deserialize full AirOS8Data object...") airos_data_obj = AirOS8Data.from_dict(derived_data) @@ -138,6 +145,17 @@ def main() -> None: _LOGGER.exception("The program failed") _LOGGER.info("------------------\n") + # Reload the JSON data + with open(sys.argv[1], encoding="utf-8") as f: # noqa: PTH123 + data = json.loads(f.read()) + + _LOGGER.info("Adding derived data") + if fw_major == 8: + derived_data = AirOS8.derived_data(data) + if fw_major == 6: + derived_data = AirOS6.derived_data(data) + _LOGGER.info("Full serialisation check: %s", derived_data) + if __name__ == "__main__": main() diff --git a/script/test.sh b/script/test.sh new file mode 100755 index 0000000..8b820d6 --- /dev/null +++ b/script/test.sh @@ -0,0 +1,2 @@ +#!/usr/bin/env sh +pytest tests --cov=airos --cov-report term-missing "${1}" diff --git a/tests/test_airos6.py b/tests/test_airos6.py index 2489957..e38f4f7 100644 --- a/tests/test_airos6.py +++ b/tests/test_airos6.py @@ -122,9 +122,12 @@ async def test_derived_data_no_interfaces_key(airos6_device: AirOS6) -> None: async def test_derived_data_no_br0_eth0_ath0(airos6_device: AirOS6) -> None: """Test derived_data() with an unexpected interface list, to test the fallback logic.""" fixture_data = { + "host": { + "fwversion": "v8.0.0", + }, "interfaces": [ {"ifname": "wan0", "enabled": True, "hwaddr": "11:22:33:44:55:66"} - ] + ], } adjusted_data = airos6_device.derived_data(fixture_data) diff --git a/tests/test_airos8.py b/tests/test_airos8.py index 17bb263..97bae8f 100644 --- a/tests/test_airos8.py +++ b/tests/test_airos8.py @@ -123,9 +123,12 @@ async def test_derived_data_no_interfaces_key(airos8_device: AirOS8) -> None: async def test_derived_data_no_br0_eth0_ath0(airos8_device: AirOS8) -> None: """Test derived_data() with an unexpected interface list, to test the fallback logic.""" fixture_data = { + "host": { + "fwversion": "v8.0.0", + }, "interfaces": [ {"ifname": "wan0", "enabled": True, "hwaddr": "11:22:33:44:55:66"} - ] + ], } adjusted_data = airos8_device.derived_data(fixture_data) @@ -254,7 +257,7 @@ async def test_status_missing_required_key_in_json(airos8_device: AirOS8) -> Non # Fixture is valid JSON, but is missing the entire 'wireless' block, # which is a required field for the AirOS8Data dataclass. invalid_data = { - "host": {"hostname": "test"}, + "host": {"hostname": "test", "fwversion": "v8.0.0"}, "interfaces": [ {"ifname": "br0", "hwaddr": "11:22:33:44:55:66", "enabled": True} ], diff --git a/tests/test_data.py b/tests/test_data.py index 7120d47..b20c1f1 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -48,7 +48,7 @@ async def test_unknown_enum_values() -> None: format_string, "unsupported_mode", "Wireless", "mode" ) mock_warning.assert_any_call( - format_string, "unsupported_ieee", "Wireless", "ieeemode" + format_string, "unsupported_ieee".upper(), "Wireless", "ieeemode" ) mock_warning.assert_any_call( format_string, "unsupported_security", "Wireless", "security" diff --git a/tests/test_helpers.py b/tests/test_helpers.py deleted file mode 100644 index d6bd092..0000000 --- a/tests/test_helpers.py +++ /dev/null @@ -1,137 +0,0 @@ -"""Test helpers for Ubiquiti airOS devices.""" - -from typing import Any -from unittest.mock import AsyncMock, MagicMock, patch - -import aiohttp -import pytest - -from airos.airos8 import AirOS8 -from airos.exceptions import AirOSKeyDataMissingError -from airos.helpers import DetectDeviceData, async_get_firmware_data - -# pylint: disable=redefined-outer-name - - -@pytest.fixture -def mock_session() -> MagicMock: - """Return a mock aiohttp ClientSession.""" - return MagicMock(spec=aiohttp.ClientSession) - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - ( - "mock_response", - "expected_fw_major", - "expected_mac", - "expected_hostname", - "expected_exception", - ), - [ - # Success case for AirOS 8 - ( - { - "host": {"fwversion": "v8.7.4", "hostname": "test-host-8"}, - "interfaces": [ - {"hwaddr": "AA:BB:CC:DD:EE:FF", "ifname": "br0", "enabled": True} - ], - }, - 8, - "AA:BB:CC:DD:EE:FF", - "test-host-8", - None, - ), - # Success case for AirOS 6 - ( - { - "host": {"fwversion": "v6.3.16", "hostname": "test-host-6"}, - "wireless": {"mode": "sta", "apmac": "11:22:33:44:55:66"}, - "interfaces": [ - {"hwaddr": "11:22:33:44:55:66", "ifname": "br0", "enabled": True} - ], - }, - 6, - "11:22:33:44:55:66", - "test-host-6", - None, - ), - # Failure case: Missing host key - ({"wireless": {}}, 0, "", "", AirOSKeyDataMissingError), - # Failure case: Missing fwversion key - ( - {"host": {"hostname": "test-host"}, "interfaces": []}, - 0, - "", - "", - AirOSKeyDataMissingError, - ), - # Failure case: Invalid fwversion value - ( - { - "host": {"fwversion": "not-a-number", "hostname": "test-host"}, - "interfaces": [], - }, - 0, - "", - "", - AirOSKeyDataMissingError, - ), - # Failure case: Missing hostname key - ( - {"host": {"fwversion": "v8.7.4"}, "interfaces": []}, - 0, - "", - "", - AirOSKeyDataMissingError, - ), - # Failure case: Missing MAC address - ( - {"host": {"fwversion": "v8.7.4", "hostname": "test-host"}}, - 0, - "", - "", - AirOSKeyDataMissingError, - ), - ], -) -async def test_firmware_detection( - mock_session: aiohttp.ClientSession, - mock_response: dict[str, Any], - expected_fw_major: int, - expected_mac: str, - expected_hostname: str, - expected_exception: Any, -) -> None: - """Test helper firmware detection.""" - - mock_request_json = AsyncMock( - side_effect=[ - {}, # First call for login() - mock_response, # Second call for the status() endpoint - ] - ) - - with patch.object(AirOS8, "_request_json", new=mock_request_json): - if expected_exception: - with pytest.raises(expected_exception): - await async_get_firmware_data( - host="192.168.1.3", - username="testuser", - password="testpassword", - session=mock_session, - use_ssl=True, - ) - else: - # Test the success case - device_data: DetectDeviceData = await async_get_firmware_data( - host="192.168.1.3", - username="testuser", - password="testpassword", - session=mock_session, - use_ssl=True, - ) - - assert device_data["fw_major"] == expected_fw_major - assert device_data["mac"] == expected_mac - assert device_data["hostname"] == expected_hostname diff --git a/tests/test_stations6.py b/tests/test_stations6.py index 9a1d762..f93ff3e 100644 --- a/tests/test_stations6.py +++ b/tests/test_stations6.py @@ -4,7 +4,7 @@ import json import os from typing import Any -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import aiofiles import pytest @@ -29,6 +29,8 @@ async def _read_fixture(fixture: str = "NanoStation_M5_sta_v6.3.16") -> Any: @pytest.mark.parametrize( ("mode", "fixture"), [ + ("sta", "NanoStation_loco_M5_v6.3.16_XM_sta"), + ("ap", "NanoStation_loco_M5_v6.3.16_XM_ap"), ("sta", "NanoStation_M5_sta_v6.3.16"), ], ) @@ -65,3 +67,53 @@ async def test_ap_object( cookie = SimpleCookie() cookie["session_id"] = "test-cookie" cookie["AIROS_TOKEN"] = "abc123" + + +@pytest.mark.asyncio +async def test_login_v6_flow() -> None: + """Test AirOS v6 XM login flow with manual cookie handling.""" + + # Create a mock session + session = MagicMock() + + # Mock response for GET /login.cgi + get_login_response = MagicMock() + get_login_response.__aenter__.return_value = get_login_response + get_login_response.status = 200 + get_login_response.cookies = { + "AIROS_ABC123": MagicMock(key="AIROS_ABC123", value="xyz789") + } + + # Mock response for POST /login.cgi + post_login_response = MagicMock() + post_login_response.__aenter__.return_value = post_login_response + post_login_response.status = 302 + + # Mock response for GET /index.cgi + get_index_response = MagicMock() + get_index_response.__aenter__.return_value = get_index_response + get_index_response.status = 200 + get_index_response.url = "http://192.168.1.3/index.cgi" + + # Set side effects for session.request + session.request.side_effect = [ + get_login_response, + post_login_response, + get_index_response, + ] + + # Create device + airos6_device = AirOS6( + host="http://192.168.1.3", + username="ubnt", + password="ubnt", + session=session, + ) + + await airos6_device._login_v6() # noqa: SLF001 + + # Assertions + assert airos6_device.connected is True + assert airos6_device.api_version == 6 + assert airos6_device._auth_cookie == "AIROS_ABC123=xyz789" # noqa: SLF001 + assert session.request.call_count == 3