diff --git a/python_otbr_api/models.py b/python_otbr_api/models.py index dcf0ee7..b1e9b93 100644 --- a/python_otbr_api/models.py +++ b/python_otbr_api/models.py @@ -7,6 +7,57 @@ import voluptuous as vol # type: ignore[import] +# The OTBR REST API uses camelCase keys (per the OpenAPI spec in ot-br-posix), +# but this library historically expected PascalCase. This mapping normalizes +# camelCase keys to PascalCase so both formats are accepted. +_CAMEL_TO_PASCAL: dict[str, str] = { + # Timestamp + "authoritative": "Authoritative", + "seconds": "Seconds", + "ticks": "Ticks", + # SecurityPolicy + "autonomousEnrollment": "AutonomousEnrollment", + "commercialCommissioning": "CommercialCommissioning", + "externalCommissioning": "ExternalCommissioning", + "nativeCommissioning": "NativeCommissioning", + "networkKeyProvisioning": "NetworkKeyProvisioning", + "nonCcmRouters": "NonCcmRouters", + "obtainNetworkKey": "ObtainNetworkKey", + "rotationTime": "RotationTime", + "routers": "Routers", + "tobleLink": "TobleLink", + # ActiveDataSet + "activeTimestamp": "ActiveTimestamp", + "channelMask": "ChannelMask", + "channel": "Channel", + "extPanId": "ExtPanId", + "meshLocalPrefix": "MeshLocalPrefix", + "networkKey": "NetworkKey", + "networkName": "NetworkName", + "panId": "PanId", + "pskc": "PSKc", + "securityPolicy": "SecurityPolicy", + # PendingDataSet + "activeDataset": "ActiveDataset", + "delay": "Delay", + "pendingTimestamp": "PendingTimestamp", +} + + +def _normalize_keys(data: Any) -> Any: + """Normalize camelCase JSON keys to PascalCase. + + The OTBR REST API returns camelCase keys, but the schemas in this module + expect PascalCase. This function converts known camelCase keys so that both + formats are accepted. Unknown keys and non-dict values pass through unchanged. + """ + if not isinstance(data, dict): + return data + return { + _CAMEL_TO_PASCAL.get(k, k): (_normalize_keys(v) if isinstance(v, dict) else v) + for k, v in data.items() + } + @dataclass class Timestamp: @@ -38,6 +89,7 @@ def as_json(self) -> dict: @classmethod def from_json(cls, json_data: Any) -> Timestamp: """Deserialize from JSON.""" + json_data = _normalize_keys(json_data) cls.SCHEMA(json_data) return cls( json_data.get("Authoritative"), @@ -104,6 +156,7 @@ def as_json(self) -> dict: @classmethod def from_json(cls, json_data: Any) -> SecurityPolicy: """Deserialize from JSON.""" + json_data = _normalize_keys(json_data) cls.SCHEMA(json_data) return cls( json_data.get("AutonomousEnrollment"), @@ -177,6 +230,7 @@ def as_json(self) -> dict: @classmethod def from_json(cls, json_data: Any) -> ActiveDataSet: """Deserialize from JSON.""" + json_data = _normalize_keys(json_data) cls.SCHEMA(json_data) active_timestamp = None security_policy = None @@ -229,6 +283,7 @@ def as_json(self) -> dict: @classmethod def from_json(cls, json_data: Any) -> PendingDataSet: """Deserialize from JSON.""" + json_data = _normalize_keys(json_data) cls.SCHEMA(json_data) active_dataset = None pending_timestamp = None diff --git a/tests/test_models.py b/tests/test_models.py index c79610c..152056b 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -18,3 +18,51 @@ def test_deserialize_pending_dataset(): 12345, python_otbr_api.Timestamp(), ) + + +def test_deserialize_active_dataset_camelcase(): + """Test deserializing with camelCase keys as returned by the OTBR REST API.""" + dataset = python_otbr_api.ActiveDataSet.from_json( + { + "activeTimestamp": {"seconds": 1, "ticks": 0, "authoritative": False}, + "networkKey": "00112233445566778899aabbccddeeff", + "networkName": "OpenThread-1234", + "extPanId": "dead00beef00cafe", + "meshLocalPrefix": "fd11:2222:3333::/64", + "panId": 12345, + "channel": 15, + "pskc": "aabbccddeeff00112233445566778899", + "securityPolicy": { + "rotationTime": 672, + "obtainNetworkKey": True, + "routers": True, + }, + "channelMask": 134215680, + } + ) + assert dataset.channel == 15 + assert dataset.network_name == "OpenThread-1234" + assert dataset.extended_pan_id == "dead00beef00cafe" + assert dataset.pan_id == 12345 + assert dataset.psk_c == "aabbccddeeff00112233445566778899" + assert dataset.active_timestamp == python_otbr_api.Timestamp( + authoritative=False, seconds=1, ticks=0 + ) + assert dataset.security_policy is not None + assert dataset.security_policy.rotation_time == 672 + + +def test_deserialize_pending_dataset_camelcase(): + """Test that camelCase works for PendingDataSet with nested objects.""" + result = python_otbr_api.PendingDataSet.from_json( + { + "activeDataset": {"networkName": "OpenThread HA"}, + "delay": 12345, + "pendingTimestamp": {}, + } + ) + assert result == python_otbr_api.PendingDataSet( + python_otbr_api.ActiveDataSet(network_name="OpenThread HA"), + 12345, + python_otbr_api.Timestamp(), + )