From ea41b206b77d66b137d73bd156b06bcf9b7d7a9d Mon Sep 17 00:00:00 2001 From: Christian Glombek Date: Wed, 5 Aug 2026 21:30:43 +0200 Subject: [PATCH 1/2] Add pending-dataset TLV write support Add set_pending_dataset_tlvs(), the pending-endpoint twin of set_active_dataset_tlvs(), so a controller can hand a border router a new operational dataset as a raw TLV pending set. The TLV parser gains the write-side counterparts callers need to build such a dataset: Timestamp.from_values(), with range checks for the 48-bit seconds and 15-bit ticks fields, and DelayTimer.from_milliseconds(). PENDINGTIMESTAMP and DELAYTIMER now decode to typed items, matching ACTIVETIMESTAMP. Also raise TLVError instead of leaking struct.error on a malformed timestamp or delay timer TLV, and log unknown TLVs by type and length only -- the value may hold network credentials. Like the channel change, the write is refused while a pending dataset is already in place. A not-newer one would be silently ignored by the mesh, and superseding is never safe to do implicitly: the replacement races the delay timer on every device that already holds the old dataset, so a late replacement can split the mesh, and it would also silently undo whatever the in-flight dataset was doing. A caller that finds the write refused should surface that, not retry. The check runs locally first, and again on the border router: unless allow_replace is set the request carries "If-None-Match: *", which a border router with openthread/ot-br-posix#3552 evaluates atomically with the write, closing the race the local check leaves open. Older border routers ignore the header and keep relying on the local check. Both refusals raise PendingDatasetConflictError rather than a plain OTBRError, so a caller can tell them apart from a transport or protocol failure without matching on the message: nothing was written, and the state that made the write wrong is there to be read back. Assisted-By: Claude Fable 5 --- pyproject.toml | 2 +- python_otbr_api/__init__.py | 46 ++++++++++++++++ python_otbr_api/tlv_parser.py | 62 ++++++++++++++++++++-- tests/test_init_legacy.py | 87 +++++++++++++++++++++++++++++++ tests/test_tlv_parser.py | 98 ++++++++++++++++++++++++++++++++++- 5 files changed, 288 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1c494e9..04329aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-otbr-api" -version = "2.10.0" +version = "2.11.0" license = {text = "MIT"} description = "API to interact with an OTBR via its REST API" readme = "README.md" diff --git a/python_otbr_api/__init__.py b/python_otbr_api/__init__.py index 045e70f..74f09d5 100644 --- a/python_otbr_api/__init__.py +++ b/python_otbr_api/__init__.py @@ -113,6 +113,15 @@ class EphemeralKeyConflictError(OTBRError): key is already active.""" +class PendingDatasetConflictError(OTBRError): + """Raised when a pending dataset write is refused because one is in place. + + Its own type rather than a plain OTBRError: the caller's next step is + different from the one a transport or protocol failure calls for, since + nothing was written and the state that made the write wrong is readable. + """ + + def _rewrite_keys(data: Any, mapping: dict[str, str]) -> Any: """Recursively rename dict keys according to mapping; pass through others.""" if not isinstance(data, dict): @@ -372,6 +381,43 @@ async def set_active_dataset_tlvs(self, dataset: bytes) -> None: if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK): raise OTBRError(f"unexpected http status {response.status}") + async def set_pending_dataset_tlvs(self, dataset: bytes) -> None: + """Set the pending operational dataset, when none is in place yet. + + A write while a pending dataset is in flight is refused outright. + Superseding one is never safe to do implicitly: the replacement races + the delay timer on every device that already holds the old dataset, so + a late replacement can split the mesh, and it would also silently undo + whatever the in-flight dataset was doing, such as a channel change. + A caller that finds the write refused should surface that to whoever + asked for it, not retry. + + The check runs locally first, and again on the border router for one + that honors If-None-Match on this endpoint + (https://github.com/openthread/ot-br-posix/pull/3552), where it is + atomic with the write; older border routers ignore the header. + + Raises PendingDatasetConflictError when either check refuses the + write, and OTBRError if the http status is 400 or higher for any + other reason or the response is invalid. + """ + if await self.get_pending_dataset_tlvs() is not None: + raise PendingDatasetConflictError("a pending dataset is already in place") + await self._maybe_detect_key_format() + response = await self._session.put( + f"{self._url}/node/dataset/pending", + data=dataset.hex(), + headers={"Content-Type": "text/plain", "If-None-Match": "*"}, + timeout=aiohttp.ClientTimeout(total=10), + ) + + if response.status == HTTPStatus.CONFLICT: + raise ThreadNetworkActiveError + if response.status == HTTPStatus.PRECONDITION_FAILED: + raise PendingDatasetConflictError("a pending dataset is already in place") + if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK): + raise OTBRError(f"unexpected http status {response.status}") + async def set_channel( self, channel: int, delay: int = PENDING_DATASET_DELAY_TIMER ) -> None: diff --git a/python_otbr_api/tlv_parser.py b/python_otbr_api/tlv_parser.py index a213679..9a8fcff 100644 --- a/python_otbr_api/tlv_parser.py +++ b/python_otbr_api/tlv_parser.py @@ -116,11 +116,62 @@ def __post_init__(self) -> None: """Decode the timestamp.""" # The timestamps are packed in 8 bytes: # [seconds 48 bits][ticks 15 bits][authoritative flag 1 bit] - unpacked: int = struct.unpack("!Q", self.data)[0] + try: + unpacked: int = struct.unpack("!Q", self.data)[0] + except struct.error as err: + raise TLVError(f"invalid timestamp '{self.data.hex()}'") from err self.authoritative = bool(unpacked & 1) self.seconds = unpacked >> 16 self.ticks = (unpacked >> 1) & 0x7FFF + @classmethod + def from_values( + cls, + tag: MeshcopTLVType, + seconds: int, + ticks: int = 0, + authoritative: bool = False, + ) -> Timestamp: + """Construct a timestamp from its field values. + + Raises TLVError if seconds or ticks don't fit the wire format. + """ + if not 0 <= seconds < 2**48: + raise TLVError(f"timestamp seconds out of range: {seconds}") + if not 0 <= ticks < 2**15: + raise TLVError(f"timestamp ticks out of range: {ticks}") + packed = (seconds << 16) | (ticks << 1) | int(authoritative) + return cls(tag, struct.pack("!Q", packed)) + + +@dataclass +class DelayTimer(MeshcopTLVItem): + """Delay timer.""" + + delay: int = field(init=False) + + def __post_init__(self) -> None: + """Decode the delay in milliseconds. + + Raises TLVError if the data is not the four bytes the wire format + defines, so a malformed delay timer is rejected instead of being + accepted at whatever width it arrived with. + """ + try: + self.delay = struct.unpack("!L", self.data)[0] + except struct.error as err: + raise TLVError(f"invalid delay timer '{self.data.hex()}'") from err + + @classmethod + def from_milliseconds(cls, delay: int) -> DelayTimer: + """Construct a delay timer from a delay in milliseconds. + + Raises TLVError if the delay doesn't fit the wire format. + """ + if not 0 <= delay < 2**32: + raise TLVError(f"delay timer out of range: {delay}") + return cls(MeshcopTLVType.DELAYTIMER, struct.pack("!L", delay)) + def _encode_item(item: MeshcopTLVItem) -> bytes: """Encode a dataset item to TLV format.""" @@ -143,10 +194,12 @@ def encode_tlv(items: dict[MeshcopTLVType | int, MeshcopTLVItem]) -> str: def _parse_item(tag: MeshcopTLVType | int, data: bytes) -> MeshcopTLVItem: """Parse a TLV encoded dataset item.""" - if tag == MeshcopTLVType.ACTIVETIMESTAMP: + if tag in (MeshcopTLVType.ACTIVETIMESTAMP, MeshcopTLVType.PENDINGTIMESTAMP): return Timestamp(tag, data) if tag == MeshcopTLVType.CHANNEL: return Channel(tag, data) + if tag == MeshcopTLVType.DELAYTIMER: + return DelayTimer(tag, data) if tag == MeshcopTLVType.NETWORKNAME: return NetworkName(tag, data) @@ -190,9 +243,10 @@ def parse_tlv(data: str) -> dict[MeshcopTLVType | int, MeshcopTLVItem]: val = data_bytes[pos : pos + _len] pos += _len - # Once we have the value, we can log a warning about the unknown TLV + # Log unknown TLVs by type and length only: the value may hold + # network credentials, which don't belong in the log. if not isinstance(tag, MeshcopTLVType): - _LOGGER.warning("unknown TLV type %d=%r", raw_tag, val) + _LOGGER.warning("unknown TLV type %d (%d bytes)", raw_tag, _len) if tag in result: raise TLVError(f"duplicated tag {tag!r}") diff --git a/tests/test_init_legacy.py b/tests/test_init_legacy.py index 8e0885b..75429b5 100644 --- a/tests/test_init_legacy.py +++ b/tests/test_init_legacy.py @@ -790,3 +790,90 @@ async def test_get_coprocessor_version_invalid(aioclient_mock: AiohttpClientMock with pytest.raises(python_otbr_api.OTBRError): await otbr.get_coprocessor_version() + + +async def test_set_pending_dataset_tlvs(aioclient_mock: AiohttpClientMocker) -> None: + """Test set_pending_dataset_tlvs.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + dataset = bytes.fromhex( + "0E080000000000010000000300000F35060004001FFFE0020811111111222222220708FDAD" + "70BFE5AA15DD051000112233445566778899AABBCCDDEEFF030E4F70656E54687265616444" + "656D6F010212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8" + ) + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT) + aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CREATED) + + await otbr.set_pending_dataset_tlvs(dataset) + 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] == dataset.hex() + assert aioclient_mock.mock_calls[-1][3]["If-None-Match"] == "*" + + +async def test_set_pending_dataset_tlvs_thread_active( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test set_pending_dataset_tlvs with error.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT) + aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CONFLICT) + + with pytest.raises(python_otbr_api.ThreadNetworkActiveError): + await otbr.set_pending_dataset_tlvs(b"") + + +async def test_set_pending_dataset_tlvs_refused_while_pending( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test a write is refused while a pending dataset is in place.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + in_flight = "0E080000000000010000340400006699000300000C" + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", text=in_flight) + + with pytest.raises(python_otbr_api.PendingDatasetConflictError): + await otbr.set_pending_dataset_tlvs(bytes.fromhex(in_flight)) + assert aioclient_mock.call_count == 1 + + +async def test_set_pending_dataset_tlvs_refused_by_router( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test the border router refusing the precondition is surfaced.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT) + aioclient_mock.put( + f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.PRECONDITION_FAILED + ) + + # The dedicated type, so a caller can tell "nothing was written because a + # dataset is in place" from a transport or protocol failure. + with pytest.raises(python_otbr_api.PendingDatasetConflictError): + await otbr.set_pending_dataset_tlvs(b"") + + +async def test_set_pending_dataset_tlvs_202( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test set_pending_dataset_tlvs with error.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT) + aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.ACCEPTED) + + with pytest.raises(python_otbr_api.OTBRError): + await otbr.set_pending_dataset_tlvs(b"") diff --git a/tests/test_tlv_parser.py b/tests/test_tlv_parser.py index 9aeb3e9..3c6e75f 100644 --- a/tests/test_tlv_parser.py +++ b/tests/test_tlv_parser.py @@ -5,6 +5,7 @@ from python_otbr_api.tlv_parser import ( Timestamp, Channel, + DelayTimer, MeshcopTLVItem, MeshcopTLVType, NetworkName, @@ -31,10 +32,10 @@ MeshcopTLVType.IPV6_ADDRESS_TLV, bytes.fromhex("20010db8000000000000000000000001"), ), - MeshcopTLVType.PENDINGTIMESTAMP: MeshcopTLVItem( + MeshcopTLVType.PENDINGTIMESTAMP: Timestamp( MeshcopTLVType.PENDINGTIMESTAMP, bytes.fromhex("0000000000010000") ), - MeshcopTLVType.DELAYTIMER: MeshcopTLVItem( + MeshcopTLVType.DELAYTIMER: DelayTimer( MeshcopTLVType.DELAYTIMER, bytes.fromhex("00001388") ), MeshcopTLVType.COUNT: MeshcopTLVItem(MeshcopTLVType.COUNT, bytes.fromhex("03")), @@ -252,3 +253,96 @@ def test_timestamp_parsing_full_integrity() -> None: # 3. Check Authoritative: Ensures the lowest bit is read correctly assert timestamp.authoritative is True + + +def test_timestamp_from_values() -> None: + """Test constructing a timestamp from field values.""" + timestamp = Timestamp.from_values( + MeshcopTLVType.ACTIVETIMESTAMP, seconds=400, ticks=32767, authoritative=True + ) + assert timestamp.data == bytes.fromhex("000000000190FFFF") + assert timestamp.seconds == 400 + assert timestamp.ticks == 32767 + assert timestamp.authoritative is True + + pending = Timestamp.from_values(MeshcopTLVType.PENDINGTIMESTAMP, seconds=1) + assert pending.tag == MeshcopTLVType.PENDINGTIMESTAMP + assert pending.data == bytes.fromhex("0000000000010000") + assert pending.ticks == 0 + assert pending.authoritative is False + + # The largest encodable timestamp survives a roundtrip + ceiling = Timestamp.from_values( + MeshcopTLVType.ACTIVETIMESTAMP, seconds=2**48 - 1, ticks=2**15 - 1 + ) + assert ceiling.seconds == 2**48 - 1 + assert ceiling.ticks == 2**15 - 1 + + +@pytest.mark.parametrize( + ("seconds", "ticks", "msg"), + ( + (2**48, 0, "timestamp seconds out of range"), + (-1, 0, "timestamp seconds out of range"), + (0, 2**15, "timestamp ticks out of range"), + (0, -1, "timestamp ticks out of range"), + ), +) +def test_timestamp_from_values_out_of_range(seconds, ticks, msg) -> None: + """Test constructing a timestamp from values which don't fit the wire format.""" + with pytest.raises(TLVError, match=msg): + Timestamp.from_values( + MeshcopTLVType.ACTIVETIMESTAMP, seconds=seconds, ticks=ticks + ) + + +def test_timestamp_invalid_data() -> None: + """Test a malformed timestamp raises TLVError, not struct.error.""" + with pytest.raises(TLVError, match="invalid timestamp '00'"): + Timestamp(MeshcopTLVType.ACTIVETIMESTAMP, bytes.fromhex("00")) + # Also via the parser, as it would arrive from user supplied TLVs + with pytest.raises(TLVError, match="invalid timestamp '00000000000100'"): + parse_tlv("0E0700000000000100") + + +def test_pending_timestamp_and_delay_timer_parsed() -> None: + """Test PENDINGTIMESTAMP and DELAYTIMER decode to typed items.""" + dataset = parse_tlv("33080000000000010000340400006699") + pending = dataset[MeshcopTLVType.PENDINGTIMESTAMP] + assert isinstance(pending, Timestamp) + assert pending.seconds == 1 + delay = dataset[MeshcopTLVType.DELAYTIMER] + assert isinstance(delay, DelayTimer) + assert delay.delay == 0x6699 + + +def test_delay_timer_from_milliseconds() -> None: + """Test constructing a delay timer from a delay in milliseconds.""" + delay = DelayTimer.from_milliseconds(5 * 60 * 1000) + assert delay.tag == MeshcopTLVType.DELAYTIMER + assert delay.data == bytes.fromhex("000493E0") + assert delay.delay == 300000 + assert encode_tlv({MeshcopTLVType.DELAYTIMER: delay}) == "3404000493e0" + + +@pytest.mark.parametrize("delay", (2**32, -1)) +def test_delay_timer_out_of_range(delay) -> None: + """Test constructing a delay timer which doesn't fit the wire format.""" + with pytest.raises(TLVError, match="delay timer out of range"): + DelayTimer.from_milliseconds(delay) + + +def test_delay_timer_invalid_data() -> None: + """Test a malformed delay timer raises TLVError, not struct.error.""" + with pytest.raises(TLVError, match="invalid delay timer '00'"): + DelayTimer(MeshcopTLVType.DELAYTIMER, bytes.fromhex("00")) + # Also via the parser, as it would arrive from user supplied TLVs + with pytest.raises(TLVError, match="invalid delay timer '000000'"): + parse_tlv("3403000000") + + +def test_unknown_tlv_value_not_logged(caplog: pytest.LogCaptureFixture) -> None: + """Test the unknown-TLV warning does not log the value.""" + parse_tlv("BD03ABCDEF") + assert "unknown TLV type 189 (3 bytes)" in caplog.text + assert "abcdef" not in caplog.text.lower() From aebad9bc975d28cb512df57ab8c9fad667bce927 Mon Sep 17 00:00:00 2001 From: Christian Glombek Date: Tue, 25 Aug 2026 10:08:00 +0200 Subject: [PATCH 2/2] Support conditional pending-dataset replaces with entity tags set_pending_dataset_tlvs() refuses whenever a pending dataset is in place, and deliberately offers no blanket override: superseding races the delay timer on every device holding the old dataset and silently undoes whatever that dataset was doing. What a caller may reasonably want is narrower: "replace exactly the dataset I read", as an explicit, informed action. Add get_pending_dataset_tlvs_with_etag(), returning the dataset together with the entity tag a border router hands out for it, and an if_match keyword on set_pending_dataset_tlvs() that passes the tag back: the router then only replaces the dataset the tag stands for, checked atomically with the write (openthread/ot-br-posix#3553). An older border router hands out no tag and would ignore the header, making the replace there unconditional, which is why callers must only expose this as an explicit choice, never as a default. A conditional replace that the router refuses raises the same PendingDatasetConflictError the guarded write already raises, with a message naming the other reason: the dataset changed since it was read. Assisted-By: Claude Fable 5 --- python_otbr_api/__init__.py | 62 +++++++++++++++++++++++---- tests/test_init_legacy.py | 84 +++++++++++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 8 deletions(-) diff --git a/python_otbr_api/__init__.py b/python_otbr_api/__init__.py index 74f09d5..fe873b7 100644 --- a/python_otbr_api/__init__.py +++ b/python_otbr_api/__init__.py @@ -114,7 +114,11 @@ class EphemeralKeyConflictError(OTBRError): class PendingDatasetConflictError(OTBRError): - """Raised when a pending dataset write is refused because one is in place. + """Raised when a pending dataset write is refused. + + Either a pending dataset is already in place for a guarded write, or the + one whose entity tag the write carried has been replaced since it was + read. Its own type rather than a plain OTBRError: the caller's next step is different from the one a transport or protocol failure calls for, since @@ -281,6 +285,21 @@ async def get_pending_dataset_tlvs(self) -> bytes | None: Returns None if there is no pending operational dataset. Raises if the http status is 400 or higher or if the response is invalid. """ + result = await self.get_pending_dataset_tlvs_with_etag() + return None if result is None else result[0] + + async def get_pending_dataset_tlvs_with_etag( + self, + ) -> tuple[bytes, str | None] | None: + """Get the pending dataset TLVs together with its entity tag, or None. + + Returns None if there is no pending operational dataset. The entity + tag is None on a border router that does not hand one out; when it + does (https://github.com/openthread/ot-br-posix/pull/3553), the tag + can be passed to set_pending_dataset_tlvs() as if_match to only + replace the dataset that was read here. + Raises if the http status is 400 or higher or if the response is invalid. + """ await self._maybe_detect_key_format() response = await self._session.get( f"{self._url}/node/dataset/pending", @@ -295,7 +314,10 @@ async def get_pending_dataset_tlvs(self) -> bytes | None: raise OTBRError(f"unexpected http status {response.status}") try: - return bytes.fromhex(await response.text("ASCII")) + return ( + bytes.fromhex(await response.text("ASCII")), + response.headers.get("ETag"), + ) except ValueError as exc: raise OTBRError("unexpected API response") from exc @@ -381,7 +403,9 @@ async def set_active_dataset_tlvs(self, dataset: bytes) -> None: if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK): raise OTBRError(f"unexpected http status {response.status}") - async def set_pending_dataset_tlvs(self, dataset: bytes) -> None: + async def set_pending_dataset_tlvs( + self, dataset: bytes, *, if_match: str | None = None + ) -> None: """Set the pending operational dataset, when none is in place yet. A write while a pending dataset is in flight is refused outright. @@ -397,23 +421,45 @@ async def set_pending_dataset_tlvs(self, dataset: bytes) -> None: (https://github.com/openthread/ot-br-posix/pull/3552), where it is atomic with the write; older border routers ignore the header. - Raises PendingDatasetConflictError when either check refuses the - write, and OTBRError if the http status is 400 or higher for any + The one deliberate way to replace an in-flight dataset is by its + entity tag: a caller that read it via + get_pending_dataset_tlvs_with_etag() passes the tag as if_match, and + the write then only replaces exactly the dataset that was read, + refused if it changed in between. That check happens on the border + router, atomically with the write, so no local check runs on this + path; an older border router that hands out no entity tags ignores + the header, making the replace there unconditional, which is why a + caller must only offer this as an explicit, informed choice. + + Raises PendingDatasetConflictError when any of these checks refuses + the write, and OTBRError if the http status is 400 or higher for any other reason or the response is invalid. """ - if await self.get_pending_dataset_tlvs() is not None: - raise PendingDatasetConflictError("a pending dataset is already in place") + if if_match is None: + if await self.get_pending_dataset_tlvs() is not None: + raise PendingDatasetConflictError( + "a pending dataset is already in place" + ) await self._maybe_detect_key_format() + headers = {"Content-Type": "text/plain"} + if if_match is not None: + headers["If-Match"] = if_match + else: + headers["If-None-Match"] = "*" response = await self._session.put( f"{self._url}/node/dataset/pending", data=dataset.hex(), - headers={"Content-Type": "text/plain", "If-None-Match": "*"}, + headers=headers, timeout=aiohttp.ClientTimeout(total=10), ) if response.status == HTTPStatus.CONFLICT: raise ThreadNetworkActiveError if response.status == HTTPStatus.PRECONDITION_FAILED: + if if_match is not None: + raise PendingDatasetConflictError( + "the pending dataset changed since it was read" + ) raise PendingDatasetConflictError("a pending dataset is already in place") if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK): raise OTBRError(f"unexpected http status {response.status}") diff --git a/tests/test_init_legacy.py b/tests/test_init_legacy.py index 75429b5..6e9a607 100644 --- a/tests/test_init_legacy.py +++ b/tests/test_init_legacy.py @@ -864,6 +864,90 @@ async def test_set_pending_dataset_tlvs_refused_by_router( await otbr.set_pending_dataset_tlvs(b"") +async def test_get_pending_dataset_tlvs_with_etag( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test get_pending_dataset_tlvs_with_etag.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + mock_response = "0E080000000000010000340400006699000300000C" + aioclient_mock.get( + f"{BASE_URL}/node/dataset/pending", + text=mock_response, + headers={"ETag": '"8311BDCD94E7107C"'}, + ) + + assert await otbr.get_pending_dataset_tlvs_with_etag() == ( + bytes.fromhex(mock_response), + '"8311BDCD94E7107C"', + ) + + +async def test_get_pending_dataset_tlvs_with_etag_no_tag( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test the tag is None on a border router that hands out none.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + mock_response = "0E080000000000010000340400006699000300000C" + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", text=mock_response) + + assert await otbr.get_pending_dataset_tlvs_with_etag() == ( + bytes.fromhex(mock_response), + None, + ) + + +async def test_get_pending_dataset_tlvs_with_etag_empty( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test get_pending_dataset_tlvs_with_etag without a pending dataset.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT) + assert await otbr.get_pending_dataset_tlvs_with_etag() is None + + +async def test_set_pending_dataset_tlvs_if_match( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test if_match makes the replace conditional on the tag, with no local check.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + dataset = bytes.fromhex("0E080000000000010000340400006699000300000C") + aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.OK) + + await otbr.set_pending_dataset_tlvs(dataset, if_match='"8311BDCD94E7107C"') + assert aioclient_mock.call_count == 1 + assert aioclient_mock.mock_calls[-1][0] == "PUT" + assert aioclient_mock.mock_calls[-1][3]["If-Match"] == '"8311BDCD94E7107C"' + assert "If-None-Match" not in aioclient_mock.mock_calls[-1][3] + + +async def test_set_pending_dataset_tlvs_if_match_changed( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Test a conditional replace of a dataset that changed is surfaced.""" + otbr = python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE + ) + + aioclient_mock.put( + f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.PRECONDITION_FAILED + ) + + with pytest.raises(python_otbr_api.PendingDatasetConflictError): + await otbr.set_pending_dataset_tlvs(b"", if_match='"8311BDCD94E7107C"') + + async def test_set_pending_dataset_tlvs_202( aioclient_mock: AiohttpClientMocker, ) -> None: