fix!: reboot_host instead of reboot; drop run_device_action (0.2.0) #7

Merged
christianmanivong merged 1 commits from fix/reboot-host into master 2026-10-17 06:59:43 +00:00
3 changed files with 28 additions and 48 deletions

No files matched your search

+13 -38
View File
@@ -24,7 +24,6 @@ from napalm.base.exceptions import ConnectionClosedException, ConnectionExceptio
from napalm_device_types import FingerprintRule, ResidentialGatewayDriver
from napalm_device_types.models import (
DeviceActionResultDict,
HostDict,
NATTranslationDict,
PortForwardDict,
@@ -806,42 +805,18 @@ class ZyxelVMGDriver(ResidentialGatewayDriver):
}
return radios
def reboot(self) -> DeviceActionResultDict:
"""Reboot the device via DAL."""
def reboot_host(self) -> None:
"""Restart the gateway (``HostRebootMixin``); raises when it refuses."""
result = self._dal_query("REBOOT", {"Command": "reboot"})
if result is not None and result.get("result") == "ZCFG_SUCCESS":
return {"success": True, "action": "reboot", "message": "Device is rebooting"}
return {"success": False, "action": "reboot", "message": "Reboot command failed"}
if result is None or result.get("result") != "ZCFG_SUCCESS":
raise RuntimeError(f"the gateway refused to reboot: {result}")
def run_device_action(self, action: str) -> DeviceActionResultDict:
"""Run a device action such as reboot or WiFi restart."""
result: DeviceActionResultDict = {
"success": False,
"action": action,
"message": "",
}
try:
if action == "reboot":
return self.reboot()
elif action.startswith("restart_wifi"):
band = "both"
if ":" in action:
band = action.split(":", 1)[1]
dal = self._dal_query("WIFI_RESTART", {"Band": band})
if dal is not None and dal.get("result") == "ZCFG_SUCCESS":
result["success"] = True
result["message"] = f"WiFi restarting ({band})"
else:
result["message"] = "WiFi restart command failed"
elif action == "get_diagnostics":
diag = self._dal_query("DIAGNOSTICS", {})
if diag:
result["success"] = True
result["message"] = json.dumps(diag)
else:
result["message"] = "Diagnostics not available"
else:
result["message"] = f"Unknown action: {action}"
except Exception as e:
result["message"] = f"Action '{action}' failed: {e}"
return result
def restart_wifi(self, band: str = "both") -> None:
"""Restart the radios of *band* (``2.4G``, ``5G`` or ``both``); raises when refused."""
result = self._dal_query("WIFI_RESTART", {"Band": band})
if result is None or result.get("result") != "ZCFG_SUCCESS":
raise RuntimeError(f"the gateway refused to restart Wi-Fi ({band}): {result}")
def get_diagnostics(self) -> dict:
"""What the gateway's diagnostics page reports; empty when it reports nothing."""
return self._dal_query("DIAGNOSTICS", {}) or {}
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-zyxel"
version = "0.1.0"
version = "0.2.0"
description = "NAPALM driver for Zyxel VMG series residential gateways (VMG4005-B50A and similar)."
readme = "README.md"
license = { text = "Apache-2.0" }
+14 -9
View File
@@ -1,6 +1,7 @@
import json
from unittest.mock import MagicMock, patch
import pytest
import requests
from napalm.base.exceptions import ConnectionException
@@ -285,16 +286,20 @@ class TestZyxelVMGDriver:
self._setup_mocked_http()
assert self.driver.get_arp_table() == []
def test_reboot(self):
self._setup_mocked_http()
result = self.driver.run_device_action("reboot")
assert result["action"] == "reboot"
def test_reboot_host_is_the_contract_the_application_calls(self):
"""It was reboot(), which nothing looks for: the gateway could not be
restarted from the application (#5)."""
self.driver._dal_query = lambda name, data: {"result": "ZCFG_SUCCESS"}
assert self.driver.reboot_host() is None
def test_run_device_action_unknown(self):
self._setup_mocked_http()
result = self.driver.run_device_action("foobar")
assert result["success"] is False
assert "Unknown" in result["message"]
def test_a_refused_reboot_raises(self):
self.driver._dal_query = lambda name, data: {"result": "ZCFG_FAIL"}
with pytest.raises(RuntimeError):
self.driver.reboot_host()
def test_run_device_action_and_reboot_are_gone(self):
assert not hasattr(type(self.driver), "run_device_action")
assert not hasattr(type(self.driver), "reboot")
def test_load_replace_candidate_raises(self):
self._setup_mocked_http()