diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index 988e00e..c02a1dd 100644 --- a/napalm_openwrt/openwrt.py +++ b/napalm_openwrt/openwrt.py @@ -65,6 +65,11 @@ def _sources(settings: dict) -> list[str]: ] +#: ``reboot`` detached from the session and two seconds late: the answer comes +#: back first, and a hang-up from the closing session does not stop it. +_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; reboot) /dev/null 2>&1 &'" + + class OpenWrtDriver( OpenWrtInterfaceMixin, OpenWrtVLANMixin, @@ -223,6 +228,22 @@ class OpenWrtDriver( self._check_privilege(privileged) return open_stream_on_transport(self._transport(), command) + def reboot_host(self) -> None: + """Restart the AP (``HostRebootMixin``); returns once the restart is under way. + + Detached and two seconds late, so the answer comes back before the + session goes down; a refusal is raised, not mistaken for that. + + :raises RuntimeError: for a login other than root, or a refused reboot. + """ + try: + result = self.run_command(_REBOOT_DETACHED, privileged=True, timeout=30) + except PermissionError as exc: + raise RuntimeError(str(exc)) from exc + if result.exit_code: + reason = result.stderr.strip() or result.stdout.strip() + raise RuntimeError(reason or f"reboot exited with status {result.exit_code}") + def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str: """Run the listening-socket command for ``ListeningSocketsMixin``. diff --git a/pyproject.toml b/pyproject.toml index 605870b..52e591e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-openwrt" -version = "0.4.0" +version = "0.5.0" description = "NAPALM driver for OpenWrt routers/access-points" readme = "README.md" license = { text = "Apache-2.0" } diff --git a/tests/unit/test_command_channel.py b/tests/unit/test_command_channel.py index 56060fc..ccd5116 100644 --- a/tests/unit/test_command_channel.py +++ b/tests/unit/test_command_channel.py @@ -77,3 +77,33 @@ def test_no_warnings_of_its_own(): for c in OpenWrtDriver.__mro__ if c.__module__.startswith("napalm_openwrt") ) + + +class TestRebootHost: + """``reboot_host`` (napalm-device-types ``HostRebootMixin``): netOrk sent + ``/sbin/reboot`` through the private ``_send_command`` and could not tell a + refusal from the session dropping (netOrk #1113).""" + + def test_detached_so_the_answer_comes_back_first(self): + drv = _driver() + with patch("napalm_openwrt.openwrt.run_on_transport", return_value=CommandResult("", "", 0)) as run: + drv.reboot_host() + + command = run.call_args.args[1] + assert "reboot" in command and command.rstrip().endswith("&'") + assert "trap" in command and "sleep" in command + + def test_a_refusal_is_raised(self): + drv = _driver() + failed = CommandResult("", "reboot: Operation not permitted", 1) + with patch("napalm_openwrt.openwrt.run_on_transport", return_value=failed), pytest.raises( + RuntimeError, match="not permitted" + ): + drv.reboot_host() + + def test_a_login_other_than_root_is_refused(self): + drv = _driver(username="admin") + with patch("napalm_openwrt.openwrt.run_on_transport") as run, pytest.raises(RuntimeError): + drv.reboot_host() + + run.assert_not_called()