From 7c4d8f3d16d2540a1fc7a99971bd2614b8cccfbf Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 09:39:25 +0200 Subject: [PATCH 01/27] feat: add complete LAN DHCP write helpers --- src/nr2301/namespaces/lan.py | 393 +++++++++++++++++++++++++++++------ 1 file changed, 327 insertions(+), 66 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index bd1949c..6baca1c 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -65,6 +65,12 @@ class LANAddressResponse(TypedDict, total=False): router: LANAddress +class StaticReservation(TypedDict): + index: str + mac: str + ip: str + + class DNSSettings(TypedDict): """Verified DNS subset of the combined DHCP settings object.""" @@ -161,6 +167,271 @@ def static_reservations(self, *, timeout: float | None = None) -> dict[str, Any] timeout=timeout, ) + def set_dhcp_settings( + self, + settings: Mapping[str, Any], + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> DHCPSettings: + """Write the complete verified 12-field combined DHCP object. + + Callers should start from `lan.dhcp()`, modify only intended fields, + then pass the full object here. The helper never invents missing values. + """ + + if not isinstance(settings, Mapping): + raise TypeError("settings must be a mapping") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + payload: dict[str, str] = {} + missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] + if missing: + raise ProtocolError( + "combined DHCP write requires all fields: " + ", ".join(missing) + ) + for key in _REQUIRED_COMBINED_FIELDS: + value = settings[key] + if not isinstance(value, str): + raise TypeError(f"{key} must be a str") + payload[key] = value + + current = self.dhcp(timeout=recovery_timeout) + current_cmp = {key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS} + if current_cmp == payload: + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [{ + "path": "router", + "method": "router_set_dhcp_settings_comb", + "data": payload, + "timeout": write_timeout, + }], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: DHCPSettings | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.dhcp(timeout=recovery_timeout) + last_actual = actual + actual_cmp = { + key: actual.get(key) for key in _REQUIRED_COMBINED_FIELDS + } + if actual_cmp == payload: + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": payload, + "actual": ( + {key: last_actual.get(key) for key in _REQUIRED_COMBINED_FIELDS} + if last_actual is not None else None + ), + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + raise APIError( + "combined DHCP write could not be verified by exact read-back", + method_id="router/router_set_dhcp_settings_comb", + response=details, + ) + + def static_reservation_list( + self, + *, + timeout: float | None = None, + ) -> list[StaticReservation]: + """Return normalized static DHCP reservation items.""" + + response = self.static_reservations(timeout=timeout) + dhcp = response.get("dhcp") + if not isinstance(dhcp, Mapping): + raise ProtocolError( + "router/router_get_dhcp_static_ip did not return a dhcp object" + ) + raw = dhcp.get("data") + if not isinstance(raw, list): + raise ProtocolError( + "router/router_get_dhcp_static_ip did not return dhcp.data as a list" + ) + result: list[StaticReservation] = [] + for item in raw: + if not isinstance(item, Mapping): + raise ProtocolError("static DHCP reservation is not an object") + result.append(self._normalize_reservation(item)) + return result + + def set_static_reservations( + self, + reservations: list[Mapping[str, Any]], + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> list[StaticReservation]: + """Replace the complete 10-slot static DHCP reservation table.""" + + if not isinstance(reservations, list): + raise TypeError("reservations must be a list") + if len(reservations) > 10: + raise ValueError("the stock frontend supports at most 10 reservations") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + expected = [self._normalize_reservation(item) for item in reservations] + indices = [item["index"] for item in expected] + if len(indices) != len(set(indices)): + raise ValueError("reservation indices must be unique") + + current = self.static_reservation_list(timeout=recovery_timeout) + if self._reservation_cmp(current) == self._reservation_cmp(expected): + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [{ + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": expected}, + "timeout": write_timeout, + }], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: list[StaticReservation] | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.static_reservation_list(timeout=recovery_timeout) + last_actual = actual + if self._reservation_cmp(actual) == self._reservation_cmp(expected): + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected_count": len(expected), + "actual_count": len(last_actual) if last_actual is not None else None, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + raise APIError( + "static DHCP reservation write could not be verified by read-back", + method_id="router/router_set_dhcp_static_ip", + response=details, + ) + + def set_address_legacy( + self, + lan_ip: str, + lan_netmask: str, + *, + write_timeout: float = 30.0, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> LANAddressResponse: + """Use the deprecated dedicated LAN-address setter and require read-back.""" + + _validate_ip(lan_ip, version=4, field="lan_ip") + _validate_ip(lan_netmask, version=4, field="lan_netmask") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + current = self.address(timeout=recovery_timeout) + router = current.get("router") + if isinstance(router, Mapping) and router.get("lan_ip") == lan_ip and router.get("lan_netmask") == lan_netmask: + return current + + write_error: NR2301Error | None = None + try: + self._client.call( + "router", + "router_set_lan_ip", + data={"lan_ip": lan_ip, "lan_netmask": lan_netmask}, + timeout=write_timeout, + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: LANAddressResponse | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.address(timeout=recovery_timeout) + last_actual = actual + router = actual.get("router") + if isinstance(router, Mapping) and router.get("lan_ip") == lan_ip and router.get("lan_netmask") == lan_netmask: + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": {"lan_ip": lan_ip, "lan_netmask": lan_netmask}, + "actual": last_actual, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + raise APIError( + "legacy LAN address write could not be verified by read-back", + method_id="router/router_set_lan_ip", + response=details, + ) + def dns(self, *, timeout: float | None = None) -> DNSSettings: """Return the five DNS fields from the combined DHCP object.""" @@ -249,6 +520,62 @@ def _set_dns_fields( recovery_delay: float, recovery_timeout: float, ) -> DNSSettings: + before = self.dhcp() + payload: dict[str, Any] = dict(before) + payload.update(expected) + verified = self.set_dhcp_settings( + payload, + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + return cast( + DNSSettings, + {key: verified[key] for key in ("dnsmode", "dns1", "dns2", "ipv6dns1", "ipv6dns2")}, + ) + + @staticmethod + def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: + raw_index = item.get("index") + if isinstance(raw_index, bool): + raise TypeError("reservation index must be a string or integer") + try: + index = int(raw_index) + except (TypeError, ValueError) as exc: + raise ValueError("reservation index must be numeric") from exc + if index < 0 or index > 9: + raise ValueError("reservation index must be between 0 and 9") + + mac = item.get("mac") + if not isinstance(mac, str): + raise TypeError("reservation mac must be a str") + parts = mac.split(":") + if len(parts) != 6 or any(len(part) != 2 or any(ch not in "0123456789abcdefABCDEF" for ch in part) for part in parts): + raise ValueError("reservation mac must be a colon-separated MAC address") + + ip = item.get("ip") + if not isinstance(ip, str): + raise TypeError("reservation ip must be a str") + _validate_ip(ip, version=4, field="reservation ip") + + return StaticReservation(index=str(index), mac=mac.lower(), ip=str(ipaddress.ip_address(ip))) + + @staticmethod + def _reservation_cmp(items: list[StaticReservation]) -> list[tuple[str, str, str]]: + return sorted( + ((item["index"], item["mac"].lower(), item["ip"]) for item in items), + key=lambda value: int(value[0]), + ) + + @staticmethod + def _validate_recovery_options( + *, + write_timeout: float, + recovery_attempts: int, + recovery_delay: float, + recovery_timeout: float, + ) -> None: if write_timeout <= 0: raise ValueError("write_timeout must be greater than zero") if recovery_attempts <= 0: @@ -258,72 +585,6 @@ def _set_dns_fields( if recovery_timeout <= 0: raise ValueError("recovery_timeout must be greater than zero") - before = self.dhcp() - missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in before] - if missing: - raise ProtocolError( - "refusing combined DHCP write because the read-back object is " - f"missing required fields: {', '.join(missing)}" - ) - - payload: dict[str, Any] = dict(before) - payload.update(expected) - - write_error: NR2301Error | None = None - try: - self._client.multicall( - [ - { - "path": "router", - "method": "router_set_dhcp_settings_comb", - "data": payload, - "timeout": write_timeout, - } - ], - timeout=float(write_timeout), - ) - except (TransportError, ProtocolError) as exc: - # The documented write may reset management TCP. The write outcome - # is therefore determined by read-back, not by transport success. - write_error = exc - - last_actual: DNSSettings | None = None - last_error: NR2301Error | None = None - - for attempt in range(recovery_attempts): - try: - actual = self.dns(timeout=recovery_timeout) - last_actual = actual - if actual == expected: - return actual - except NR2301Error as exc: - last_error = exc - if self._client.password is not None: - try: - self._client.login() - except NR2301Error as login_exc: - last_error = login_exc - - if attempt + 1 < recovery_attempts and recovery_delay: - time.sleep(recovery_delay) - - details: dict[str, Any] = { - "expected": dict(expected), - "actual": dict(last_actual) if last_actual is not None else None, - } - if write_error is not None: - details["write_transport_error"] = type(write_error).__name__ - if last_error is not None: - details["last_recovery_error"] = type(last_error).__name__ - - raise APIError( - "DNS write could not be verified by exact read-back; " - "the router state may be unchanged or the management connection " - "may still be recovering", - method_id="router/router_set_dhcp_settings_comb", - response=details, - ) - @staticmethod def _extract_dhcp(response: Mapping[str, Any]) -> Mapping[str, Any]: dhcp = response.get("dhcp") From 13ecdb54a91d33864a7293084fdb7fd097a0b2e6 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 09:59:52 +0200 Subject: [PATCH 02/27] feat: add complete LAN DHCP write helpers --- src/nr2301/namespaces/lan.py | 328 ++++++++++++++++++++++++++++++++++- 1 file changed, 322 insertions(+), 6 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index 6baca1c..fbc429c 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -71,6 +71,12 @@ class StaticReservation(TypedDict): ip: str +class StaticReservation(TypedDict): + index: str + mac: str + ip: str + + class DNSSettings(TypedDict): """Verified DNS subset of the combined DHCP settings object.""" @@ -432,6 +438,295 @@ def set_address_legacy( response=details, ) + def set_dhcp_settings( + self, + settings: Mapping[str, Any], + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> DHCPSettings: + """Write the complete verified 12-field combined DHCP object. + + Start from `lan.dhcp()`, modify only intended fields, then pass the + complete object here. Missing values are never invented. + """ + + if not isinstance(settings, Mapping): + raise TypeError("settings must be a mapping") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] + if missing: + raise ProtocolError( + "combined DHCP write requires all fields: " + ", ".join(missing) + ) + + payload: dict[str, str] = {} + for key in _REQUIRED_COMBINED_FIELDS: + value = settings[key] + if not isinstance(value, str): + raise TypeError(f"{key} must be a str") + payload[key] = value + + current = self.dhcp(timeout=recovery_timeout) + current_cmp = {key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS} + if current_cmp == payload: + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [ + { + "path": "router", + "method": "router_set_dhcp_settings_comb", + "data": payload, + "timeout": write_timeout, + } + ], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: DHCPSettings | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.dhcp(timeout=recovery_timeout) + last_actual = actual + actual_cmp = { + key: actual.get(key) for key in _REQUIRED_COMBINED_FIELDS + } + if actual_cmp == payload: + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": payload, + "actual": ( + { + key: last_actual.get(key) + for key in _REQUIRED_COMBINED_FIELDS + } + if last_actual is not None + else None + ), + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + + raise APIError( + "combined DHCP write could not be verified by exact read-back", + method_id="router/router_set_dhcp_settings_comb", + response=details, + ) + + def static_reservation_list( + self, + *, + timeout: float | None = None, + ) -> list[StaticReservation]: + """Return normalized static DHCP reservation items.""" + + response = self.static_reservations(timeout=timeout) + dhcp = response.get("dhcp") + if not isinstance(dhcp, Mapping): + raise ProtocolError( + "router/router_get_dhcp_static_ip did not return a dhcp object" + ) + raw = dhcp.get("data") + if not isinstance(raw, list): + raise ProtocolError( + "router/router_get_dhcp_static_ip did not return dhcp.data as a list" + ) + + result: list[StaticReservation] = [] + for item in raw: + if not isinstance(item, Mapping): + raise ProtocolError("static DHCP reservation is not an object") + result.append(self._normalize_reservation(item)) + return result + + def set_static_reservations( + self, + reservations: list[Mapping[str, Any]], + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> list[StaticReservation]: + """Replace the complete static DHCP reservation table.""" + + if not isinstance(reservations, list): + raise TypeError("reservations must be a list") + if len(reservations) > 10: + raise ValueError("the stock frontend supports at most 10 reservations") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + expected = [self._normalize_reservation(item) for item in reservations] + indices = [item["index"] for item in expected] + if len(indices) != len(set(indices)): + raise ValueError("reservation indices must be unique") + + current = self.static_reservation_list(timeout=recovery_timeout) + if self._reservation_cmp(current) == self._reservation_cmp(expected): + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [ + { + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": expected}, + "timeout": write_timeout, + } + ], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: list[StaticReservation] | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.static_reservation_list(timeout=recovery_timeout) + last_actual = actual + if self._reservation_cmp(actual) == self._reservation_cmp(expected): + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected_count": len(expected), + "actual_count": len(last_actual) if last_actual is not None else None, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + + raise APIError( + "static DHCP reservation write could not be verified by read-back", + method_id="router/router_set_dhcp_static_ip", + response=details, + ) + + def set_address_legacy( + self, + lan_ip: str, + lan_netmask: str, + *, + write_timeout: float = 30.0, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + ) -> LANAddressResponse: + """Use the deprecated dedicated LAN-address setter and require read-back.""" + + _validate_ip(lan_ip, version=4, field="lan_ip") + _validate_ip(lan_netmask, version=4, field="lan_netmask") + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + + current = self.address(timeout=recovery_timeout) + current_router = current.get("router") + if ( + isinstance(current_router, Mapping) + and current_router.get("lan_ip") == lan_ip + and current_router.get("lan_netmask") == lan_netmask + ): + return current + + write_error: NR2301Error | None = None + try: + self._client.call( + "router", + "router_set_lan_ip", + data={"lan_ip": lan_ip, "lan_netmask": lan_netmask}, + timeout=write_timeout, + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: LANAddressResponse | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.address(timeout=recovery_timeout) + last_actual = actual + actual_router = actual.get("router") + if ( + isinstance(actual_router, Mapping) + and actual_router.get("lan_ip") == lan_ip + and actual_router.get("lan_netmask") == lan_netmask + ): + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": {"lan_ip": lan_ip, "lan_netmask": lan_netmask}, + "actual": last_actual, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + + raise APIError( + "legacy LAN address write could not be verified by read-back", + method_id="router/router_set_lan_ip", + response=details, + ) + def dns(self, *, timeout: float | None = None) -> DNSSettings: """Return the five DNS fields from the combined DHCP object.""" @@ -532,7 +827,10 @@ def _set_dns_fields( ) return cast( DNSSettings, - {key: verified[key] for key in ("dnsmode", "dns1", "dns2", "ipv6dns1", "ipv6dns2")}, + { + key: verified[key] + for key in ("dnsmode", "dns1", "dns2", "ipv6dns1", "ipv6dns2") + }, ) @staticmethod @@ -551,20 +849,38 @@ def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: if not isinstance(mac, str): raise TypeError("reservation mac must be a str") parts = mac.split(":") - if len(parts) != 6 or any(len(part) != 2 or any(ch not in "0123456789abcdefABCDEF" for ch in part) for part in parts): - raise ValueError("reservation mac must be a colon-separated MAC address") + if ( + len(parts) != 6 + or any( + len(part) != 2 + or any(ch not in "0123456789abcdefABCDEF" for ch in part) + for part in parts + ) + ): + raise ValueError( + "reservation mac must be a colon-separated MAC address" + ) ip = item.get("ip") if not isinstance(ip, str): raise TypeError("reservation ip must be a str") _validate_ip(ip, version=4, field="reservation ip") - return StaticReservation(index=str(index), mac=mac.lower(), ip=str(ipaddress.ip_address(ip))) + return StaticReservation( + index=str(index), + mac=mac.lower(), + ip=str(ipaddress.ip_address(ip)), + ) @staticmethod - def _reservation_cmp(items: list[StaticReservation]) -> list[tuple[str, str, str]]: + def _reservation_cmp( + items: list[StaticReservation], + ) -> list[tuple[str, str, str]]: return sorted( - ((item["index"], item["mac"].lower(), item["ip"]) for item in items), + ( + (item["index"], item["mac"].lower(), item["ip"]) + for item in items + ), key=lambda value: int(value[0]), ) From 3861d9e83a719ac7d37162eb8ce920c5e3be781f Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:00:40 +0200 Subject: [PATCH 03/27] feat: allow explicit legacy LAN setter verification --- src/nr2301/namespaces/lan.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index fbc429c..3cadaff 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -377,8 +377,14 @@ def set_address_legacy( recovery_attempts: int = 10, recovery_delay: float = 1.0, recovery_timeout: float = 3.0, + force: bool = False, ) -> LANAddressResponse: - """Use the deprecated dedicated LAN-address setter and require read-back.""" + """Use the deprecated dedicated LAN-address setter and require read-back. + + `force=True` is intended for explicit transport verification when the + caller wants to execute the setter even though the requested address + already matches the current state. + """ _validate_ip(lan_ip, version=4, field="lan_ip") _validate_ip(lan_netmask, version=4, field="lan_netmask") @@ -670,8 +676,11 @@ def set_address_legacy( current = self.address(timeout=recovery_timeout) current_router = current.get("router") + if not isinstance(force, bool): + raise TypeError("force must be a bool") if ( - isinstance(current_router, Mapping) + not force + and isinstance(current_router, Mapping) and current_router.get("lan_ip") == lan_ip and current_router.get("lan_netmask") == lan_netmask ): From 577afb602e7b334aad602718dd18cfd14ee2c93f Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:01:03 +0200 Subject: [PATCH 04/27] feat: add disruptive work mode setter --- src/nr2301/namespaces/device.py | 100 +++++++++++++++++++++++++++++++- 1 file changed, 99 insertions(+), 1 deletion(-) diff --git a/src/nr2301/namespaces/device.py b/src/nr2301/namespaces/device.py index 4ef3e69..5fc33b5 100644 --- a/src/nr2301/namespaces/device.py +++ b/src/nr2301/namespaces/device.py @@ -2,9 +2,10 @@ from __future__ import annotations +import time from typing import TYPE_CHECKING, Any, TypedDict, cast -from ..exceptions import APIError, ProtocolError +from ..exceptions import APIError, NR2301Error, ProtocolError, TransportError if TYPE_CHECKING: from ..client import NR2301Client @@ -182,6 +183,94 @@ def work_mode(self, *, timeout: float | None = None) -> WorkMode: self._client.call("router", "router_get_work_mode", timeout=timeout), ) + def set_work_mode( + self, + mode: str, + *, + write_timeout: int = 30, + recovery_attempts: int = 10, + recovery_delay: float = 1.0, + recovery_timeout: float = 3.0, + force: bool = False, + ) -> WorkMode: + """Set router/bridge work mode through the verified multicall setter. + + Changing work mode can fundamentally alter addressing and management + reachability. A lost write response is therefore inconclusive; success + requires recovery plus exact getter read-back. + + `force=True` executes the setter even when the current mode already + matches, which is useful for explicit transport verification. + """ + + if mode not in {"router", "bridge"}: + raise ValueError("mode must be 'router' or 'bridge'") + if write_timeout <= 0: + raise ValueError("write_timeout must be greater than zero") + if recovery_attempts <= 0: + raise ValueError("recovery_attempts must be greater than zero") + if recovery_delay < 0: + raise ValueError("recovery_delay must not be negative") + if recovery_timeout <= 0: + raise ValueError("recovery_timeout must be greater than zero") + if not isinstance(force, bool): + raise TypeError("force must be a bool") + + current = self.work_mode(timeout=recovery_timeout) + current_mode = self._work_mode_code(current) + if current_mode == mode and not force: + return current + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [ + { + "path": "router", + "method": "router_set_work_mode", + "data": {"work_mode": mode}, + "timeout": write_timeout, + } + ], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: str | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + verified = self.work_mode(timeout=recovery_timeout) + last_actual = self._work_mode_code(verified) + if last_actual == mode: + return verified + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": mode, + "actual": last_actual, + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + + raise APIError( + "router work-mode write could not be verified by exact read-back", + method_id="router/router_set_work_mode", + response=details, + ) + def set_ui_language( self, language: str, @@ -292,6 +381,15 @@ def set_sleep_wait_time( ) return verified + @staticmethod + def _work_mode_code(response: WorkMode) -> str: + value: Any = response.get("mode") + if not isinstance(value, str) or not value: + raise ProtocolError( + "router/router_get_work_mode did not return a non-empty mode string" + ) + return value + @staticmethod def _ui_language_code(response: UILanguage) -> str: value: Any = response.get("language") From 0f6655cb08e6475a36a3de19e03ca4d0a13d87f3 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:01:29 +0200 Subject: [PATCH 05/27] feat: export static DHCP reservation type --- src/nr2301/namespaces/__init__.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/nr2301/namespaces/__init__.py b/src/nr2301/namespaces/__init__.py index 4d9f25c..4327b8f 100644 --- a/src/nr2301/namespaces/__init__.py +++ b/src/nr2301/namespaces/__init__.py @@ -43,6 +43,7 @@ LANAddress, LANAddressResponse, LANNamespace, + StaticReservation, ) from .maintenance import MaintenanceNamespace, MaintenanceRecoveryResult, TimedRebootSettings from .mobile import ( @@ -215,6 +216,7 @@ "SMSSendResult", "SignalInfo", "SleepWaitTime", + "StaticReservation", "StatisticsNamespace", "TimedRebootSettings", "TrafficCounters", From 2d6c0c2c71a5b1cae6f21cbc4b6911a25aa5235d Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:02:04 +0200 Subject: [PATCH 06/27] test: cover complete LAN and DHCP write helpers --- tests/test_lan_namespace.py | 202 ++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) diff --git a/tests/test_lan_namespace.py b/tests/test_lan_namespace.py index d1d174e..c2fa622 100644 --- a/tests/test_lan_namespace.py +++ b/tests/test_lan_namespace.py @@ -226,3 +226,205 @@ def test_legacy_dhcp_settings_rejects_malformed_multicall(payload): with pytest.raises(ProtocolError): client.lan.legacy_dhcp_settings() + + + +def test_set_dhcp_settings_writes_complete_multicall_object_and_verifies(): + before = dhcp_payload() + after = dhcp_payload(leasetime="43200") + client, session = authenticated_client( + [(before, 200), ({"responses": [{"result": 0}]}, 200), (after, 200)] + ) + + requested = dict(after["dhcp"]) + result = client.lan.set_dhcp_settings(requested, recovery_attempts=1) + + assert result["leasetime"] == "43200" + method, _, kwargs = session.calls[1] + assert method == "POST" + assert kwargs["params"] == {"multicalls": 1} + member = kwargs["json"]["requests"][0] + assert member["path"] == "router" + assert member["method"] == "router_set_dhcp_settings_comb" + assert member["timeout"] == 30 + assert member["data"] == requested + assert set(member["data"]) == { + "disabled", + "lan_ip", + "lan_netmask", + "start", + "end", + "leasetime", + "mtu", + "dnsmode", + "dns1", + "dns2", + "ipv6dns1", + "ipv6dns2", + } + + +def test_set_dhcp_settings_uses_readback_after_transport_failure(): + before = dhcp_payload() + after = dhcp_payload(mtu="1499") + client, _ = authenticated_client( + [(before, 200), ({}, 500), (after, 200)] + ) + + result = client.lan.set_dhcp_settings( + dict(after["dhcp"]), + recovery_attempts=1, + ) + + assert result["mtu"] == "1499" + + +def test_set_dhcp_settings_rejects_partial_or_non_string_fields(): + incomplete = dict(dhcp_payload()["dhcp"]) + incomplete.pop("mtu") + client, session = authenticated_client([]) + + with pytest.raises(ProtocolError, match="requires all fields"): + client.lan.set_dhcp_settings(incomplete) + + wrong_type = dict(dhcp_payload()["dhcp"]) + wrong_type["mtu"] = 1500 + with pytest.raises(TypeError, match="mtu must be a str"): + client.lan.set_dhcp_settings(wrong_type) + + assert session.calls == [] + + +def test_static_reservation_list_normalizes_verified_item_shape(): + payload = { + "dhcp": { + "cnt": 1, + "data": [ + { + "index": 0, + "mac": "02:AA:BB:CC:DD:EE", + "ip": "192.0.2.10", + } + ], + } + } + client, _ = authenticated_client([(payload, 200)]) + + assert client.lan.static_reservation_list() == [ + { + "index": "0", + "mac": "02:aa:bb:cc:dd:ee", + "ip": "192.0.2.10", + } + ] + + +def test_set_static_reservations_uses_multicall_and_exact_readback(): + before = {"dhcp": {"cnt": 0, "data": []}} + expected_item = { + "index": "0", + "mac": "02:00:00:00:00:01", + "ip": "192.0.2.254", + } + after = {"dhcp": {"cnt": 1, "data": [expected_item]}} + client, session = authenticated_client( + [(before, 200), ({"responses": [{"result": 0}]}, 200), (after, 200)] + ) + + result = client.lan.set_static_reservations( + [expected_item], + recovery_attempts=1, + ) + + assert result == [expected_item] + method, _, kwargs = session.calls[1] + assert method == "POST" + assert kwargs["params"] == {"multicalls": 1} + member = kwargs["json"]["requests"][0] + assert member == { + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": [expected_item]}, + "timeout": 30, + } + + +def test_set_static_reservations_recovers_after_write_transport_failure(): + before = {"dhcp": {"cnt": 0, "data": []}} + expected_item = { + "index": "0", + "mac": "02:00:00:00:00:01", + "ip": "192.0.2.254", + } + after = {"dhcp": {"cnt": 1, "data": [expected_item]}} + client, _ = authenticated_client( + [(before, 200), ({}, 500), (after, 200)] + ) + + assert client.lan.set_static_reservations( + [expected_item], + recovery_attempts=1, + ) == [expected_item] + + +@pytest.mark.parametrize( + "item", + [ + {"index": "10", "mac": "02:00:00:00:00:01", "ip": "192.0.2.10"}, + {"index": "0", "mac": "invalid", "ip": "192.0.2.10"}, + {"index": "0", "mac": "02:00:00:00:00:01", "ip": "not-an-ip"}, + ], +) +def test_set_static_reservations_rejects_invalid_items_before_network(item): + client, session = authenticated_client([]) + + with pytest.raises((TypeError, ValueError)): + client.lan.set_static_reservations([item]) + + assert session.calls == [] + + +def test_set_address_legacy_force_executes_same_state_setter_and_readback(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + client, session = authenticated_client( + [(address, 200), ({"router": {"setting_response": "OK"}}, 200), (address, 200)] + ) + + result = client.lan.set_address_legacy( + "192.168.1.1", + "255.255.255.0", + force=True, + recovery_attempts=1, + ) + + assert result == address + assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] + _, _, kwargs = session.calls[1] + assert kwargs["params"]["path"] == "router" + assert kwargs["params"]["method"] == "router_set_lan_ip" + assert kwargs["json"] == { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + + +def test_set_address_legacy_same_state_without_force_avoids_write(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + client, session = authenticated_client([(address, 200)]) + + assert client.lan.set_address_legacy( + "192.168.1.1", + "255.255.255.0", + ) == address + + assert [call[0] for call in session.calls] == ["GET"] From f63d671728c36cb6ca4d9c9490ed4c478e2bd985 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:02:22 +0200 Subject: [PATCH 07/27] test: cover disruptive work mode setter --- tests/test_device_namespace.py | 78 ++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/tests/test_device_namespace.py b/tests/test_device_namespace.py index a64ca48..948efc4 100644 --- a/tests/test_device_namespace.py +++ b/tests/test_device_namespace.py @@ -210,3 +210,81 @@ def test_set_sleep_wait_time_raises_when_readback_does_not_match(): assert exc_info.value.method_id == "aoc/set_sleep_wait_time" assert exc_info.value.response == {"expected": 20, "actual": 30} + + + +def test_set_work_mode_uses_verified_multicall_and_readback(): + client, session = authenticated_client( + {"mode": "router", "result": 0}, + {"responses": [{"result": 0}]}, + {"mode": "bridge", "result": 0}, + ) + + result = client.device.set_work_mode( + "bridge", + recovery_attempts=1, + ) + + assert result == {"mode": "bridge", "result": 0} + assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] + _, _, kwargs = session.calls[1] + assert kwargs["params"] == {"multicalls": 1} + assert kwargs["json"] == { + "requests": [ + { + "path": "router", + "method": "router_set_work_mode", + "data": {"work_mode": "bridge"}, + "timeout": 30, + } + ] + } + + +def test_set_work_mode_force_executes_same_state_write(): + client, session = authenticated_client( + {"mode": "router", "result": 0}, + {"responses": [{"result": 0}]}, + {"mode": "router", "result": 0}, + ) + + result = client.device.set_work_mode( + "router", + force=True, + recovery_attempts=1, + ) + + assert result["mode"] == "router" + assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] + + +def test_set_work_mode_uses_readback_after_transport_failure(): + client, _ = authenticated_client( + {"mode": "router", "result": 0}, + FakeResponse({}, status_code=500), + {"mode": "bridge", "result": 0}, + ) + + result = client.device.set_work_mode( + "bridge", + recovery_attempts=1, + ) + + assert result["mode"] == "bridge" + + +def test_set_work_mode_same_state_without_force_avoids_write(): + client, session = authenticated_client({"mode": "router", "result": 0}) + + assert client.device.set_work_mode("router")["mode"] == "router" + assert [call[0] for call in session.calls] == ["GET"] + + +@pytest.mark.parametrize("mode", ["", "Router", "modem", "invalid"]) +def test_set_work_mode_rejects_unknown_mode_before_network(mode): + client, session = authenticated_client() + + with pytest.raises(ValueError, match="router.*bridge"): + client.device.set_work_mode(mode) + + assert session.calls == [] From a2686f2ad749f6cd73ea1c3fe85f7683f5434f16 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:03:18 +0200 Subject: [PATCH 08/27] feat: validate complete DHCP write payload --- src/nr2301/namespaces/lan.py | 40 ++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index 3cadaff..4877a12 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -209,6 +209,8 @@ def set_dhcp_settings( raise TypeError(f"{key} must be a str") payload[key] = value + self._validate_dhcp_payload(payload) + current = self.dhcp(timeout=recovery_timeout) current_cmp = {key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS} if current_cmp == payload: @@ -842,6 +844,44 @@ def _set_dns_fields( }, ) + @staticmethod + def _validate_dhcp_payload(payload: Mapping[str, str]) -> None: + if payload["disabled"] not in {"0", "1"}: + raise ValueError("disabled must be '0' or '1'") + + for field in ("lan_ip", "start", "end"): + _validate_ip(payload[field], version=4, field=field) + + try: + ipaddress.IPv4Network( + f"0.0.0.0/{payload['lan_netmask']}", + strict=False, + ) + except ValueError as exc: + raise ValueError("lan_netmask must be a valid IPv4 netmask") from exc + + try: + lease = int(payload["leasetime"]) + except ValueError as exc: + raise ValueError("leasetime must be an integer string") from exc + if lease < 60 or lease > 604800: + raise ValueError("leasetime must be between 60 and 604800 seconds") + + try: + mtu = int(payload["mtu"]) + except ValueError as exc: + raise ValueError("mtu must be an integer string") from exc + if mtu < 1280 or mtu > 1500: + raise ValueError("mtu must be between 1280 and 1500") + + if payload["dnsmode"] not in {"auto", "manual"}: + raise ValueError("dnsmode must be 'auto' or 'manual'") + + _validate_optional_ip(payload["dns1"], version=4, field="dns1") + _validate_optional_ip(payload["dns2"], version=4, field="dns2") + _validate_optional_ip(payload["ipv6dns1"], version=6, field="ipv6dns1") + _validate_optional_ip(payload["ipv6dns2"], version=6, field="ipv6dns2") + @staticmethod def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: raw_index = item.get("index") From aded23d1beb264824e6288778beb4cc2b3015e87 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:04:02 +0200 Subject: [PATCH 09/27] refactor: share combined DHCP recovery writer --- src/nr2301/namespaces/lan.py | 204 ++++++++++++++++++++++------------- 1 file changed, 129 insertions(+), 75 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index 4877a12..e1c9804 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -184,12 +184,11 @@ def set_dhcp_settings( ) -> DHCPSettings: """Write the complete verified 12-field combined DHCP object. - Callers should start from `lan.dhcp()`, modify only intended fields, - then pass the full object here. The helper never invents missing values. + Start from `lan.dhcp()`, modify only intended fields, then pass the + complete object here. Missing values are never invented. """ - if not isinstance(settings, Mapping): - raise TypeError("settings must be a mapping") + payload = self._normalize_dhcp_payload(settings) self._validate_recovery_options( write_timeout=write_timeout, recovery_attempts=recovery_attempts, @@ -197,75 +196,14 @@ def set_dhcp_settings( recovery_timeout=recovery_timeout, ) - payload: dict[str, str] = {} - missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] - if missing: - raise ProtocolError( - "combined DHCP write requires all fields: " + ", ".join(missing) - ) - for key in _REQUIRED_COMBINED_FIELDS: - value = settings[key] - if not isinstance(value, str): - raise TypeError(f"{key} must be a str") - payload[key] = value - - self._validate_dhcp_payload(payload) - current = self.dhcp(timeout=recovery_timeout) - current_cmp = {key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS} - if current_cmp == payload: - return current - - write_error: NR2301Error | None = None - try: - self._client.multicall( - [{ - "path": "router", - "method": "router_set_dhcp_settings_comb", - "data": payload, - "timeout": write_timeout, - }], - timeout=float(write_timeout), - ) - except (TransportError, ProtocolError) as exc: - write_error = exc - - last_actual: DHCPSettings | None = None - last_error: NR2301Error | None = None - for attempt in range(recovery_attempts): - try: - actual = self.dhcp(timeout=recovery_timeout) - last_actual = actual - actual_cmp = { - key: actual.get(key) for key in _REQUIRED_COMBINED_FIELDS - } - if actual_cmp == payload: - return actual - except NR2301Error as exc: - last_error = exc - if self._client.password is not None: - try: - self._client.login() - except NR2301Error as login_exc: - last_error = login_exc - if attempt + 1 < recovery_attempts and recovery_delay: - time.sleep(recovery_delay) - - details: dict[str, Any] = { - "expected": payload, - "actual": ( - {key: last_actual.get(key) for key in _REQUIRED_COMBINED_FIELDS} - if last_actual is not None else None - ), - } - if write_error is not None: - details["write_transport_error"] = type(write_error).__name__ - if last_error is not None: - details["last_recovery_error"] = type(last_error).__name__ - raise APIError( - "combined DHCP write could not be verified by exact read-back", - method_id="router/router_set_dhcp_settings_comb", - response=details, + return self._write_dhcp_payload( + payload, + current=current, + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, ) def static_reservation_list( @@ -397,9 +335,17 @@ def set_address_legacy( recovery_timeout=recovery_timeout, ) + if not isinstance(force, bool): + raise TypeError("force must be a bool") + current = self.address(timeout=recovery_timeout) router = current.get("router") - if isinstance(router, Mapping) and router.get("lan_ip") == lan_ip and router.get("lan_netmask") == lan_netmask: + if ( + not force + and isinstance(router, Mapping) + and router.get("lan_ip") == lan_ip + and router.get("lan_netmask") == lan_netmask + ): return current write_error: NR2301Error | None = None @@ -829,8 +775,16 @@ def _set_dns_fields( before = self.dhcp() payload: dict[str, Any] = dict(before) payload.update(expected) - verified = self.set_dhcp_settings( - payload, + normalized = self._normalize_dhcp_payload(payload) + self._validate_recovery_options( + write_timeout=write_timeout, + recovery_attempts=recovery_attempts, + recovery_delay=recovery_delay, + recovery_timeout=recovery_timeout, + ) + verified = self._write_dhcp_payload( + normalized, + current=before, write_timeout=write_timeout, recovery_attempts=recovery_attempts, recovery_delay=recovery_delay, @@ -844,6 +798,106 @@ def _set_dns_fields( }, ) + def _write_dhcp_payload( + self, + payload: dict[str, str], + *, + current: Mapping[str, Any], + write_timeout: int, + recovery_attempts: int, + recovery_delay: float, + recovery_timeout: float, + ) -> DHCPSettings: + current_cmp = { + key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS + } + if current_cmp == payload: + return cast(DHCPSettings, dict(current)) + + write_error: NR2301Error | None = None + try: + self._client.multicall( + [ + { + "path": "router", + "method": "router_set_dhcp_settings_comb", + "data": payload, + "timeout": write_timeout, + } + ], + timeout=float(write_timeout), + ) + except (TransportError, ProtocolError) as exc: + write_error = exc + + last_actual: DHCPSettings | None = None + last_error: NR2301Error | None = None + for attempt in range(recovery_attempts): + try: + actual = self.dhcp(timeout=recovery_timeout) + last_actual = actual + actual_cmp = { + key: actual.get(key) for key in _REQUIRED_COMBINED_FIELDS + } + if actual_cmp == payload: + return actual + except NR2301Error as exc: + last_error = exc + if self._client.password is not None: + try: + self._client.login() + except NR2301Error as login_exc: + last_error = login_exc + + if attempt + 1 < recovery_attempts and recovery_delay: + time.sleep(recovery_delay) + + details: dict[str, Any] = { + "expected": payload, + "actual": ( + { + key: last_actual.get(key) + for key in _REQUIRED_COMBINED_FIELDS + } + if last_actual is not None + else None + ), + } + if write_error is not None: + details["write_transport_error"] = type(write_error).__name__ + if last_error is not None: + details["last_recovery_error"] = type(last_error).__name__ + + raise APIError( + "combined DHCP write could not be verified by exact read-back", + method_id="router/router_set_dhcp_settings_comb", + response=details, + ) + + @classmethod + def _normalize_dhcp_payload( + cls, + settings: Mapping[str, Any], + ) -> dict[str, str]: + if not isinstance(settings, Mapping): + raise TypeError("settings must be a mapping") + + missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] + if missing: + raise ProtocolError( + "combined DHCP write requires all fields: " + ", ".join(missing) + ) + + payload: dict[str, str] = {} + for key in _REQUIRED_COMBINED_FIELDS: + value = settings[key] + if not isinstance(value, str): + raise TypeError(f"{key} must be a str") + payload[key] = value + + cls._validate_dhcp_payload(payload) + return payload + @staticmethod def _validate_dhcp_payload(payload: Mapping[str, str]) -> None: if payload["disabled"] not in {"0", "1"}: From 40a0af95fc34bea1e99b540029ea106dfb3a7038 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:04:34 +0200 Subject: [PATCH 10/27] fix: remove duplicated LAN write helper block --- src/nr2301/namespaces/lan.py | 292 ----------------------------------- 1 file changed, 292 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index e1c9804..a55540e 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -392,298 +392,6 @@ def set_address_legacy( response=details, ) - def set_dhcp_settings( - self, - settings: Mapping[str, Any], - *, - write_timeout: int = 30, - recovery_attempts: int = 10, - recovery_delay: float = 1.0, - recovery_timeout: float = 3.0, - ) -> DHCPSettings: - """Write the complete verified 12-field combined DHCP object. - - Start from `lan.dhcp()`, modify only intended fields, then pass the - complete object here. Missing values are never invented. - """ - - if not isinstance(settings, Mapping): - raise TypeError("settings must be a mapping") - self._validate_recovery_options( - write_timeout=write_timeout, - recovery_attempts=recovery_attempts, - recovery_delay=recovery_delay, - recovery_timeout=recovery_timeout, - ) - - missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] - if missing: - raise ProtocolError( - "combined DHCP write requires all fields: " + ", ".join(missing) - ) - - payload: dict[str, str] = {} - for key in _REQUIRED_COMBINED_FIELDS: - value = settings[key] - if not isinstance(value, str): - raise TypeError(f"{key} must be a str") - payload[key] = value - - current = self.dhcp(timeout=recovery_timeout) - current_cmp = {key: current.get(key) for key in _REQUIRED_COMBINED_FIELDS} - if current_cmp == payload: - return current - - write_error: NR2301Error | None = None - try: - self._client.multicall( - [ - { - "path": "router", - "method": "router_set_dhcp_settings_comb", - "data": payload, - "timeout": write_timeout, - } - ], - timeout=float(write_timeout), - ) - except (TransportError, ProtocolError) as exc: - write_error = exc - - last_actual: DHCPSettings | None = None - last_error: NR2301Error | None = None - for attempt in range(recovery_attempts): - try: - actual = self.dhcp(timeout=recovery_timeout) - last_actual = actual - actual_cmp = { - key: actual.get(key) for key in _REQUIRED_COMBINED_FIELDS - } - if actual_cmp == payload: - return actual - except NR2301Error as exc: - last_error = exc - if self._client.password is not None: - try: - self._client.login() - except NR2301Error as login_exc: - last_error = login_exc - - if attempt + 1 < recovery_attempts and recovery_delay: - time.sleep(recovery_delay) - - details: dict[str, Any] = { - "expected": payload, - "actual": ( - { - key: last_actual.get(key) - for key in _REQUIRED_COMBINED_FIELDS - } - if last_actual is not None - else None - ), - } - if write_error is not None: - details["write_transport_error"] = type(write_error).__name__ - if last_error is not None: - details["last_recovery_error"] = type(last_error).__name__ - - raise APIError( - "combined DHCP write could not be verified by exact read-back", - method_id="router/router_set_dhcp_settings_comb", - response=details, - ) - - def static_reservation_list( - self, - *, - timeout: float | None = None, - ) -> list[StaticReservation]: - """Return normalized static DHCP reservation items.""" - - response = self.static_reservations(timeout=timeout) - dhcp = response.get("dhcp") - if not isinstance(dhcp, Mapping): - raise ProtocolError( - "router/router_get_dhcp_static_ip did not return a dhcp object" - ) - raw = dhcp.get("data") - if not isinstance(raw, list): - raise ProtocolError( - "router/router_get_dhcp_static_ip did not return dhcp.data as a list" - ) - - result: list[StaticReservation] = [] - for item in raw: - if not isinstance(item, Mapping): - raise ProtocolError("static DHCP reservation is not an object") - result.append(self._normalize_reservation(item)) - return result - - def set_static_reservations( - self, - reservations: list[Mapping[str, Any]], - *, - write_timeout: int = 30, - recovery_attempts: int = 10, - recovery_delay: float = 1.0, - recovery_timeout: float = 3.0, - ) -> list[StaticReservation]: - """Replace the complete static DHCP reservation table.""" - - if not isinstance(reservations, list): - raise TypeError("reservations must be a list") - if len(reservations) > 10: - raise ValueError("the stock frontend supports at most 10 reservations") - self._validate_recovery_options( - write_timeout=write_timeout, - recovery_attempts=recovery_attempts, - recovery_delay=recovery_delay, - recovery_timeout=recovery_timeout, - ) - - expected = [self._normalize_reservation(item) for item in reservations] - indices = [item["index"] for item in expected] - if len(indices) != len(set(indices)): - raise ValueError("reservation indices must be unique") - - current = self.static_reservation_list(timeout=recovery_timeout) - if self._reservation_cmp(current) == self._reservation_cmp(expected): - return current - - write_error: NR2301Error | None = None - try: - self._client.multicall( - [ - { - "path": "router", - "method": "router_set_dhcp_static_ip", - "data": {"data": expected}, - "timeout": write_timeout, - } - ], - timeout=float(write_timeout), - ) - except (TransportError, ProtocolError) as exc: - write_error = exc - - last_actual: list[StaticReservation] | None = None - last_error: NR2301Error | None = None - for attempt in range(recovery_attempts): - try: - actual = self.static_reservation_list(timeout=recovery_timeout) - last_actual = actual - if self._reservation_cmp(actual) == self._reservation_cmp(expected): - return actual - except NR2301Error as exc: - last_error = exc - if self._client.password is not None: - try: - self._client.login() - except NR2301Error as login_exc: - last_error = login_exc - - if attempt + 1 < recovery_attempts and recovery_delay: - time.sleep(recovery_delay) - - details: dict[str, Any] = { - "expected_count": len(expected), - "actual_count": len(last_actual) if last_actual is not None else None, - } - if write_error is not None: - details["write_transport_error"] = type(write_error).__name__ - if last_error is not None: - details["last_recovery_error"] = type(last_error).__name__ - - raise APIError( - "static DHCP reservation write could not be verified by read-back", - method_id="router/router_set_dhcp_static_ip", - response=details, - ) - - def set_address_legacy( - self, - lan_ip: str, - lan_netmask: str, - *, - write_timeout: float = 30.0, - recovery_attempts: int = 10, - recovery_delay: float = 1.0, - recovery_timeout: float = 3.0, - ) -> LANAddressResponse: - """Use the deprecated dedicated LAN-address setter and require read-back.""" - - _validate_ip(lan_ip, version=4, field="lan_ip") - _validate_ip(lan_netmask, version=4, field="lan_netmask") - self._validate_recovery_options( - write_timeout=write_timeout, - recovery_attempts=recovery_attempts, - recovery_delay=recovery_delay, - recovery_timeout=recovery_timeout, - ) - - current = self.address(timeout=recovery_timeout) - current_router = current.get("router") - if not isinstance(force, bool): - raise TypeError("force must be a bool") - if ( - not force - and isinstance(current_router, Mapping) - and current_router.get("lan_ip") == lan_ip - and current_router.get("lan_netmask") == lan_netmask - ): - return current - - write_error: NR2301Error | None = None - try: - self._client.call( - "router", - "router_set_lan_ip", - data={"lan_ip": lan_ip, "lan_netmask": lan_netmask}, - timeout=write_timeout, - ) - except (TransportError, ProtocolError) as exc: - write_error = exc - - last_actual: LANAddressResponse | None = None - last_error: NR2301Error | None = None - for attempt in range(recovery_attempts): - try: - actual = self.address(timeout=recovery_timeout) - last_actual = actual - actual_router = actual.get("router") - if ( - isinstance(actual_router, Mapping) - and actual_router.get("lan_ip") == lan_ip - and actual_router.get("lan_netmask") == lan_netmask - ): - return actual - except NR2301Error as exc: - last_error = exc - if self._client.password is not None: - try: - self._client.login() - except NR2301Error as login_exc: - last_error = login_exc - - if attempt + 1 < recovery_attempts and recovery_delay: - time.sleep(recovery_delay) - - details: dict[str, Any] = { - "expected": {"lan_ip": lan_ip, "lan_netmask": lan_netmask}, - "actual": last_actual, - } - if write_error is not None: - details["write_transport_error"] = type(write_error).__name__ - if last_error is not None: - details["last_recovery_error"] = type(last_error).__name__ - - raise APIError( - "legacy LAN address write could not be verified by read-back", - method_id="router/router_set_lan_ip", - response=details, - ) - def dns(self, *, timeout: float | None = None) -> DNSSettings: """Return the five DNS fields from the combined DHCP object.""" From 8fec59a0272138ce388dbc1edad5373c922515f5 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:05:53 +0200 Subject: [PATCH 11/27] test: add physical LAN router write lifecycle --- tests/integration/test_lan_router_writes.py | 245 ++++++++++++++++++++ 1 file changed, 245 insertions(+) create mode 100644 tests/integration/test_lan_router_writes.py diff --git a/tests/integration/test_lan_router_writes.py b/tests/integration/test_lan_router_writes.py new file mode 100644 index 0000000..93865ea --- /dev/null +++ b/tests/integration/test_lan_router_writes.py @@ -0,0 +1,245 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import os + +import pytest + +from nr2301 import NR2301Client + + +if os.environ.get("NR2301_DESTRUCTIVE_INTEGRATION") != "1": + pytest.skip( + "physical LAN/router write tests require " + "NR2301_DESTRUCTIVE_INTEGRATION=1", + allow_module_level=True, + ) + + +pytestmark = pytest.mark.integration + + +@pytest.fixture(scope="module") +def router(): + password = os.environ.get("NR2301_PASSWORD") + if not password: + pytest.skip("NR2301_PASSWORD is required") + + with NR2301Client( + os.environ.get("NR2301_URL", "http://zyxel.home"), + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=15.0, + ) as client: + client.login() + yield client + + +def _dhcp_fingerprint(value): + keys = ( + "disabled", + "lan_ip", + "lan_netmask", + "start", + "end", + "leasetime", + "mtu", + "dnsmode", + "dns1", + "dns2", + "ipv6dns1", + "ipv6dns2", + ) + return tuple((key, value.get(key)) for key in keys) + + +def _reservation_fingerprint(items): + return sorted( + ( + str(item["index"]), + str(item["mac"]).lower(), + str(item["ip"]), + ) + for item in items + ) + + +def _synthetic_reservation(original): + used_indices = {int(item["index"]) for item in original} + free = next((index for index in range(10) if index not in used_indices), None) + + candidates = [ + ("02:00:00:00:00:fe", "192.0.2.254"), + ("02:00:00:00:00:fd", "192.0.2.253"), + ] + + if free is not None: + for mac, ip in candidates: + candidate = {"index": str(free), "mac": mac, "ip": ip} + if candidate not in original: + return original + [candidate] + + # All slots occupied: temporarily replace the highest slot. The complete + # original table is restored in finally. + target = max(original, key=lambda item: int(item["index"])) + for mac, ip in candidates: + candidate = { + "index": str(target["index"]), + "mac": mac, + "ip": ip, + } + if ( + str(target["mac"]).lower() != mac + or str(target["ip"]) != ip + ): + return [ + candidate if item["index"] == target["index"] else dict(item) + for item in original + ] + + raise AssertionError("could not construct a distinct synthetic reservation") + + +def test_lan_router_write_lifecycle_and_exact_restore(router): + original_dhcp = router.lan.dhcp(timeout=5.0) + original_reservations = router.lan.static_reservation_list(timeout=5.0) + original_address = router.lan.address(timeout=5.0) + original_mode = router.device.work_mode(timeout=5.0) + + original_dhcp_fp = _dhcp_fingerprint(original_dhcp) + original_reservation_fp = _reservation_fingerprint(original_reservations) + + router_address = original_address.get("router") + if not isinstance(router_address, dict): + pytest.fail("router_get_lan_ip returned no router object") + lan_ip = router_address.get("lan_ip") + lan_netmask = router_address.get("lan_netmask") + if not isinstance(lan_ip, str) or not isinstance(lan_netmask, str): + pytest.fail("router_get_lan_ip returned invalid address fields") + + mode = original_mode.get("mode") + if mode not in {"router", "bridge"}: + pytest.fail("router_get_work_mode returned an unknown mode") + + mutated_dhcp = dict(original_dhcp) + mutated_dhcp["leasetime"] = ( + "43200" if original_dhcp.get("leasetime") != "43200" else "86400" + ) + synthetic_reservations = _synthetic_reservation( + [dict(item) for item in original_reservations] + ) + + dhcp_changed = False + reservations_changed = False + + try: + verified_dhcp = router.lan.set_dhcp_settings( + mutated_dhcp, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + dhcp_changed = True + assert _dhcp_fingerprint(verified_dhcp) == _dhcp_fingerprint(mutated_dhcp) + assert _dhcp_fingerprint(verified_dhcp) != original_dhcp_fp + print( + "LAN_DHCP_WRITE" + " field=leasetime" + " changed=True" + " readback=True", + flush=True, + ) + + verified_reservations = router.lan.set_static_reservations( + synthetic_reservations, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + reservations_changed = True + assert _reservation_fingerprint(verified_reservations) == ( + _reservation_fingerprint(synthetic_reservations) + ) + assert _reservation_fingerprint(verified_reservations) != ( + original_reservation_fp + ) + print( + "LAN_STATIC_RESERVATION_WRITE" + f" original_count={len(original_reservations)}" + f" synthetic_count={len(synthetic_reservations)}" + " readback=True", + flush=True, + ) + + verified_address = router.lan.set_address_legacy( + lan_ip, + lan_netmask, + force=True, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + verified_router = verified_address.get("router") + assert isinstance(verified_router, dict) + assert verified_router.get("lan_ip") == lan_ip + assert verified_router.get("lan_netmask") == lan_netmask + print( + "LAN_LEGACY_ADDRESS_WRITE force_same_state=True readback=True", + flush=True, + ) + + verified_mode = router.device.set_work_mode( + mode, + force=True, + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert verified_mode.get("mode") == mode + print( + "ROUTER_WORK_MODE_WRITE" + f" force_same_state=True" + f" mode_preserved=True", + flush=True, + ) + + finally: + if reservations_changed: + restored_reservations = router.lan.set_static_reservations( + [dict(item) for item in original_reservations], + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert _reservation_fingerprint(restored_reservations) == ( + original_reservation_fp + ) + + if dhcp_changed: + restored_dhcp = router.lan.set_dhcp_settings( + dict(original_dhcp), + recovery_attempts=30, + recovery_delay=1.0, + recovery_timeout=4.0, + ) + assert _dhcp_fingerprint(restored_dhcp) == original_dhcp_fp + + final_dhcp = router.lan.dhcp(timeout=5.0) + final_reservations = router.lan.static_reservation_list(timeout=5.0) + final_address = router.lan.address(timeout=5.0) + final_mode = router.device.work_mode(timeout=5.0) + + assert _dhcp_fingerprint(final_dhcp) == original_dhcp_fp + assert _reservation_fingerprint(final_reservations) == original_reservation_fp + assert final_address == original_address + assert final_mode == original_mode + + print( + "LAN_ROUTER_FINAL" + " dhcp_restored=True" + " reservations_restored=True" + " address_preserved=True" + " work_mode_preserved=True", + flush=True, + ) From 8bd2336f8e6f5cca285e9a006851b2b5fea1032c Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:07:01 +0200 Subject: [PATCH 12/27] fix: preserve established DHCP validation error --- src/nr2301/namespaces/lan.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index a55540e..86cebd0 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -593,7 +593,8 @@ def _normalize_dhcp_payload( missing = [key for key in _REQUIRED_COMBINED_FIELDS if key not in settings] if missing: raise ProtocolError( - "combined DHCP write requires all fields: " + ", ".join(missing) + "refusing combined DHCP write because the read-back object is " + "missing required fields: " + ", ".join(missing) ) payload: dict[str, str] = {} From 7420791b2fbb46744cc070cd326072479457d82f Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:07:58 +0200 Subject: [PATCH 13/27] test: align DHCP validation expectation --- tests/test_lan_namespace.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_lan_namespace.py b/tests/test_lan_namespace.py index c2fa622..0a6e778 100644 --- a/tests/test_lan_namespace.py +++ b/tests/test_lan_namespace.py @@ -284,7 +284,7 @@ def test_set_dhcp_settings_rejects_partial_or_non_string_fields(): incomplete.pop("mtu") client, session = authenticated_client([]) - with pytest.raises(ProtocolError, match="requires all fields"): + with pytest.raises(ProtocolError, match="missing required fields"): client.lan.set_dhcp_settings(incomplete) wrong_type = dict(dhcp_payload()["dhcp"]) From 5247ddc5a67697a6cc713e8148356e071f1ab8fc Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:15:49 +0200 Subject: [PATCH 14/27] research: add read-only work mode response probe --- examples/probe_work_mode_response.py | 32 ++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 examples/probe_work_mode_response.py diff --git a/examples/probe_work_mode_response.py b/examples/probe_work_mode_response.py new file mode 100644 index 0000000..c051ce5 --- /dev/null +++ b/examples/probe_work_mode_response.py @@ -0,0 +1,32 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import json +import os + +from nr2301 import NR2301Client + + +def main() -> None: + password = os.environ.get("NR2301_PASSWORD") + if not password: + raise RuntimeError("NR2301_PASSWORD is required") + + with NR2301Client( + os.environ.get("NR2301_URL", "http://zyxel.home"), + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=10.0, + ) as client: + client.login() + response = client.device.work_mode(timeout=5.0) + + print( + "WORK_MODE_RESPONSE = " + + json.dumps(response, ensure_ascii=True, sort_keys=True) + ) + + +if __name__ == "__main__": + main() From 02305c444dd4afe055d194bc8e820650ea022a80 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:32:08 +0200 Subject: [PATCH 15/27] fix: use live work_mode response field --- src/nr2301/namespaces/device.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/nr2301/namespaces/device.py b/src/nr2301/namespaces/device.py index 5fc33b5..8f9bd80 100644 --- a/src/nr2301/namespaces/device.py +++ b/src/nr2301/namespaces/device.py @@ -87,8 +87,7 @@ class UILanguage(TypedDict, total=False): class WorkMode(TypedDict, total=False): - mode: str - result: int + work_mode: str class BatteryInfo(TypedDict, total=False): @@ -383,10 +382,11 @@ def set_sleep_wait_time( @staticmethod def _work_mode_code(response: WorkMode) -> str: - value: Any = response.get("mode") + value: Any = response.get("work_mode") if not isinstance(value, str) or not value: raise ProtocolError( - "router/router_get_work_mode did not return a non-empty mode string" + "router/router_get_work_mode did not return a non-empty " + "work_mode string" ) return value From c6504a3b19398ddbcd856a95d5f23d9702d5f55e Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:32:17 +0200 Subject: [PATCH 16/27] test: align work mode fixtures with live response --- tests/test_device_namespace.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/tests/test_device_namespace.py b/tests/test_device_namespace.py index 948efc4..2223cab 100644 --- a/tests/test_device_namespace.py +++ b/tests/test_device_namespace.py @@ -65,7 +65,7 @@ def test_device_internet_preserves_documented_raw_access_value(): def test_work_mode_preserves_unknown_raw_mode_without_coercion(): - payload = {"mode": "future-mode", "result": 0} + payload = {"work_mode": "future-mode"} client, _ = authenticated_client(payload) assert client.device.work_mode() == payload @@ -215,9 +215,9 @@ def test_set_sleep_wait_time_raises_when_readback_does_not_match(): def test_set_work_mode_uses_verified_multicall_and_readback(): client, session = authenticated_client( - {"mode": "router", "result": 0}, + {"work_mode": "router"}, {"responses": [{"result": 0}]}, - {"mode": "bridge", "result": 0}, + {"work_mode": "bridge"}, ) result = client.device.set_work_mode( @@ -225,7 +225,7 @@ def test_set_work_mode_uses_verified_multicall_and_readback(): recovery_attempts=1, ) - assert result == {"mode": "bridge", "result": 0} + assert result == {"work_mode": "bridge"} assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] _, _, kwargs = session.calls[1] assert kwargs["params"] == {"multicalls": 1} @@ -243,9 +243,9 @@ def test_set_work_mode_uses_verified_multicall_and_readback(): def test_set_work_mode_force_executes_same_state_write(): client, session = authenticated_client( - {"mode": "router", "result": 0}, + {"work_mode": "router"}, {"responses": [{"result": 0}]}, - {"mode": "router", "result": 0}, + {"work_mode": "router"}, ) result = client.device.set_work_mode( @@ -254,15 +254,15 @@ def test_set_work_mode_force_executes_same_state_write(): recovery_attempts=1, ) - assert result["mode"] == "router" + assert result["work_mode"] == "router" assert [call[0] for call in session.calls] == ["GET", "POST", "GET"] def test_set_work_mode_uses_readback_after_transport_failure(): client, _ = authenticated_client( - {"mode": "router", "result": 0}, + {"work_mode": "router"}, FakeResponse({}, status_code=500), - {"mode": "bridge", "result": 0}, + {"work_mode": "bridge"}, ) result = client.device.set_work_mode( @@ -270,13 +270,13 @@ def test_set_work_mode_uses_readback_after_transport_failure(): recovery_attempts=1, ) - assert result["mode"] == "bridge" + assert result["work_mode"] == "bridge" def test_set_work_mode_same_state_without_force_avoids_write(): - client, session = authenticated_client({"mode": "router", "result": 0}) + client, session = authenticated_client({"work_mode": "router"}) - assert client.device.set_work_mode("router")["mode"] == "router" + assert client.device.set_work_mode("router")["work_mode"] == "router" assert [call[0] for call in session.calls] == ["GET"] From b63a0665ce121ebdc78af2e3fcf7a999375d20f1 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:32:23 +0200 Subject: [PATCH 17/27] test: use live work_mode field in physical lifecycle --- tests/integration/test_lan_router_writes.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/integration/test_lan_router_writes.py b/tests/integration/test_lan_router_writes.py index 93865ea..f913fad 100644 --- a/tests/integration/test_lan_router_writes.py +++ b/tests/integration/test_lan_router_writes.py @@ -118,7 +118,7 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): if not isinstance(lan_ip, str) or not isinstance(lan_netmask, str): pytest.fail("router_get_lan_ip returned invalid address fields") - mode = original_mode.get("mode") + mode = original_mode.get("work_mode") if mode not in {"router", "bridge"}: pytest.fail("router_get_work_mode returned an unknown mode") @@ -196,7 +196,7 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): recovery_delay=1.0, recovery_timeout=4.0, ) - assert verified_mode.get("mode") == mode + assert verified_mode.get("work_mode") == mode print( "ROUTER_WORK_MODE_WRITE" f" force_same_state=True" From dd283cfde58def505234c581e9c871ebc63d8e16 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:41:08 +0200 Subject: [PATCH 18/27] research: capture static DHCP frontend contract --- examples/capture_dhcp_static_contract.py | 101 +++++++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 examples/capture_dhcp_static_contract.py diff --git a/examples/capture_dhcp_static_contract.py b/examples/capture_dhcp_static_contract.py new file mode 100644 index 0000000..7c68536 --- /dev/null +++ b/examples/capture_dhcp_static_contract.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from urllib.parse import urljoin, urlsplit + +from nr2301 import NR2301Client + + +TARGETS = ( + "router_set_dhcp_static_ip", + "router_get_dhcp_static_ip", + "static", + "mapping", + "mac", + "index", + "toStringData", +) + + +def main() -> None: + password = os.environ.get("NR2301_PASSWORD") + if not password: + raise RuntimeError("NR2301_PASSWORD is required") + + out_dir = Path( + os.environ.get( + "NR2301_DHCP_SOURCE_OUT", + "dhcp_static_contract_capture", + ) + ).resolve() + out_dir.mkdir(parents=True, exist_ok=True) + + base = os.environ.get("NR2301_URL", "http://zyxel.home").rstrip("/") + origin = f"{urlsplit(base).scheme}://{urlsplit(base).netloc}" + + with NR2301Client( + base, + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=15.0, + ) as client: + client.login() + session = client.transport.session + + page_url = urljoin(base + "/", "html/set_dhcp.html") + response = session.get(page_url, timeout=15.0) + response.raise_for_status() + page = response.text + + script_urls = [] + for match in re.finditer( + r"""]+src\s*=\s*['"]([^'"]+)['"]""", + page, + re.IGNORECASE, + ): + url = urljoin(page_url, match.group(1)) + if urlsplit(url).netloc == urlsplit(origin).netloc: + script_urls.append(url) + + assets = [(page_url, page)] + for url in script_urls: + r = session.get(url, timeout=15.0) + r.raise_for_status() + assets.append((url, r.text)) + + report = {"assets": []} + for url, text in assets: + hits = [] + for target in TARGETS: + for match in re.finditer(re.escape(target), text, re.IGNORECASE): + start = max(0, match.start() - 2200) + end = min(len(text), match.end() + 4200) + snippet = text[start:end] + hits.append( + { + "target": target, + "start": start, + "end": end, + "snippet": snippet, + } + ) + if hits: + report["assets"].append({"url": url, "hits": hits}) + + report_path = out_dir / "report.json" + report_path.write_text( + json.dumps(report, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + print(f"DHCP_STATIC_SOURCE_REPORT = {report_path}") + print("DHCP_STATIC_SOURCE_CAPTURE = PASS") + + +if __name__ == "__main__": + main() From a74d886b0e3ed1383b6083931601aa350858282d Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:41:49 +0200 Subject: [PATCH 19/27] research: guard cleanup of failed static DHCP probe --- examples/check_cleanup_dhcp_static_residue.py | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 examples/check_cleanup_dhcp_static_residue.py diff --git a/examples/check_cleanup_dhcp_static_residue.py b/examples/check_cleanup_dhcp_static_residue.py new file mode 100644 index 0000000..3ad447e --- /dev/null +++ b/examples/check_cleanup_dhcp_static_residue.py @@ -0,0 +1,96 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import os +import time + +from nr2301 import NR2301Client, NR2301Error + + +SYNTHETIC = { + "index": "0", + "mac": "02:00:00:00:00:fe", + "ip": "192.0.2.254", +} + + +def fingerprint(item): + return ( + str(item["index"]), + str(item["mac"]).lower(), + str(item["ip"]), + ) + + +def main() -> None: + password = os.environ.get("NR2301_PASSWORD") + if not password: + raise RuntimeError("NR2301_PASSWORD is required") + + with NR2301Client( + os.environ.get("NR2301_URL", "http://zyxel.home"), + username=os.environ.get("NR2301_USERNAME", "admin"), + password=password, + timeout=15.0, + ) as client: + client.login() + current = client.lan.static_reservation_list(timeout=5.0) + + synthetic_present = any( + fingerprint(item) == fingerprint(SYNTHETIC) + for item in current + ) + print( + "DHCP_STATIC_RESIDUE" + f" count={len(current)}" + f" synthetic_present={synthetic_present}" + ) + + if not synthetic_present: + print("DHCP_STATIC_RESIDUE_CLEANUP = NOT_NEEDED") + return + + if len(current) != 1: + raise RuntimeError( + "synthetic reservation is present together with other entries; " + "refusing automatic cleanup" + ) + + # The failed integration test started from an empty reservation table. + # Clear only the exact sole synthetic residue using the already + # historically live-verified empty-table restore shape. + try: + client.multicall( + [ + { + "path": "router", + "method": "router_set_dhcp_static_ip", + "data": {"data": []}, + "timeout": 30, + } + ], + timeout=30.0, + ) + except NR2301Error: + pass + + for attempt in range(30): + try: + after = client.lan.static_reservation_list(timeout=4.0) + if not after: + print("DHCP_STATIC_RESIDUE_CLEANUP = PASS") + return + except NR2301Error: + try: + client.login() + except NR2301Error: + pass + if attempt < 29: + time.sleep(1.0) + + raise RuntimeError("synthetic DHCP reservation residue could not be cleared") + + +if __name__ == "__main__": + main() From 1611f56830195647ddcf0b9b2599074fe2d3b40f Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:42:13 +0200 Subject: [PATCH 20/27] test: restore LAN state after attempted writes --- tests/integration/test_lan_router_writes.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/integration/test_lan_router_writes.py b/tests/integration/test_lan_router_writes.py index f913fad..8685695 100644 --- a/tests/integration/test_lan_router_writes.py +++ b/tests/integration/test_lan_router_writes.py @@ -130,17 +130,17 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): [dict(item) for item in original_reservations] ) - dhcp_changed = False - reservations_changed = False + dhcp_attempted = False + reservations_attempted = False try: + dhcp_attempted = True verified_dhcp = router.lan.set_dhcp_settings( mutated_dhcp, recovery_attempts=30, recovery_delay=1.0, recovery_timeout=4.0, ) - dhcp_changed = True assert _dhcp_fingerprint(verified_dhcp) == _dhcp_fingerprint(mutated_dhcp) assert _dhcp_fingerprint(verified_dhcp) != original_dhcp_fp print( @@ -151,13 +151,13 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): flush=True, ) + reservations_attempted = True verified_reservations = router.lan.set_static_reservations( synthetic_reservations, recovery_attempts=30, recovery_delay=1.0, recovery_timeout=4.0, ) - reservations_changed = True assert _reservation_fingerprint(verified_reservations) == ( _reservation_fingerprint(synthetic_reservations) ) @@ -205,7 +205,7 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): ) finally: - if reservations_changed: + if reservations_attempted: restored_reservations = router.lan.set_static_reservations( [dict(item) for item in original_reservations], recovery_attempts=30, @@ -216,7 +216,7 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): original_reservation_fp ) - if dhcp_changed: + if dhcp_attempted: restored_dhcp = router.lan.set_dhcp_settings( dict(original_dhcp), recovery_attempts=30, From e7673044a963b76f36dcabbe76422d1dcb8f908f Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:51:12 +0200 Subject: [PATCH 21/27] fix: match static DHCP frontend wire contract --- src/nr2301/namespaces/lan.py | 65 ++++++++++++++++++++++++++++++------ 1 file changed, 54 insertions(+), 11 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index 86cebd0..0f5c0bf 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -66,13 +66,7 @@ class LANAddressResponse(TypedDict, total=False): class StaticReservation(TypedDict): - index: str - mac: str - ip: str - - -class StaticReservation(TypedDict): - index: str + index: int mac: str ip: str @@ -258,6 +252,16 @@ def set_static_reservations( if len(indices) != len(set(indices)): raise ValueError("reservation indices must be unique") + macs = [item["mac"].lower() for item in expected] + if len(macs) != len(set(macs)): + raise ValueError("reservation MAC addresses must be unique") + ips = [item["ip"] for item in expected] + if len(ips) != len(set(ips)): + raise ValueError("reservation IPv4 addresses must be unique") + + address = self.address(timeout=recovery_timeout) + self._validate_reservations_against_lan(expected, address) + current = self.static_reservation_list(timeout=recovery_timeout) if self._reservation_cmp(current) == self._reservation_cmp(expected): return current @@ -678,22 +682,61 @@ def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: raise TypeError("reservation ip must be a str") _validate_ip(ip, version=4, field="reservation ip") + normalized_mac = mac.lower() + first_octet = int(normalized_mac.split(":")[0], 16) + if first_octet & 1: + raise ValueError("reservation mac must not be multicast") + return StaticReservation( - index=str(index), - mac=mac.lower(), + index=index, + mac=normalized_mac, ip=str(ipaddress.ip_address(ip)), ) + @staticmethod + def _validate_reservations_against_lan( + items: list[StaticReservation], + address: Mapping[str, Any], + ) -> None: + router = address.get("router") + if not isinstance(router, Mapping): + raise ProtocolError( + "router/router_get_lan_ip did not return a router object" + ) + lan_ip = router.get("lan_ip") + lan_netmask = router.get("lan_netmask") + if not isinstance(lan_ip, str) or not isinstance(lan_netmask, str): + raise ProtocolError( + "router/router_get_lan_ip did not return usable LAN address fields" + ) + try: + network = ipaddress.IPv4Network( + f"{lan_ip}/{lan_netmask}", + strict=False, + ) + except ValueError as exc: + raise ProtocolError( + "router/router_get_lan_ip returned an invalid LAN/network pair" + ) from exc + + for item in items: + candidate = ipaddress.IPv4Address(item["ip"]) + if candidate not in network: + raise ValueError( + f"reservation ip {item['ip']} must be inside current LAN subnet " + f"{network.with_netmask}" + ) + @staticmethod def _reservation_cmp( items: list[StaticReservation], - ) -> list[tuple[str, str, str]]: + ) -> list[tuple[int, str, str]]: return sorted( ( (item["index"], item["mac"].lower(), item["ip"]) for item in items ), - key=lambda value: int(value[0]), + key=lambda value: value[0], ) @staticmethod From 4209bd24ddd11e8f8992007125b4d27b586f1adb Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:51:55 +0200 Subject: [PATCH 22/27] test: match static DHCP frontend validation --- tests/test_lan_namespace.py | 128 +++++++++++++++++++++++++++++++++--- 1 file changed, 120 insertions(+), 8 deletions(-) diff --git a/tests/test_lan_namespace.py b/tests/test_lan_namespace.py index 0a6e778..14477c6 100644 --- a/tests/test_lan_namespace.py +++ b/tests/test_lan_namespace.py @@ -106,7 +106,7 @@ def test_set_dns_preserves_combined_settings_and_verifies_readback(): "ipv6dns2": "2606:4700:4700::1001", } - method, _, kwargs = session.calls[1] + method, _, kwargs = session.calls[2] assert method == "POST" assert kwargs["params"] == {"multicalls": 1} member = kwargs["json"]["requests"][0] @@ -312,7 +312,7 @@ def test_static_reservation_list_normalizes_verified_item_shape(): assert client.lan.static_reservation_list() == [ { - "index": "0", + "index": 0, "mac": "02:aa:bb:cc:dd:ee", "ip": "192.0.2.10", } @@ -320,15 +320,26 @@ def test_static_reservation_list_normalizes_verified_item_shape(): def test_set_static_reservations_uses_multicall_and_exact_readback(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } before = {"dhcp": {"cnt": 0, "data": []}} expected_item = { - "index": "0", + "index": 0, "mac": "02:00:00:00:00:01", - "ip": "192.0.2.254", + "ip": "192.168.1.254", } after = {"dhcp": {"cnt": 1, "data": [expected_item]}} client, session = authenticated_client( - [(before, 200), ({"responses": [{"result": 0}]}, 200), (after, 200)] + [ + (address, 200), + (before, 200), + ({"responses": [{"result": 0}]}, 200), + (after, 200), + ] ) result = client.lan.set_static_reservations( @@ -350,15 +361,21 @@ def test_set_static_reservations_uses_multicall_and_exact_readback(): def test_set_static_reservations_recovers_after_write_transport_failure(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } before = {"dhcp": {"cnt": 0, "data": []}} expected_item = { - "index": "0", + "index": 0, "mac": "02:00:00:00:00:01", - "ip": "192.0.2.254", + "ip": "192.168.1.254", } after = {"dhcp": {"cnt": 1, "data": [expected_item]}} client, _ = authenticated_client( - [(before, 200), ({}, 500), (after, 200)] + [(address, 200), (before, 200), ({}, 500), (after, 200)] ) assert client.lan.set_static_reservations( @@ -428,3 +445,98 @@ def test_set_address_legacy_same_state_without_force_avoids_write(): ) == address assert [call[0] for call in session.calls] == ["GET"] + + + +def test_set_static_reservations_rejects_ip_outside_current_lan_before_write(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + client, session = authenticated_client([(address, 200)]) + + with pytest.raises(ValueError, match="inside current LAN subnet"): + client.lan.set_static_reservations( + [ + { + "index": 0, + "mac": "02:00:00:00:00:01", + "ip": "192.0.2.254", + } + ] + ) + + assert len(session.calls) == 1 + assert session.calls[0][0] == "GET" + + +def test_set_static_reservations_sends_numeric_index_even_if_input_is_string(): + address = { + "router": { + "lan_ip": "192.168.1.1", + "lan_netmask": "255.255.255.0", + } + } + before = {"dhcp": {"cnt": 0, "data": []}} + after = { + "dhcp": { + "cnt": 1, + "data": [ + { + "index": 0, + "mac": "02:00:00:00:00:01", + "ip": "192.168.1.254", + } + ], + } + } + client, session = authenticated_client( + [ + (address, 200), + (before, 200), + ({"responses": [{"data": {"dhcp": {"setting_response": "OK"}}}]}, 200), + (after, 200), + ] + ) + + client.lan.set_static_reservations( + [ + { + "index": "0", + "mac": "02:00:00:00:00:01", + "ip": "192.168.1.254", + } + ], + recovery_attempts=1, + ) + + member = session.calls[2][2]["json"]["requests"][0] + assert member["data"]["data"][0]["index"] == 0 + assert isinstance(member["data"]["data"][0]["index"], int) + + +@pytest.mark.parametrize( + "items", + [ + [ + {"index": 0, "mac": "02:00:00:00:00:01", "ip": "192.168.1.10"}, + {"index": 1, "mac": "02:00:00:00:00:01", "ip": "192.168.1.11"}, + ], + [ + {"index": 0, "mac": "02:00:00:00:00:01", "ip": "192.168.1.10"}, + {"index": 1, "mac": "02:00:00:00:00:02", "ip": "192.168.1.10"}, + ], + [ + {"index": 0, "mac": "01:00:5e:00:00:01", "ip": "192.168.1.10"}, + ], + ], +) +def test_set_static_reservations_matches_frontend_duplicate_and_multicast_checks(items): + client, session = authenticated_client([]) + + with pytest.raises(ValueError): + client.lan.set_static_reservations(items) + + assert session.calls == [] From 907d8edfdacade0eabe1198a2ca0cf7a64ebab1a Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:52:37 +0200 Subject: [PATCH 23/27] test: fix static DHCP call ordering --- tests/test_lan_namespace.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_lan_namespace.py b/tests/test_lan_namespace.py index 14477c6..2a12dd4 100644 --- a/tests/test_lan_namespace.py +++ b/tests/test_lan_namespace.py @@ -106,7 +106,7 @@ def test_set_dns_preserves_combined_settings_and_verifies_readback(): "ipv6dns2": "2606:4700:4700::1001", } - method, _, kwargs = session.calls[2] + method, _, kwargs = session.calls[1] assert method == "POST" assert kwargs["params"] == {"multicalls": 1} member = kwargs["json"]["requests"][0] @@ -348,7 +348,7 @@ def test_set_static_reservations_uses_multicall_and_exact_readback(): ) assert result == [expected_item] - method, _, kwargs = session.calls[1] + method, _, kwargs = session.calls[2] assert method == "POST" assert kwargs["params"] == {"multicalls": 1} member = kwargs["json"]["requests"][0] From 2f72f1d0373011e72780aa502712b2ba9685a417 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:53:13 +0200 Subject: [PATCH 24/27] test: choose valid static reservation in active LAN --- tests/integration/test_lan_router_writes.py | 88 +++++++++++++++------ 1 file changed, 62 insertions(+), 26 deletions(-) diff --git a/tests/integration/test_lan_router_writes.py b/tests/integration/test_lan_router_writes.py index 8685695..b46f698 100644 --- a/tests/integration/test_lan_router_writes.py +++ b/tests/integration/test_lan_router_writes.py @@ -2,6 +2,7 @@ from __future__ import annotations +import ipaddress import os import pytest @@ -65,40 +66,74 @@ def _reservation_fingerprint(items): ) -def _synthetic_reservation(original): +def _synthetic_reservation(original, dhcp): used_indices = {int(item["index"]) for item in original} free = next((index for index in range(10) if index not in used_indices), None) - candidates = [ - ("02:00:00:00:00:fe", "192.0.2.254"), - ("02:00:00:00:00:fd", "192.0.2.253"), - ] + lan_ip = ipaddress.IPv4Address(dhcp["lan_ip"]) + network = ipaddress.IPv4Network( + f"{dhcp['lan_ip']}/{dhcp['lan_netmask']}", + strict=False, + ) + pool_start = ipaddress.IPv4Address(dhcp["start"]) + pool_end = ipaddress.IPv4Address(dhcp["end"]) + used_ips = { + ipaddress.IPv4Address(item["ip"]) + for item in original + } + + candidate_ip = None + lower = int(network.network_address) + 1 + upper = int(network.broadcast_address) - 1 + for raw in range(upper, lower - 1, -1): + candidate = ipaddress.IPv4Address(raw) + if candidate == lan_ip: + continue + if candidate in used_ips: + continue + if pool_start <= candidate <= pool_end: + continue + candidate_ip = str(candidate) + break + + if candidate_ip is None: + raise AssertionError( + "no free static-reservation address exists inside the LAN subnet " + "and outside the DHCP pool" + ) + + used_macs = {str(item["mac"]).lower() for item in original} + mac = next( + ( + candidate + for candidate in ( + "02:00:00:00:00:fe", + "02:00:00:00:00:fd", + "02:00:00:00:00:fc", + ) + if candidate not in used_macs + ), + None, + ) + if mac is None: + raise AssertionError("could not construct a distinct synthetic MAC") if free is not None: - for mac, ip in candidates: - candidate = {"index": str(free), "mac": mac, "ip": ip} - if candidate not in original: - return original + [candidate] + candidate = {"index": free, "mac": mac, "ip": candidate_ip} + return [dict(item) for item in original] + [candidate] # All slots occupied: temporarily replace the highest slot. The complete # original table is restored in finally. target = max(original, key=lambda item: int(item["index"])) - for mac, ip in candidates: - candidate = { - "index": str(target["index"]), - "mac": mac, - "ip": ip, - } - if ( - str(target["mac"]).lower() != mac - or str(target["ip"]) != ip - ): - return [ - candidate if item["index"] == target["index"] else dict(item) - for item in original - ] - - raise AssertionError("could not construct a distinct synthetic reservation") + candidate = { + "index": int(target["index"]), + "mac": mac, + "ip": candidate_ip, + } + return [ + candidate if int(item["index"]) == int(target["index"]) else dict(item) + for item in original + ] def test_lan_router_write_lifecycle_and_exact_restore(router): @@ -127,7 +162,8 @@ def test_lan_router_write_lifecycle_and_exact_restore(router): "43200" if original_dhcp.get("leasetime") != "43200" else "86400" ) synthetic_reservations = _synthetic_reservation( - [dict(item) for item in original_reservations] + [dict(item) for item in original_reservations], + original_dhcp, ) dhcp_attempted = False From 5c13b37b5c5805ee63112610433a0a32cec3110a Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 12:25:18 +0200 Subject: [PATCH 25/27] fix: normalize static DHCP getter MAC format --- src/nr2301/namespaces/lan.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/nr2301/namespaces/lan.py b/src/nr2301/namespaces/lan.py index 0f5c0bf..4e3f742 100644 --- a/src/nr2301/namespaces/lan.py +++ b/src/nr2301/namespaces/lan.py @@ -664,7 +664,12 @@ def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: mac = item.get("mac") if not isinstance(mac, str): raise TypeError("reservation mac must be a str") - parts = mac.split(":") + + # ACIY.3 accepts colon-separated MACs on write but live getter + # read-back canonicalizes the same value to uppercase hyphen-separated + # form. Accept either representation and expose one stable SDK form. + separator = ":" if ":" in mac else "-" if "-" in mac else None + parts = mac.split(separator) if separator is not None else [] if ( len(parts) != 6 or any( @@ -674,7 +679,7 @@ def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: ) ): raise ValueError( - "reservation mac must be a colon-separated MAC address" + "reservation mac must be a colon- or hyphen-separated MAC address" ) ip = item.get("ip") @@ -682,8 +687,8 @@ def _normalize_reservation(item: Mapping[str, Any]) -> StaticReservation: raise TypeError("reservation ip must be a str") _validate_ip(ip, version=4, field="reservation ip") - normalized_mac = mac.lower() - first_octet = int(normalized_mac.split(":")[0], 16) + normalized_mac = ":".join(part.lower() for part in parts) + first_octet = int(parts[0], 16) if first_octet & 1: raise ValueError("reservation mac must not be multicast") From 37cb8e1bf59f13e51ed8f3f20ada9838a0ad6983 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 12:25:53 +0200 Subject: [PATCH 26/27] test: cover hyphenated static DHCP read-back --- tests/test_lan_namespace.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/test_lan_namespace.py b/tests/test_lan_namespace.py index 2a12dd4..0d36e57 100644 --- a/tests/test_lan_namespace.py +++ b/tests/test_lan_namespace.py @@ -302,7 +302,7 @@ def test_static_reservation_list_normalizes_verified_item_shape(): "data": [ { "index": 0, - "mac": "02:AA:BB:CC:DD:EE", + "mac": "02-AA-BB-CC-DD-EE", "ip": "192.0.2.10", } ], @@ -332,7 +332,18 @@ def test_set_static_reservations_uses_multicall_and_exact_readback(): "mac": "02:00:00:00:00:01", "ip": "192.168.1.254", } - after = {"dhcp": {"cnt": 1, "data": [expected_item]}} + after = { + "dhcp": { + "cnt": 1, + "data": [ + { + "index": 0, + "mac": "02-00-00-00-00-01", + "ip": "192.168.1.254", + } + ], + } + } client, session = authenticated_client( [ (address, 200), @@ -389,6 +400,7 @@ def test_set_static_reservations_recovers_after_write_transport_failure(): [ {"index": "10", "mac": "02:00:00:00:00:01", "ip": "192.0.2.10"}, {"index": "0", "mac": "invalid", "ip": "192.0.2.10"}, + {"index": "0", "mac": "02:00-AA:BB:CC:DD", "ip": "192.0.2.10"}, {"index": "0", "mac": "02:00:00:00:00:01", "ip": "not-an-ip"}, ], ) From 9f3c80f30174d6f20e3c9a02c555a00b9e16d991 Mon Sep 17 00:00:00 2001 From: Marc Leinen <34336531+MarcLeinenDE@users.noreply.github.com> Date: Mon, 21 Sep 2026 12:26:13 +0200 Subject: [PATCH 27/27] fix: clean current static DHCP test residue safely --- examples/check_cleanup_dhcp_static_residue.py | 31 +++++++++---------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/examples/check_cleanup_dhcp_static_residue.py b/examples/check_cleanup_dhcp_static_residue.py index 3ad447e..3856a9c 100644 --- a/examples/check_cleanup_dhcp_static_residue.py +++ b/examples/check_cleanup_dhcp_static_residue.py @@ -8,19 +8,19 @@ from nr2301 import NR2301Client, NR2301Error -SYNTHETIC = { - "index": "0", - "mac": "02:00:00:00:00:fe", - "ip": "192.0.2.254", +SYNTHETIC_MACS = { + "02:00:00:00:00:fe", + "02:00:00:00:00:fd", + "02:00:00:00:00:fc", } -def fingerprint(item): - return ( - str(item["index"]), - str(item["mac"]).lower(), - str(item["ip"]), - ) +def normalized_mac(value): + return str(value).replace("-", ":").lower() + + +def is_synthetic(item): + return normalized_mac(item["mac"]) in SYNTHETIC_MACS def main() -> None: @@ -37,10 +37,7 @@ def main() -> None: client.login() current = client.lan.static_reservation_list(timeout=5.0) - synthetic_present = any( - fingerprint(item) == fingerprint(SYNTHETIC) - for item in current - ) + synthetic_present = any(is_synthetic(item) for item in current) print( "DHCP_STATIC_RESIDUE" f" count={len(current)}" @@ -57,9 +54,9 @@ def main() -> None: "refusing automatic cleanup" ) - # The failed integration test started from an empty reservation table. - # Clear only the exact sole synthetic residue using the already - # historically live-verified empty-table restore shape. + # The integration probe starts from an empty reservation table and + # uses only the reserved locally-administered synthetic MACs above. + # Clear only when one such test reservation is the sole current entry. try: client.multicall( [