From aae45b67e5cbece451149c6c445726cff360b546 Mon Sep 17 00:00:00 2001 From: Paul Spooren Date: Tue, 21 May 2024 17:08:50 +0300 Subject: [PATCH] power: add ubus NetworkPowerPort Using PoE switches as power supply is convenient since a switch is needed in many cases anyway and with the right adapter most embedded devices can be powered. OpenWrt offers `ubus` as micro bus system to control the system, including PoE ports on switches. This commit adds a driver to handle PoE switches running OpenWrt. An ACL example is given below: ```shell root@switch:~# cat /usr/share/rpcd/acl.d/unauthenticated.json { "unauthenticated": { "description": "Access controls for unauthenticated requests", "read": { "ubus": { "session": [ "access", "login" ], "poe": [ "info", "manage" ] } } } } ``` Signed-off-by: Paul Spooren --- doc/configuration.rst | 4 +++ labgrid/driver/power/ubus.py | 48 ++++++++++++++++++++++++++++++++++++ tests/test_powerdriver.py | 40 ++++++++++++++++++++++++++++-- 3 files changed, 90 insertions(+), 2 deletions(-) create mode 100644 labgrid/driver/power/ubus.py diff --git a/doc/configuration.rst b/doc/configuration.rst index a4c83b040..67e687eb5 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -234,6 +234,10 @@ Currently available are: ``poe_mib`` Controls PoE switches using the PoE SNMP administration MiBs. +``ubus`` + Controls *PoE switches* running OpenWrt using the *ubus* interface. + Further infromation available at + Used by: - `NetworkPowerDriver`_ diff --git a/labgrid/driver/power/ubus.py b/labgrid/driver/power/ubus.py new file mode 100644 index 000000000..3efcc268f --- /dev/null +++ b/labgrid/driver/power/ubus.py @@ -0,0 +1,48 @@ +""" + UBUS jsonrpc interface for PoE management on OpenWrt devices. This comes in + handy if devices are connected to a PoE switch running OpenWrt. + + The URL given in hosts in exporter.yaml must accept unauthenticated UBUS + calls for the two `poe` calls `info` and `manage`. + + Further information is availbe at https://openwrt.org/docs/techref/ubus#acls + + NetworkPowerPort: + model: ubus + host: 'http://192.168.1.1/ubus' + index: 1 +""" + +import requests + + +def jsonrpc_call(host, path, method, message): + r = requests.post( + host, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "call", + "params": ["00000000000000000000000000000000", path, method, message], + }, + ) + r.raise_for_status() + return r.json()["result"] + + +def power_set(host, port, index, value): + assert port is None + + jsonrpc_call(host, "poe", "manage", {"port": f"lan{index}", "enable": value == 1}) + + +def power_get(host, port, index): + assert port is None + + poe_info = jsonrpc_call(host, "poe", "info", {})[1] + + assert ( + f"lan{index}" in poe_info["ports"] + ), f"Port lan{index} not found in {poe_info['ports']}" + + return poe_info["ports"][f"lan{index}"] != "Disabled" diff --git a/tests/test_powerdriver.py b/tests/test_powerdriver.py index 37d6c2af7..657b76e41 100644 --- a/tests/test_powerdriver.py +++ b/tests/test_powerdriver.py @@ -234,11 +234,46 @@ def test_create_shelly_gen1_backend_with_url_in_host(self, target, mocker, host) expected_host = f"{host}/relay/{index}" url = urlparse(expected_host) if url.port is None: - implicit_port = 443 if url.scheme == 'https' else 80 - expected_host = expected_host.replace(url.netloc, f'{url.netloc}:{implicit_port}') + implicit_port = 443 if url.scheme == "https" else 80 + expected_host = expected_host.replace( + url.netloc, f"{url.netloc}:{implicit_port}" + ) get.assert_called_with(expected_host) + def test_create_ubus_backend(self, target, mocker): + post = mocker.patch("requests.post") + post.return_value.json.return_value = { + "jsonrpc": "2.0", + "id": 1, + "result": [ + 0, + { + "firmware": "v80.1", + "budget": 77.000000, + "consumption": 1.700000, + "ports": { + "lan1": { + "priority": 0, + "mode": "PoE+", + "status": "Delivering power", + "consumption": 1.700000, + } + }, + }, + ], + } + + index = "1" + NetworkPowerPort( + target, "power", model="ubus", host="http://example.com/ubus", index=index + ) + d = NetworkPowerDriver(target, "power") + target.activate(d) + + d.cycle() + assert d.get() is True + def test_import_backends(self): import labgrid.driver.power import labgrid.driver.power.apc @@ -253,6 +288,7 @@ def test_import_backends(self): import labgrid.driver.power.sentry import labgrid.driver.power.eg_pms2_network import labgrid.driver.power.shelly_gen1 + import labgrid.driver.power.ubus def test_import_backend_eaton(self): pytest.importorskip("pysnmp")