Merge pull request #52 from CoMPaTech/validate_fixture

Fix age field in Remote for nanostation 8.7.11

Refactor

Improved the fixture generation script for better maintainability and dynamic processing of JSON files.
Updated internal data structure inheritance for consistency.
Style

Expanded numeric fields in several data classes to accept integer and null values for improved flexibility.

Author: CoMPaTech

airos/data.py

@@ -92,6 +92,12 @@ def _redact(d: dict):
 # Data class start
 
 
+class AirOSDataClass(DataClassDictMixin):
+    """A base class for all mashumaro dataclasses."""
+
+    pass
+
+
 def _check_and_log_unknown_enum_value(
     data_dict: dict[str, Any],
     key: str,
@@ -149,15 +155,15 @@ class NetRole(Enum):
 
 
 @dataclass
-class ChainName:
+class ChainName(AirOSDataClass):
     """Leaf definition."""
 
     number: int
     name: str
 
 
 @dataclass
-class Host:
+class Host(AirOSDataClass):
     """Leaf definition."""
 
     hostname: str
@@ -169,11 +175,11 @@ class Host:
     fwversion: str
     devmodel: str
     netrole: NetRole
-    loadavg: float
+    loadavg: float | int | None
     totalram: int
     freeram: int
     temperature: int
-    cpuload: float
+    cpuload: float | int | None
     height: int | None  # Reported none on LiteBeam 5AC
 
     @classmethod
@@ -184,7 +190,7 @@ def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
 
 
 @dataclass
-class Services:
+class Services(AirOSDataClass):
     """Leaf definition."""
 
     dhcpc: bool
@@ -195,7 +201,7 @@ class Services:
 
 
 @dataclass
-class Firewall:
+class Firewall(AirOSDataClass):
     """Leaf definition."""
 
     iptables: bool
@@ -205,23 +211,23 @@ class Firewall:
 
 
 @dataclass
-class Throughput:
+class Throughput(AirOSDataClass):
     """Leaf definition."""
 
     tx: int
     rx: int
 
 
 @dataclass
-class ServiceTime:
+class ServiceTime(AirOSDataClass):
     """Leaf definition."""
 
     time: int
     link: int
 
 
 @dataclass
-class Polling:
+class Polling(AirOSDataClass):
     """Leaf definition."""
 
     cb_capacity: int
@@ -238,7 +244,7 @@ class Polling:
 
 
 @dataclass
-class Stats:
+class Stats(AirOSDataClass):
     """Leaf definition."""
 
     rx_bytes: int
@@ -250,7 +256,7 @@ class Stats:
 
 
 @dataclass
-class EvmData:
+class EvmData(AirOSDataClass):
     """Leaf definition."""
 
     usage: int
@@ -259,7 +265,7 @@ class EvmData:
 
 
 @dataclass
-class Airmax:
+class Airmax(AirOSDataClass):
     """Leaf definition."""
 
     actual_priority: int
@@ -274,7 +280,7 @@ class Airmax:
 
 
 @dataclass
-class EthList:
+class EthList(AirOSDataClass):
     """Leaf definition."""
 
     ifname: str
@@ -287,38 +293,37 @@ class EthList:
 
 
 @dataclass
-class GPSData:
+class GPSData(AirOSDataClass):
     """Leaf definition."""
 
-    lat: float | None = None
-    lon: float | None = None
+    lat: float | int | None = None
+    lon: float | int | None = None
     fix: int | None = None
     sats: int | None = None  # LiteAP GPS
     dim: int | None = None  # LiteAP GPS
-    dop: float | None = None  # LiteAP GPS
-    alt: float | None = None  # LiteAP GPS
+    dop: float | int | None = None  # LiteAP GPS
+    alt: float | int | None = None  # LiteAP GPS
     time_synced: int | None = None  # LiteAP GPS
 
 
 @dataclass
-class UnmsStatus:
+class UnmsStatus(AirOSDataClass):
     """Leaf definition."""
 
     status: int
     timestamp: str | None = None
 
 
 @dataclass
-class Remote:
+class Remote(AirOSDataClass):
     """Leaf definition."""
 
-    age: int
     device_id: str
     hostname: str
     platform: str
     version: str
     time: str
-    cpuload: float
+    cpuload: float | int | None
     temperature: int
     totalram: int
     freeram: int
@@ -351,6 +356,7 @@ class Remote:
     mode: WirelessMode | None = None  # Investigate why remotes can have no mode set
     ip6addr: list[str] | None = None  # For v4 only devices
     height: int | None = None
+    age: int | None = None  # At least not present on 8.7.11
 
     @classmethod
     def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
@@ -360,7 +366,7 @@ def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
 
 
 @dataclass
-class Disconnected:
+class Disconnected(AirOSDataClass):
     """Leaf definition for disconnected devices."""
 
     mac: str
@@ -374,7 +380,7 @@ class Disconnected:
 
 
 @dataclass
-class Station:
+class Station(AirOSDataClass):
     """Leaf definition for connected/active devices."""
 
     mac: str
@@ -413,7 +419,7 @@ class Station:
 
 
 @dataclass
-class Wireless:
+class Wireless(AirOSDataClass):
     """Leaf definition."""
 
     essid: str
@@ -465,7 +471,7 @@ def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]:
 
 
 @dataclass
-class InterfaceStatus:
+class InterfaceStatus(AirOSDataClass):
     """Leaf definition."""
 
     plugged: bool
@@ -486,7 +492,7 @@ class InterfaceStatus:
 
 
 @dataclass
-class Interface:
+class Interface(AirOSDataClass):
     """Leaf definition."""
 
     ifname: str
@@ -497,30 +503,30 @@ class Interface:
 
 
 @dataclass
-class ProvisioningMode:
+class ProvisioningMode(AirOSDataClass):
     """Leaf definition."""
 
     pass
 
 
 @dataclass
-class NtpClient:
+class NtpClient(AirOSDataClass):
     """Leaf definition."""
 
     pass
 
 
 @dataclass
-class GPSMain:
+class GPSMain(AirOSDataClass):
     """Leaf definition."""
 
-    lat: float
-    lon: float
+    lat: float | int | None
+    lon: float | int | None
     fix: int
 
 
 @dataclass
-class Derived:
+class Derived(AirOSDataClass):
     """Contain custom data generated by this module."""
 
     mac: str  # Base device MAC address (i.e. eth0)
@@ -536,7 +542,7 @@ class Derived:
 
 
 @dataclass
-class AirOS8Data(DataClassDictMixin):
+class AirOS8Data(AirOSDataClass):
     """Dataclass for AirOS v8 devices."""
 
     chain_names: list[ChainName]

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "airos"
-version         = "0.2.6"
+version         = "0.2.7"
 license         = "MIT"
 description     = "Ubiquity airOS module(s) for Python 3."
 readme          = "README.md"

script/generate_ha_fixture.py

@@ -1,4 +1,4 @@
-"""Generate mock airos fixture for testing."""
+"""Generate mock airos fixtures for testing."""
 
 import json
 import logging
@@ -13,52 +13,53 @@
 if project_root_dir not in sys.path:
     sys.path.append(project_root_dir)
 
+# NOTE: This assumes the airos module is correctly installed or available in the project path.
+# If not, you might need to adjust the import statement.
 from airos.airos8 import AirOS, AirOSData  # noqa: E402
 
-# Define the path to save the fixture
-fixture_dir = os.path.join(os.path.dirname(__file__), "../fixtures")
-userdata_dir = os.path.join(os.path.dirname(__file__), "../fixtures/userdata")
-new_fixture_path = os.path.join(fixture_dir, "airos_loco5ac_ap-ptp.json")
-base_fixture_path = os.path.join(userdata_dir, "loco5ac_ap-ptp.json")
-
-with open(base_fixture_path) as source, open(new_fixture_path, "w") as new:
-    source_data = json.loads(source.read())
-    derived_data = AirOS.derived_data(None, source_data)
-    new_data = AirOSData.from_dict(derived_data)
-    json.dump(new_data.to_dict(), new, indent=2, sort_keys=True)
-
-new_fixture_path = os.path.join(fixture_dir, "airos_loco5ac_sta-ptp.json")
-base_fixture_path = os.path.join(userdata_dir, "loco5ac_sta-ptp.json")
-
-with open(base_fixture_path) as source, open(new_fixture_path, "w") as new:
-    source_data = json.loads(source.read())
-    derived_data = AirOS.derived_data(None, source_data)
-    new_data = AirOSData.from_dict(derived_data)
-    json.dump(new_data.to_dict(), new, indent=2, sort_keys=True)
-
-new_fixture_path = os.path.join(fixture_dir, "airos_mocked_sta-ptmp.json")
-base_fixture_path = os.path.join(userdata_dir, "mocked_sta-ptmp.json")
-
-with open(base_fixture_path) as source, open(new_fixture_path, "w") as new:
-    source_data = json.loads(source.read())
-    derived_data = AirOS.derived_data(None, source_data)
-    new_data = AirOSData.from_dict(derived_data)
-    json.dump(new_data.to_dict(), new, indent=2, sort_keys=True)
-
-new_fixture_path = os.path.join(fixture_dir, "airos_liteapgps_ap_ptmp_40mhz.json")
-base_fixture_path = os.path.join(userdata_dir, "liteapgps_ap_ptmp_40mhz.json")
-
-with open(base_fixture_path) as source, open(new_fixture_path, "w") as new:
-    source_data = json.loads(source.read())
-    derived_data = AirOS.derived_data(None, source_data)
-    new_data = AirOSData.from_dict(derived_data)
-    json.dump(new_data.to_dict(), new, indent=2, sort_keys=True)
-
-new_fixture_path = os.path.join(fixture_dir, "airos_nanobeam5ac_sta_ptmp_40mhz.json")
-base_fixture_path = os.path.join(userdata_dir, "nanobeam5ac_sta_ptmp_40mhz.json")
-
-with open(base_fixture_path) as source, open(new_fixture_path, "w") as new:
-    source_data = json.loads(source.read())
-    derived_data = AirOS.derived_data(None, source_data)
-    new_data = AirOSData.from_dict(derived_data)
-    json.dump(new_data.to_dict(), new, indent=2, sort_keys=True)
+
+def generate_airos_fixtures():
+    """Process all (intended) JSON files from the userdata directory to potential fixtures."""
+
+    # Define the paths to the directories
+    fixture_dir = os.path.join(os.path.dirname(__file__), "../fixtures")
+    userdata_dir = os.path.join(os.path.dirname(__file__), "../fixtures/userdata")
+
+    # Ensure the fixture directory exists
+    os.makedirs(fixture_dir, exist_ok=True)
+
+    # Iterate over all files in the userdata_dir
+    for filename in os.listdir(userdata_dir):
+        if "mocked" in filename:
+            continue
+        if filename.endswith(".json"):
+            # Construct the full paths for the base and new fixtures
+            base_fixture_path = os.path.join(userdata_dir, filename)
+            new_filename = f"airos_{filename}"
+            new_fixture_path = os.path.join(fixture_dir, new_filename)
+
+            _LOGGER.info("Processing '%s'...", filename)
+
+            try:
+                with open(base_fixture_path) as source:
+                    source_data = json.loads(source.read())
+
+                derived_data = AirOS.derived_data(None, source_data)
+                new_data = AirOSData.from_dict(derived_data)
+
+                with open(new_fixture_path, "w") as new:
+                    json.dump(new_data.to_dict(), new, indent=2, sort_keys=True)
+
+                _LOGGER.info("Successfully created '%s'", new_filename)
+
+            except json.JSONDecodeError:
+                _LOGGER.error("Skipping '%s': Not a valid JSON file.", filename)
+            except Exception as e:
+                _LOGGER.error("Error processing '%s': %s", filename, e)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+    )
+    generate_airos_fixtures()