Emit camelCase keys in as_json() to match OTBR REST API spec

python_otbr_api/models.py

@@ -44,6 +44,9 @@
 }
 
 
+_PASCAL_TO_CAMEL: dict[str, str] = {v: k for k, v in _CAMEL_TO_PASCAL.items()}
+
+
 def _normalize_keys(data: Any) -> Any:
     """Normalize camelCase JSON keys to PascalCase.
 
@@ -59,6 +62,27 @@ def _normalize_keys(data: Any) -> Any:
     }
 
 
+def _to_dual_keys(data: dict) -> dict:
+    """Emit both PascalCase and camelCase variants for known keys.
+
+    Recent ot-br-posix versions switched the REST API to camelCase (per the
+    OpenAPI spec), but older versions still expect PascalCase. Both versions
+    parse incoming payloads case-sensitively and silently ignore unknown keys,
+    so we emit both capitalization formats for every known key. This keeps the
+    library compatible with both old and new border routers.
+
+    Recursion is not needed: nested dicts come from nested ``as_json()`` calls
+    that already return dual-key dicts.
+    """
+    result: dict = {}
+    for k, v in data.items():
+        result[k] = v
+        other = _PASCAL_TO_CAMEL.get(k) or _CAMEL_TO_PASCAL.get(k)
+        if other is not None and other != k:
+            result[other] = v
+    return result
+
+
 @dataclass
 class Timestamp:
     """Timestamp."""
@@ -84,7 +108,7 @@ def as_json(self) -> dict:
             result["Seconds"] = self.seconds
         if self.ticks is not None:
             result["Ticks"] = self.ticks
-        return result
+        return _to_dual_keys(result)
 
     @classmethod
     def from_json(cls, json_data: Any) -> Timestamp:
@@ -151,7 +175,7 @@ def as_json(self) -> dict:
             result["Routers"] = self.routers
         if self.to_ble_link is not None:
             result["TobleLink"] = self.to_ble_link
-        return result
+        return _to_dual_keys(result)
 
     @classmethod
     def from_json(cls, json_data: Any) -> SecurityPolicy:
@@ -225,7 +249,7 @@ def as_json(self) -> dict:
             result["PSKc"] = self.psk_c
         if self.security_policy is not None:
             result["SecurityPolicy"] = self.security_policy.as_json()
-        return result
+        return _to_dual_keys(result)
 
     @classmethod
     def from_json(cls, json_data: Any) -> ActiveDataSet:
@@ -278,7 +302,7 @@ def as_json(self) -> dict:
             result["Delay"] = self.delay
         if self.pending_timestamp is not None:
             result["PendingTimestamp"] = self.pending_timestamp.as_json()
-        return result
+        return _to_dual_keys(result)
 
     @classmethod
     def from_json(cls, json_data: Any) -> PendingDataSet:

tests/test_init.py

@@ -5,9 +5,20 @@
 
 import pytest
 import python_otbr_api
+from python_otbr_api.models import _to_dual_keys
 
 from tests.test_util.aiohttp import AiohttpClientMocker
 
+
+def _dual(data: dict) -> dict:
+    """Expand a camelCase payload to the dual-key format emitted by as_json()."""
+    result = _to_dual_keys(data)
+    for key in list(result):
+        if isinstance(result[key], dict):
+            result[key] = _dual(result[key])
+    return result
+
+
 BASE_URL = "http://core-silabs-multiprotocol:8081"
 DATASET_JSON: dict[str, Any] = {
     "ActiveTimestamp": {
@@ -170,7 +181,34 @@ async def test_get_active_dataset(aioclient_mock: AiohttpClientMocker):
         DATASET_JSON["PSKc"],
         security_policy,
     )
-    assert active_dataset.as_json() == DATASET_JSON
+    # as_json() emits both PascalCase and camelCase keys for compatibility with
+    # both older and newer ot-br-posix REST API versions.
+    expected = _dual(
+        {
+            "activeTimestamp": {"authoritative": False, "seconds": 1, "ticks": 0},
+            "channelMask": 134215680,
+            "channel": 15,
+            "extPanId": "8478E3379E047B92",
+            "meshLocalPrefix": "fd89:bde7:42ed:a901::/64",
+            "networkKey": "96271D6ECC78749114AB6A591E0D06F1",
+            "networkName": "OpenThread HA",
+            "panId": 33991,
+            "pskc": "9760C89414D461AC717DCD105EB87E5B",
+            "securityPolicy": {
+                "autonomousEnrollment": False,
+                "commercialCommissioning": False,
+                "externalCommissioning": True,
+                "nativeCommissioning": True,
+                "networkKeyProvisioning": False,
+                "nonCcmRouters": False,
+                "obtainNetworkKey": True,
+                "rotationTime": 672,
+                "routers": True,
+                "tobleLink": True,
+            },
+        }
+    )
+    assert active_dataset.as_json() == expected
 
 
 async def test_get_active_dataset_empty(aioclient_mock: AiohttpClientMocker):
@@ -245,18 +283,20 @@ async def test_create_active_dataset(aioclient_mock: AiohttpClientMocker):
     assert aioclient_mock.call_count == 2
     assert aioclient_mock.mock_calls[-1][0] == "PUT"
     assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/active"
-    assert aioclient_mock.mock_calls[-1][2] == {"NetworkName": "OpenThread HA"}
+    assert aioclient_mock.mock_calls[-1][2] == _dual({"networkName": "OpenThread HA"})
 
     await otbr.create_active_dataset(
         python_otbr_api.ActiveDataSet(network_name="OpenThread HA", channel=15)
     )
     assert aioclient_mock.call_count == 3
     assert aioclient_mock.mock_calls[-1][0] == "PUT"
     assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/active"
-    assert aioclient_mock.mock_calls[-1][2] == {
-        "NetworkName": "OpenThread HA",
-        "Channel": 15,
-    }
+    assert aioclient_mock.mock_calls[-1][2] == _dual(
+        {
+            "networkName": "OpenThread HA",
+            "channel": 15,
+        }
+    )
 
 
 async def test_delete_active_dataset(aioclient_mock: AiohttpClientMocker):
@@ -294,13 +334,15 @@ async def test_create_pending_dataset(aioclient_mock: AiohttpClientMocker):
     assert aioclient_mock.call_count == 2
     assert aioclient_mock.mock_calls[-1][0] == "PUT"
     assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/pending"
-    assert aioclient_mock.mock_calls[-1][2] == {
-        "ActiveDataset": {
-            "NetworkName": "OpenThread HA",
-        },
-        "Delay": 12345,
-        "PendingTimestamp": {},
-    }
+    assert aioclient_mock.mock_calls[-1][2] == _dual(
+        {
+            "activeDataset": {
+                "networkName": "OpenThread HA",
+            },
+            "delay": 12345,
+            "pendingTimestamp": {},
+        }
+    )
 
     await otbr.create_pending_dataset(
         python_otbr_api.PendingDataSet(
@@ -311,13 +353,15 @@ async def test_create_pending_dataset(aioclient_mock: AiohttpClientMocker):
     assert aioclient_mock.call_count == 3
     assert aioclient_mock.mock_calls[-1][0] == "PUT"
     assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/pending"
-    assert aioclient_mock.mock_calls[-1][2] == {
-        "ActiveDataset": {
-            "Channel": 15,
-            "NetworkName": "OpenThread HA",
-        },
-        "Delay": 23456,
-    }
+    assert aioclient_mock.mock_calls[-1][2] == _dual(
+        {
+            "activeDataset": {
+                "channel": 15,
+                "networkName": "OpenThread HA",
+            },
+            "delay": 23456,
+        }
+    )
 
 
 async def test_delete_pending_dataset(aioclient_mock: AiohttpClientMocker):
@@ -340,24 +384,17 @@ async def test_set_channel(aioclient_mock: AiohttpClientMocker) -> None:
     aioclient_mock.get(f"{BASE_URL}/node/dataset/active", json=DATASET_JSON)
     aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CREATED)
     new_channel = 16
-    expected_active_timestamp = DATASET_JSON["ActiveTimestamp"] | {"Seconds": 2}
-    expected_pending_dataset = {
-        "ActiveDataset": DATASET_JSON
-        | {
-            "ActiveTimestamp": expected_active_timestamp,
-            "Channel": new_channel,
-        },
-        "Delay": 1234,
-    }
-
     assert new_channel != DATASET_JSON["Channel"]
     await otbr.set_channel(new_channel, 1234)
     assert aioclient_mock.call_count == 2
     assert aioclient_mock.mock_calls[0][0] == "GET"
     assert aioclient_mock.mock_calls[0][1].path == "/node/dataset/active"
     assert aioclient_mock.mock_calls[1][0] == "PUT"
     assert aioclient_mock.mock_calls[1][1].path == "/node/dataset/pending"
-    assert aioclient_mock.mock_calls[1][2] == expected_pending_dataset
+    pending = aioclient_mock.mock_calls[1][2]
+    assert pending["delay"] == 1234
+    assert pending["activeDataset"]["channel"] == new_channel
+    assert pending["activeDataset"]["activeTimestamp"]["seconds"] == 2
 
 
 async def test_set_channel_default_delay(aioclient_mock: AiohttpClientMocker) -> None:
@@ -367,24 +404,17 @@ async def test_set_channel_default_delay(aioclient_mock: AiohttpClientMocker) ->
     aioclient_mock.get(f"{BASE_URL}/node/dataset/active", json=DATASET_JSON)
     aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CREATED)
     new_channel = 16
-    expected_active_timestamp = DATASET_JSON["ActiveTimestamp"] | {"Seconds": 2}
-    expected_pending_dataset = {
-        "ActiveDataset": DATASET_JSON
-        | {
-            "ActiveTimestamp": expected_active_timestamp,
-            "Channel": new_channel,
-        },
-        "Delay": 300000,
-    }
-
     assert new_channel != DATASET_JSON["Channel"]
     await otbr.set_channel(new_channel)
     assert aioclient_mock.call_count == 2
     assert aioclient_mock.mock_calls[0][0] == "GET"
     assert aioclient_mock.mock_calls[0][1].path == "/node/dataset/active"
     assert aioclient_mock.mock_calls[1][0] == "PUT"
     assert aioclient_mock.mock_calls[1][1].path == "/node/dataset/pending"
-    assert aioclient_mock.mock_calls[1][2] == expected_pending_dataset
+    pending = aioclient_mock.mock_calls[1][2]
+    assert pending["delay"] == 300000
+    assert pending["activeDataset"]["channel"] == new_channel
+    assert pending["activeDataset"]["activeTimestamp"]["seconds"] == 2
 
 
 async def test_set_channel_no_timestamp(aioclient_mock: AiohttpClientMocker) -> None:
@@ -397,24 +427,17 @@ async def test_set_channel_no_timestamp(aioclient_mock: AiohttpClientMocker) ->
     aioclient_mock.get(f"{BASE_URL}/node/dataset/active", json=dataset_json)
     aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CREATED)
     new_channel = 16
-    expected_active_timestamp = {"Authoritative": False, "Seconds": 1, "Ticks": 0}
-    expected_pending_dataset = {
-        "ActiveDataset": DATASET_JSON
-        | {
-            "ActiveTimestamp": expected_active_timestamp,
-            "Channel": new_channel,
-        },
-        "Delay": 300000,
-    }
-
     assert new_channel != DATASET_JSON["Channel"]
     await otbr.set_channel(new_channel)
     assert aioclient_mock.call_count == 2
     assert aioclient_mock.mock_calls[0][0] == "GET"
     assert aioclient_mock.mock_calls[0][1].path == "/node/dataset/active"
     assert aioclient_mock.mock_calls[1][0] == "PUT"
     assert aioclient_mock.mock_calls[1][1].path == "/node/dataset/pending"
-    assert aioclient_mock.mock_calls[1][2] == expected_pending_dataset
+    pending = aioclient_mock.mock_calls[1][2]
+    assert pending["delay"] == 300000
+    assert pending["activeDataset"]["channel"] == new_channel
+    assert pending["activeDataset"]["activeTimestamp"]["seconds"] == 1
 
 
 async def test_set_channel_invalid_channel(aioclient_mock: AiohttpClientMocker) -> None:

tests/test_models.py

@@ -1,6 +1,7 @@
 """Test data models."""
 
 import python_otbr_api
+from python_otbr_api.models import SecurityPolicy
 
 
 def test_deserialize_pending_dataset():
@@ -66,3 +67,89 @@ def test_deserialize_pending_dataset_camelcase():
         12345,
         python_otbr_api.Timestamp(),
     )
+
+
+def test_serialize_active_dataset_dual_keys():
+    """Test that as_json() emits both PascalCase and camelCase keys.
+
+    ot-br-posix's REST API parses incoming payloads case-sensitively but
+    silently ignores unknown keys. Emitting both capitalizations keeps the
+    library compatible with both older routers (PascalCase) and newer ones
+    (camelCase, per the OpenAPI spec).
+    """
+    dataset = python_otbr_api.ActiveDataSet(
+        active_timestamp=python_otbr_api.Timestamp(
+            authoritative=False, seconds=1, ticks=0
+        ),
+        network_key="00112233445566778899aabbccddeeff",
+        network_name="OpenThread-1234",
+        extended_pan_id="dead00beef00cafe",
+        mesh_local_prefix="fd11:2222:3333::/64",
+        pan_id=12345,
+        channel=15,
+        psk_c="aabbccddeeff00112233445566778899",
+        security_policy=SecurityPolicy(
+            rotation_time=672, obtain_network_key=True, routers=True
+        ),
+        channel_mask=134215680,
+    )
+    result = dataset.as_json()
+    # Top-level keys must be present in BOTH capitalizations.
+    for camel, pascal in (
+        ("activeTimestamp", "ActiveTimestamp"),
+        ("networkKey", "NetworkKey"),
+        ("networkName", "NetworkName"),
+        ("extPanId", "ExtPanId"),
+        ("meshLocalPrefix", "MeshLocalPrefix"),
+        ("panId", "PanId"),
+        ("channel", "Channel"),
+        ("pskc", "PSKc"),
+        ("securityPolicy", "SecurityPolicy"),
+        ("channelMask", "ChannelMask"),
+    ):
+        assert camel in result, f"missing camelCase key {camel}"
+        assert pascal in result, f"missing PascalCase key {pascal}"
+        assert result[camel] == result[pascal]
+    # Nested dicts must also expose both capitalizations.
+    for camel, pascal in (("seconds", "Seconds"), ("ticks", "Ticks")):
+        assert camel in result["activeTimestamp"]
+        assert pascal in result["activeTimestamp"]
+    for camel, pascal in (
+        ("rotationTime", "RotationTime"),
+        ("obtainNetworkKey", "ObtainNetworkKey"),
+        ("routers", "Routers"),
+    ):
+        assert camel in result["securityPolicy"]
+        assert pascal in result["securityPolicy"]
+
+
+def test_serialize_pending_dataset_dual_keys():
+    """Test that PendingDataSet.as_json() emits both capitalizations."""
+    pending = python_otbr_api.PendingDataSet(
+        active_dataset=python_otbr_api.ActiveDataSet(network_name="OpenThread HA"),
+        delay=30000,
+        pending_timestamp=python_otbr_api.Timestamp(seconds=2, ticks=0),
+    )
+    result = pending.as_json()
+    for camel, pascal in (
+        ("activeDataset", "ActiveDataset"),
+        ("delay", "Delay"),
+        ("pendingTimestamp", "PendingTimestamp"),
+    ):
+        assert camel in result
+        assert pascal in result
+    assert "networkName" in result["activeDataset"]
+    assert "NetworkName" in result["activeDataset"]
+
+
+def test_roundtrip_dual_keys():
+    """Test that from_json(as_json(x)) preserves data.
+
+    from_json() must cope with dual-key input (both PascalCase and camelCase
+    present at once), which is what as_json() now emits.
+    """
+    original = python_otbr_api.ActiveDataSet(
+        network_name="Test", channel=15, pan_id=4660
+    )
+    roundtripped = python_otbr_api.ActiveDataSet.from_json(original.as_json())
+    assert roundtripped == original