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.

Signed-off-by: Paul Spooren <mail@aparcar.org>

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 <https://openwrt.org/docs/techref/ubus>
+
 Used by:
   - `NetworkPowerDriver`_
 

labgrid/driver/power/ubus.py

@@ -0,0 +1,71 @@
+"""
+  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`.
+
+  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"
+                  ]
+              }
+          }
+      }
+  }
+  ```
+
+  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"

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