Add set_channel

python_otbr_api/__init__.py

@@ -1,10 +1,17 @@
 """API to interact with the Open Thread Border Router REST API."""
 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
+from .models import OperationalDataSet, Timestamp
+
+# 5 minutes as recommended by
+# https://github.com/openthread/openthread/discussions/8567#discussioncomment-4468920
+PENDING_DATASET_DELAY_TIMER = 5 * 60 * 1000
 
 
 class OTBRError(Exception):
@@ -38,6 +45,28 @@ 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:
+        """Get current active operational dataset.
+
+        Returns None if there is no active operational dataset.
+        Raises if the http status is 400 or higher or if the response is invalid.
+        """
+        response = await self._session.get(
+            f"{self._url}/node/dataset/active",
+            timeout=aiohttp.ClientTimeout(total=self._timeout),
+        )
+
+        if response.status == HTTPStatus.NO_CONTENT:
+            return None
+
+        if response.status != HTTPStatus.OK:
+            raise OTBRError(f"unexpected http status {response.status}")
+
+        try:
+            return OperationalDataSet.from_json(await response.json())
+        except (json.JSONDecodeError, vol.Error) as exc:
+            raise OTBRError("unexpected API response") from exc
+
     async def get_active_dataset_tlvs(self) -> bytes | None:
         """Get current active operational dataset in TLVS format, or None.
 
@@ -61,16 +90,17 @@ async def get_active_dataset_tlvs(self) -> bytes | None:
         except ValueError as exc:
             raise OTBRError("unexpected API response") from exc
 
-    async def create_active_dataset(self, dataset: OperationalDataSet) -> None:
-        """Create active operational dataset.
+    async def _create_dataset(
+        self, dataset: OperationalDataSet, dataset_type: Literal["active", "pending"]
+    ) -> None:
+        """Create active or pending 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.
         """
-
         response = await self._session.post(
-            f"{self._url}/node/dataset/active",
+            f"{self._url}/node/dataset/{dataset_type}",
             json=dataset.as_json(),
             timeout=aiohttp.ClientTimeout(total=self._timeout),
         )
@@ -80,6 +110,24 @@ async def create_active_dataset(self, dataset: OperationalDataSet) -> None:
         if response.status != HTTPStatus.ACCEPTED:
             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:
+        """Create pending 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, "pending")
+
     async def set_active_dataset_tlvs(self, dataset: bytes) -> None:
         """Set current active operational dataset.
 
@@ -98,6 +146,26 @@ async def set_active_dataset_tlvs(self, dataset: bytes) -> None:
         if response.status != HTTPStatus.ACCEPTED:
             raise OTBRError(f"unexpected http status {response.status}")
 
+    async def set_channel(self, channel: int) -> None:
+        """Change the channel
+
+        The channel is changed by creating a new pending dataset based on the active
+        dataset.
+        """
+        if not 11 <= channel <= 26:
+            raise OTBRError(f"invalid channel {channel}")
+        if not (dataset := await self.get_active_dataset()):
+            raise OTBRError("router has no active dataset")
+
+        if dataset.active_timestamp and dataset.active_timestamp.seconds is not None:
+            dataset.active_timestamp.seconds += 1
+        else:
+            dataset.active_timestamp = Timestamp(False, 1, 0)
+        dataset.channel = channel
+        dataset.delay = PENDING_DATASET_DELAY_TIMER
+
+        await self.create_pending_dataset(dataset)
+
     async def get_extended_address(self) -> bytes:
         """Get extended address (EUI-64).
 

tests/test_init.py

@@ -1,13 +1,41 @@
 """Test the OTBR REST API client."""
 
 from http import HTTPStatus
+from typing import Any
 
 import pytest
 import python_otbr_api
 
 from tests.test_util.aiohttp import AiohttpClientMocker
 
 BASE_URL = "http://core-silabs-multiprotocol:8081"
+DATASET_JSON: dict[str, Any] = {
+    "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,
+    },
+}
 
 
 async def test_set_enabled(aioclient_mock: AiohttpClientMocker) -> None:
@@ -29,6 +57,56 @@ async def test_set_enabled(aioclient_mock: AiohttpClientMocker) -> None:
     assert aioclient_mock.mock_calls[-1][2] == "disable"
 
 
+async def test_get_active_dataset(aioclient_mock: AiohttpClientMocker):
+    """Test get_active_dataset."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", json=DATASET_JSON)
+
+    active_timestamp = python_otbr_api.models.Timestamp(
+        DATASET_JSON["ActiveTimestamp"]["Authoritative"],
+        DATASET_JSON["ActiveTimestamp"]["Seconds"],
+        DATASET_JSON["ActiveTimestamp"]["Ticks"],
+    )
+    security_policy = python_otbr_api.models.SecurityPolicy(
+        DATASET_JSON["SecurityPolicy"]["AutonomousEnrollment"],
+        DATASET_JSON["SecurityPolicy"]["CommercialCommissioning"],
+        DATASET_JSON["SecurityPolicy"]["ExternalCommissioning"],
+        DATASET_JSON["SecurityPolicy"]["NativeCommissioning"],
+        DATASET_JSON["SecurityPolicy"]["NetworkKeyProvisioning"],
+        DATASET_JSON["SecurityPolicy"]["NonCcmRouters"],
+        DATASET_JSON["SecurityPolicy"]["ObtainNetworkKey"],
+        DATASET_JSON["SecurityPolicy"]["RotationTime"],
+        DATASET_JSON["SecurityPolicy"]["Routers"],
+        DATASET_JSON["SecurityPolicy"]["TobleLink"],
+    )
+
+    active_dataset = await otbr.get_active_dataset()
+    assert active_dataset == python_otbr_api.OperationalDataSet(
+        active_timestamp,
+        DATASET_JSON["ChannelMask"],
+        DATASET_JSON["Channel"],
+        None,  # delay
+        DATASET_JSON["ExtPanId"],
+        DATASET_JSON["MeshLocalPrefix"],
+        DATASET_JSON["NetworkKey"],
+        DATASET_JSON["NetworkName"],
+        DATASET_JSON["PanId"],
+        None,  # pending_timestamp
+        DATASET_JSON["PSKc"],
+        security_policy,
+    )
+    assert active_dataset.as_json() == DATASET_JSON
+
+
+async def test_get_active_dataset_empty(aioclient_mock: AiohttpClientMocker):
+    """Test get_active_dataset."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", status=HTTPStatus.NO_CONTENT)
+    assert await otbr.get_active_dataset() is None
+
+
 async def test_get_active_dataset_tlvs(aioclient_mock: AiohttpClientMocker) -> None:
     """Test get_active_dataset_tlvs."""
     otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
@@ -52,38 +130,114 @@ async def test_get_active_dataset_tlvs_empty(aioclient_mock: AiohttpClientMocker
     assert await otbr.get_active_dataset_tlvs() is None
 
 
-async def test_create_active_dataset(aioclient_mock: AiohttpClientMocker):
-    """Test create_active_dataset."""
+@pytest.mark.parametrize("dataset_type", ["active", "pending"])
+async def test_create_active_pending_dataset(
+    aioclient_mock: AiohttpClientMocker, dataset_type: str
+):
+    """Test create_active_dataset + create_pending_dataset."""
     otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
 
-    aioclient_mock.post(f"{BASE_URL}/node/dataset/active", status=HTTPStatus.ACCEPTED)
+    aioclient_mock.post(
+        f"{BASE_URL}/node/dataset/{dataset_type}", status=HTTPStatus.ACCEPTED
+    )
 
-    await otbr.create_active_dataset(python_otbr_api.OperationalDataSet())
+    await getattr(otbr, f"create_{dataset_type}_dataset")(
+        python_otbr_api.OperationalDataSet()
+    )
     assert aioclient_mock.call_count == 1
     assert aioclient_mock.mock_calls[-1][0] == "POST"
-    assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/active"
+    assert aioclient_mock.mock_calls[-1][1].path == f"/node/dataset/{dataset_type}"
     assert aioclient_mock.mock_calls[-1][2] == {}
 
-    await otbr.create_active_dataset(
+    await getattr(otbr, f"create_{dataset_type}_dataset")(
         python_otbr_api.OperationalDataSet(network_name="OpenThread HA")
     )
     assert aioclient_mock.call_count == 2
     assert aioclient_mock.mock_calls[-1][0] == "POST"
-    assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/active"
+    assert aioclient_mock.mock_calls[-1][1].path == f"/node/dataset/{dataset_type}"
     assert aioclient_mock.mock_calls[-1][2] == {"NetworkName": "OpenThread HA"}
 
-    await otbr.create_active_dataset(
+    await getattr(otbr, f"create_{dataset_type}_dataset")(
         python_otbr_api.OperationalDataSet(network_name="OpenThread HA", channel=15)
     )
     assert aioclient_mock.call_count == 3
     assert aioclient_mock.mock_calls[-1][0] == "POST"
-    assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/active"
+    assert aioclient_mock.mock_calls[-1][1].path == f"/node/dataset/{dataset_type}"
     assert aioclient_mock.mock_calls[-1][2] == {
         "NetworkName": "OpenThread HA",
         "Channel": 15,
     }
 
 
+async def test_set_channel(aioclient_mock: AiohttpClientMocker) -> None:
+    """Test set_channel."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", json=DATASET_JSON)
+    aioclient_mock.post(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.ACCEPTED)
+    new_channel = 16
+    expected_active_timestamp = DATASET_JSON["ActiveTimestamp"] | {"Seconds": 2}
+    expected_pending_dataset = 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] == "POST"
+    assert aioclient_mock.mock_calls[1][1].path == "/node/dataset/pending"
+    assert aioclient_mock.mock_calls[1][2] == expected_pending_dataset
+
+
+async def test_set_channel_no_timestamp(aioclient_mock: AiohttpClientMocker) -> None:
+    """Test set_channel."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    dataset_json = dict(DATASET_JSON)
+    dataset_json.pop("ActiveTimestamp")
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", json=dataset_json)
+    aioclient_mock.post(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.ACCEPTED)
+    new_channel = 16
+    expected_active_timestamp = {"Authoritative": False, "Seconds": 1, "Ticks": 0}
+    expected_pending_dataset = 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] == "POST"
+    assert aioclient_mock.mock_calls[1][1].path == "/node/dataset/pending"
+    assert aioclient_mock.mock_calls[1][2] == expected_pending_dataset
+
+
+async def test_set_channel_invalid_channel(aioclient_mock: AiohttpClientMocker) -> None:
+    """Test set_channel."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.set_channel(123)
+
+
+async def test_set_channel_no_dataset(aioclient_mock: AiohttpClientMocker) -> None:
+    """Test set_channel."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", status=HTTPStatus.NO_CONTENT)
+
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.set_channel(16)
+
+
 async def test_get_extended_address(aioclient_mock: AiohttpClientMocker) -> None:
     """Test get_active_dataset_tlvs."""
     otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
@@ -105,6 +259,24 @@ async def test_set_enabled_201(aioclient_mock: AiohttpClientMocker) -> None:
         await otbr.set_enabled(True)
 
 
+async def test_get_active_dataset_201(aioclient_mock: AiohttpClientMocker):
+    """Test get_active_dataset with error."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", status=HTTPStatus.CREATED)
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.get_active_dataset()
+
+
+async def test_get_active_dataset_invalid(aioclient_mock: AiohttpClientMocker):
+    """Test get_active_dataset with error."""
+    otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/active", text="unexpected")
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.get_active_dataset()
+
+
 async def test_get_active_dataset_tlvs_201(aioclient_mock: AiohttpClientMocker):
     """Test get_active_dataset_tlvs with error."""
     otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
@@ -133,14 +305,19 @@ async def test_create_active_dataset_thread_active(aioclient_mock: AiohttpClient
         await otbr.create_active_dataset(python_otbr_api.OperationalDataSet())
 
 
-async def test_create_active_dataset_200(aioclient_mock: AiohttpClientMocker):
-    """Test create_active_dataset with error."""
+@pytest.mark.parametrize("dataset_type", ["active", "pending"])
+async def test_create_active_pending_dataset_200(
+    aioclient_mock: AiohttpClientMocker, dataset_type: str
+):
+    """Test create_active_dataset + create_pending_dataset with error."""
     otbr = python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
 
-    aioclient_mock.post(f"{BASE_URL}/node/dataset/active", status=HTTPStatus.OK)
+    aioclient_mock.post(f"{BASE_URL}/node/dataset/{dataset_type}", status=HTTPStatus.OK)
 
     with pytest.raises(python_otbr_api.OTBRError):
-        await otbr.create_active_dataset(python_otbr_api.OperationalDataSet())
+        await getattr(otbr, f"create_{dataset_type}_dataset")(
+            python_otbr_api.OperationalDataSet()
+        )
 
 
 async def test_set_active_dataset_tlvs_thread_active(