Merge pull request labgrid-project#1822 from AiyionPrime/feat/driver/power/poe_netgear_plus

driver/power/poe_netgear_plus: Add support

Author: Emantor

doc/configuration.rst

@@ -216,6 +216,9 @@ Currently available are:
 ``poe_mib``
   Controls PoE switches using the PoE SNMP administration MiBs.
 
+``poe_netgear_plus``
+  Controls NETGEAR Plus switches using an HTTP interface.
+
 ``raritan``
   Controls *Raritan PDUs* via SNMP.
 

labgrid/driver/power/poe_netgear_plus.py

@@ -0,0 +1,116 @@
+"""Control NETGEAR Plus devices via HTTP.
+
+Available switch models:
+https://github.com/foxey/py-netgear-plus?tab=readme-ov-file#supported-and-tested-netgear-modelsproducts-and-firmware-versions
+
+The password defaults to "P4ssword", but can be configured on a per-device basis like this:
+
+NetworkPowerPort:
+  model: poe_netgear_plus
+  host: 'http://username_is_unused:AnotherP4ssword@192.168.0.239/'
+  index: 7
+
+Omitting the password defaults as described above.
+
+NetworkPowerPort:
+  model: poe_netgear_plus
+  host: 'http://192.168.0.239/'
+  index: 7
+
+"""
+
+from urllib.parse import urlparse
+
+from py_netgear_plus import NetgearSwitchConnector
+
+from ..exception import ExecutionError
+
+
+def _get_hostname_and_password(url: str) -> tuple[str, str]:
+    """Obtain credentials from url or default and return hostname and password.
+
+    If no password is in the URL return "P4ssword", which fulfills  the minimal requirements from Netgear:
+    - 8-20 characters
+    - at least one upper case character
+    - at least one lower case character
+    - at least one number
+
+    Args:
+        url: A URL with an optional basic auth prefix.
+
+    Returns:
+        A tuple of the hostname, and the extracted or default password
+
+    """
+    parse_result = urlparse(url)
+    if parse_result.scheme != "http":
+        raise ExecutionError(f"URL must start with http://, found {parse_result.scheme} for {url}.")
+
+    password = "P4ssword" if parse_result.password is None else parse_result.password
+
+    return parse_result.hostname, password
+
+
+def power_set(host: str, _port: int, index: int, value: bool) -> None:
+    """Set the PoE output index based for a given host.
+
+    Args:
+        host: The netloc with optional password e.g. "192.168.0.239" or ":P4ssword@192.168.0.239"
+        _port: As the webserver of the switch is always on port 80, this is ignored
+        index: Zero based access to the switches network ports
+        value: Whether the port should enable PoE output
+
+    """
+    index = int(index)
+    netgear_port_number = index + 1
+
+    (hostname, password) = _get_hostname_and_password(host)
+
+    sw = NetgearSwitchConnector(hostname, password)
+    sw.autodetect_model()
+    try:
+        sw.get_login_cookie()
+        sw._get_switch_metadata()
+        if value:
+            sw.turn_on_poe_port(netgear_port_number)
+        else:
+            sw.turn_off_poe_port(netgear_port_number)
+    finally:
+        sw.delete_login_cookie()
+
+
+def power_get(host: str, _port: int, index: int) -> bool:
+    """Determine whether a given Port has PoE enabled.
+
+    Args:
+        host: The netloc with optional password e.g. "192.168.0.239" or ":P4ssword@192.168.0.239"
+        _port: As the webserver of the switch is always on port 80, this is ignored
+        index: Zero based access to the switches network ports
+
+    Returns:
+        Whether the PoE output is enabled.
+
+    Raises:
+        ExecutionError: In case the status dictionary contains unexpected PoE status values.
+
+    """
+    index = int(index)
+    netgear_port_number = index + 1
+
+    (hostname, password) = _get_hostname_and_password(host)
+
+    sw = NetgearSwitchConnector(hostname, password)
+    sw.autodetect_model()
+    try:
+        sw.get_login_cookie()
+        sw._get_switch_metadata()
+        data = sw._get_poe_port_config()
+        key = f"port_{netgear_port_number}_poe_power_active"
+        if data[key] == "on":
+            return True
+        if data[key] == "off":
+            return False
+        msg = f"Expected literal 'on'|'off', found {data[key]}"
+        raise ExecutionError(msg)
+    finally:
+        sw.delete_login_cookie()

pyproject.toml

@@ -65,6 +65,7 @@ kasa = ["python-kasa>=0.7.0"]
 modbus = ["pyModbusTCP>=0.2.0"]
 modbusrtu = ["minimalmodbus>=1.0.2"]
 mqtt = ["paho-mqtt>=2.0.0"]
+netgear = ["py-netgear-plus>=0.4.7"]
 onewire = ["onewire>=0.2"]
 pyvisa = [
     "pyvisa>=1.11.3",
@@ -78,7 +79,7 @@ vxi11 = ["python-vxi11>=0.9"]
 xena = ["xenavalkyrie>=3.0.1"]
 deb = ["labgrid[modbus,onewire,snmp]"]
 dev = [
-    "labgrid[doc,docker,graph,kasa,modbus,modbusrtu,mqtt,onewire,pyvisa,snmp,vxi11]",
+    "labgrid[doc,docker,graph,kasa,modbus,modbusrtu,mqtt,netgear,onewire,pyvisa,snmp,vxi11]",
 
     # additional dev dependencies
     "psutil>=5.8.0",
@@ -222,6 +223,7 @@ include = [
   "labgrid/driver/manualswitchdriver.py",
   "labgrid/driver/power/gude8031.py",
   "labgrid/driver/power/pe6216.py",
+  "labgrid/driver/power/poe_netgear_plus.py",
   "labgrid/driver/power/shelly_gen2.py",
   "labgrid/driver/rawnetworkinterfacedriver.py",
   "labgrid/protocol/**/*.py",

tests/test_powerdriver.py

@@ -2,14 +2,15 @@
 
 import pytest
 
+from labgrid.driver import ExecutionError
+from labgrid.driver.power.poe_netgear_plus import _get_hostname_and_password
 from labgrid.resource import NetworkPowerPort, YKUSHPowerPort
 from labgrid.driver.powerdriver import (
     ExternalPowerDriver,
     ManualPowerDriver,
     NetworkPowerDriver,
     YKUSHPowerDriver,
 )
-from labgrid.util.helper import processwrapper
 
 
 class TestManualPowerDriver:
@@ -292,6 +293,7 @@ def test_import_backends(self):
         import labgrid.driver.power.netio
         import labgrid.driver.power.netio_kshell
         import labgrid.driver.power.pe6216
+        import labgrid.driver.power.poe_netgear_plus
         import labgrid.driver.power.rest
         import labgrid.driver.power.sentry
         import labgrid.driver.power.eg_pms2_network
@@ -380,3 +382,21 @@ def test_ykush3_on(self, target, mocker):
         check_output_mock.assert_called_with(
             ["ykushcmd", "ykush3", "-s", self.YKUSH3_FAKE_SERIAL, "-u", "3"]
         )
+
+
+class TestPoeNetgearPlusPowerDriver:
+    @pytest.mark.parametrize(
+        'url, expected_host, expected_pw',
+        [
+            ("http://example.com", "example.com", "P4ssword"),
+            ("http://ignored:detected_pw@example.com", "example.com", "detected_pw"),
+        ]
+    )
+    def test_get_hostname_and_password(self, url: str, expected_host: str, expected_pw: str):
+        returned_host, returned_pw = _get_hostname_and_password(url)
+        assert returned_host == expected_host
+        assert returned_pw == expected_pw
+
+    def test_get_hostname_and_pw_non_http_raises(self):
+        with pytest.raises(ExecutionError, match="URL must start with http://"):
+            _get_hostname_and_password("no_http_protocol")