Adapt to upstream changes (#46)

* Adapt to upstream changes

* Improve test coverage

python_otbr_api/__init__.py

@@ -2,12 +2,11 @@
 from __future__ import annotations
 from http import HTTPStatus
 import json
-from typing import Literal
 
 import aiohttp
 import voluptuous as vol  # type:ignore[import]
 
-from .models import OperationalDataSet, Timestamp
+from .models import ActiveDataSet, PendingDataSet, Timestamp
 
 # 5 minutes as recommended by
 # https://github.com/openthread/openthread/discussions/8567#discussioncomment-4468920
@@ -36,7 +35,7 @@ def __init__(
     async def set_enabled(self, enabled: bool) -> None:
         """Enable or disable the router."""
 
-        response = await self._session.post(
+        response = await self._session.put(
             f"{self._url}/node/state",
             json="enable" if enabled else "disable",
             timeout=aiohttp.ClientTimeout(total=10),
@@ -45,7 +44,7 @@ async def set_enabled(self, enabled: bool) -> None:
         if response.status != HTTPStatus.OK:
             raise OTBRError(f"unexpected http status {response.status}")
 
-    async def get_active_dataset(self) -> OperationalDataSet | None:
+    async def get_active_dataset(self) -> ActiveDataSet | None:
         """Get current active operational dataset.
 
         Returns None if there is no active operational dataset.
@@ -63,7 +62,7 @@ async def get_active_dataset(self) -> OperationalDataSet | None:
             raise OTBRError(f"unexpected http status {response.status}")
 
         try:
-            return OperationalDataSet.from_json(await response.json())
+            return ActiveDataSet.from_json(await response.json())
         except (json.JSONDecodeError, vol.Error) as exc:
             raise OTBRError("unexpected API response") from exc
 
@@ -90,43 +89,41 @@ async def get_active_dataset_tlvs(self) -> bytes | None:
         except ValueError as exc:
             raise OTBRError("unexpected API response") from exc
 
-    async def _create_dataset(
-        self, dataset: OperationalDataSet, dataset_type: Literal["active", "pending"]
-    ) -> None:
-        """Create active or pending operational dataset.
+    async def create_active_dataset(self, dataset: ActiveDataSet) -> None:
+        """Create active operational dataset.
 
-        The passed in OperationalDataSet does not need to be fully populated, any fields
+        The passed in ActiveDataSet does not need to be fully populated, any fields
         not set will be automatically set by the open thread border router.
         Raises if the http status is 400 or higher or if the response is invalid.
         """
-        response = await self._session.post(
-            f"{self._url}/node/dataset/{dataset_type}",
+        response = await self._session.put(
+            f"{self._url}/node/dataset/active",
             json=dataset.as_json(),
             timeout=aiohttp.ClientTimeout(total=self._timeout),
         )
 
         if response.status == HTTPStatus.CONFLICT:
             raise ThreadNetworkActiveError
-        if response.status != HTTPStatus.ACCEPTED:
+        if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK):
             raise OTBRError(f"unexpected http status {response.status}")
 
-    async def create_active_dataset(self, dataset: OperationalDataSet) -> None:
-        """Create active operational dataset.
-
-        The passed in OperationalDataSet does not need to be fully populated, any fields
-        not set will be automatically set by the open thread border router.
-        Raises if the http status is 400 or higher or if the response is invalid.
-        """
-        await self._create_dataset(dataset, "active")
-
-    async def create_pending_dataset(self, dataset: OperationalDataSet) -> None:
+    async def create_pending_dataset(self, dataset: PendingDataSet) -> None:
         """Create pending operational dataset.
 
-        The passed in OperationalDataSet does not need to be fully populated, any fields
+        The passed in PendingDataSet does not need to be fully populated, any fields
         not set will be automatically set by the open thread border router.
         Raises if the http status is 400 or higher or if the response is invalid.
         """
-        await self._create_dataset(dataset, "pending")
+        response = await self._session.put(
+            f"{self._url}/node/dataset/pending",
+            json=dataset.as_json(),
+            timeout=aiohttp.ClientTimeout(total=self._timeout),
+        )
+
+        if response.status == HTTPStatus.CONFLICT:
+            raise ThreadNetworkActiveError
+        if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK):
+            raise OTBRError(f"unexpected http status {response.status}")
 
     async def set_active_dataset_tlvs(self, dataset: bytes) -> None:
         """Set current active operational dataset.
@@ -143,7 +140,7 @@ async def set_active_dataset_tlvs(self, dataset: bytes) -> None:
 
         if response.status == HTTPStatus.CONFLICT:
             raise ThreadNetworkActiveError
-        if response.status != HTTPStatus.ACCEPTED:
+        if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK):
             raise OTBRError(f"unexpected http status {response.status}")
 
     async def set_channel(
@@ -164,9 +161,9 @@ async def set_channel(
         else:
             dataset.active_timestamp = Timestamp(False, 1, 0)
         dataset.channel = channel
-        dataset.delay = delay
+        pending_dataset = PendingDataSet(active_dataset=dataset, delay=delay)
 
-        await self.create_pending_dataset(dataset)
+        await self.create_pending_dataset(pending_dataset)
 
     async def get_extended_address(self) -> bytes:
         """Get extended address (EUI-64).

python_otbr_api/models.py

@@ -119,21 +119,19 @@ def from_json(cls, json_data: Any) -> SecurityPolicy:
 
 
 @dataclass
-class OperationalDataSet:  # pylint: disable=too-many-instance-attributes
+class ActiveDataSet:  # pylint: disable=too-many-instance-attributes
     """Operational dataset."""
 
     SCHEMA = vol.Schema(
         {
             vol.Optional("ActiveTimestamp"): dict,
             vol.Optional("ChannelMask"): int,
             vol.Optional("Channel"): int,
-            vol.Optional("Delay"): int,
             vol.Optional("ExtPanId"): str,
             vol.Optional("MeshLocalPrefix"): str,
             vol.Optional("NetworkKey"): str,
             vol.Optional("NetworkName"): str,
             vol.Optional("PanId"): int,
-            vol.Optional("PendingTimestamp"): dict,
             vol.Optional("PSKc"): str,
             vol.Optional("SecurityPolicy"): dict,
         }
@@ -142,13 +140,11 @@ class OperationalDataSet:  # pylint: disable=too-many-instance-attributes
     active_timestamp: Timestamp | None = None
     channel_mask: int | None = None
     channel: int | None = None
-    delay: int | None = None
     extended_pan_id: str | None = None
     mesh_local_prefix: str | None = None
     network_key: str | None = None
     network_name: str | None = None
     pan_id: int | None = None
-    pending_timestamp: Timestamp | None = None
     psk_c: str | None = None
     security_policy: SecurityPolicy | None = None
 
@@ -161,8 +157,6 @@ def as_json(self) -> dict:
             result["ChannelMask"] = self.channel_mask
         if self.channel is not None:
             result["Channel"] = self.channel
-        if self.delay is not None:
-            result["Delay"] = self.delay
         if self.extended_pan_id is not None:
             result["ExtPanId"] = self.extended_pan_id
         if self.mesh_local_prefix is not None:
@@ -173,39 +167,77 @@ def as_json(self) -> dict:
             result["NetworkName"] = self.network_name
         if self.pan_id is not None:
             result["PanId"] = self.pan_id
-        if self.pending_timestamp is not None:
-            result["PendingTimestamp"] = self.pending_timestamp.as_json()
         if self.psk_c is not None:
             result["PSKc"] = self.psk_c
         if self.security_policy is not None:
             result["SecurityPolicy"] = self.security_policy.as_json()
         return result
 
     @classmethod
-    def from_json(cls, json_data: Any) -> OperationalDataSet:
+    def from_json(cls, json_data: Any) -> ActiveDataSet:
         """Deserialize from JSON."""
         cls.SCHEMA(json_data)
         active_timestamp = None
-        pending_timestamp = None
         security_policy = None
         if "ActiveTimestamp" in json_data:
             active_timestamp = Timestamp.from_json(json_data["ActiveTimestamp"])
-        if "PendingTimestamp" in json_data:
-            pending_timestamp = Timestamp.from_json(json_data["PendingTimestamp"])
         if "SecurityPolicy" in json_data:
             security_policy = SecurityPolicy.from_json(json_data["SecurityPolicy"])
 
-        return OperationalDataSet(
+        return ActiveDataSet(
             active_timestamp,
             json_data.get("ChannelMask"),
             json_data.get("Channel"),
-            json_data.get("Delay"),
             json_data.get("ExtPanId"),
             json_data.get("MeshLocalPrefix"),
             json_data.get("NetworkKey"),
             json_data.get("NetworkName"),
             json_data.get("PanId"),
-            pending_timestamp,
             json_data.get("PSKc"),
             security_policy,
         )
+
+
+@dataclass
+class PendingDataSet:  # pylint: disable=too-many-instance-attributes
+    """Operational dataset."""
+
+    SCHEMA = vol.Schema(
+        {
+            vol.Optional("ActiveDataset"): dict,
+            vol.Optional("Delay"): int,
+            vol.Optional("PendingTimestamp"): dict,
+        }
+    )
+
+    active_dataset: ActiveDataSet | None = None
+    delay: int | None = None
+    pending_timestamp: Timestamp | None = None
+
+    def as_json(self) -> dict:
+        """Serialize to JSON."""
+        result: dict[str, Any] = {}
+        if self.active_dataset is not None:
+            result["ActiveDataset"] = self.active_dataset.as_json()
+        if self.delay is not None:
+            result["Delay"] = self.delay
+        if self.pending_timestamp is not None:
+            result["PendingTimestamp"] = self.pending_timestamp.as_json()
+        return result
+
+    @classmethod
+    def from_json(cls, json_data: Any) -> PendingDataSet:
+        """Deserialize from JSON."""
+        cls.SCHEMA(json_data)
+        active_dataset = None
+        pending_timestamp = None
+        if "ActiveDataset" in json_data:
+            active_dataset = ActiveDataSet.from_json(json_data["ActiveDataset"])
+        if "PendingTimestamp" in json_data:
+            pending_timestamp = Timestamp.from_json(json_data["PendingTimestamp"])
+
+        return PendingDataSet(
+            active_dataset,
+            json_data.get("Delay"),
+            pending_timestamp,
+        )