Complete LAN and router write coverage

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"""<script[^>]+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()

examples/check_cleanup_dhcp_static_residue.py

@@ -0,0 +1,93 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from __future__ import annotations
+
+import os
+import time
+
+from nr2301 import NR2301Client, NR2301Error
+
+
+SYNTHETIC_MACS = {
+    "02:00:00:00:00:fe",
+    "02:00:00:00:00:fd",
+    "02:00:00:00:00:fc",
+}
+
+
+def normalized_mac(value):
+    return str(value).replace("-", ":").lower()
+
+
+def is_synthetic(item):
+    return normalized_mac(item["mac"]) in SYNTHETIC_MACS
+
+
+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(is_synthetic(item) 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 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(
+                [
+                    {
+                        "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()

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()

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",

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
@@ -86,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):
@@ -182,6 +182,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 +380,16 @@ def set_sleep_wait_time(
             )
         return verified
 
+    @staticmethod
+    def _work_mode_code(response: WorkMode) -> str:
+        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 "
+                "work_mode string"
+            )
+        return value
+
     @staticmethod
     def _ui_language_code(response: UILanguage) -> str:
         value: Any = response.get("language")