driver/powerdriver: ykushpower: board_name support

When 6425df9 switched from pykush to
ykushcmd, support for the YKUSH 3 and YKUSH XS was lost. Restore support
for those boards by passing the board_name argument to ykushcmd. The
board_name is determined by listing the attached YKUSH boards by model.

Added tests for YKUSHPowerPort and YKUSHPowerPort.

Signed-off-by: Paul Vittorino <[email protected]>

doc/configuration.rst

@@ -264,18 +264,19 @@ Used by:
 
 YKUSHPowerPort
 ++++++++++++++
-A :any:`YKUSHPowerPort` describes a *YEPKIT YKUSH* USB (HID) switchable USB
-hub.
+A :any:`YKUSHPowerPort` describes a *YEPKIT YKUSH* family USB (HID)
+switchable USB hub.
+Models supported include YKUSH, YKUSH 3, and YKUSH XS.
 
 .. code-block:: yaml
 
    YKUSHPowerPort:
-     serial: 'YK12345'
+     serial: 'Y3N12345'
      index: 1
 
-The example describes port 1 on the YKUSH USB hub with the
-serial ``YK12345``.
-Use ``ykushcmd -l`` to get your serial number.
+The example describes port 1 on the YKUSH 3 USB hub with the
+serial ``Y3N12345``.
+Use ``ykushcmd ykush3 -l`` to get your serial number.
 
 Arguments:
   - serial (str): serial number of the YKUSH hub

labgrid/driver/powerdriver.py

@@ -5,6 +5,7 @@
 
 import attr
 
+from ..exceptions import InvalidConfigError
 from ..factory import target_factory
 from ..protocol import PowerProtocol, DigitalOutputProtocol, ResetProtocol
 from ..resource import NetworkPowerPort
@@ -283,11 +284,28 @@ def __attrs_post_init__(self):
         else:
             self.tool = 'ykushcmd'
 
+    def on_activate(self):
+        # Search for the model of the YKUSH device
+        for model in ["ykush", "ykushxs", "ykush3"]:
+            cmd = [
+                self.tool,
+                model,
+                "-l"
+            ]
+            output = processwrapper.check_output(cmd)
+            if self.port.serial in output.decode("utf-8"):
+                self.model = model
+                break
+        else:
+            raise InvalidConfigError(f"Could not find YKUSH device with serial {self.port.serial}")
+
+
     @Driver.check_active
     @step()
     def on(self):
         cmd = [
             self.tool,
+            self.model,
             "-s", f"{self.port.serial}",
             "-u", f"{self.port.index}"
         ]
@@ -298,6 +316,7 @@ def on(self):
     def off(self):
         cmd = [
             self.tool,
+            self.model,
             "-s", f"{self.port.serial}",
             "-d", f"{self.port.index}"
         ]
@@ -314,6 +333,7 @@ def cycle(self):
     def get(self):
         cmd = [
             self.tool,
+            self.model,
             "-s", f"{self.port.serial}",
             "-g", f"{self.port.index}"
         ]

tests/test_powerdriver.py

@@ -2,8 +2,14 @@
 
 import pytest
 
-from labgrid.resource import NetworkPowerPort
-from labgrid.driver.powerdriver import ExternalPowerDriver, ManualPowerDriver, NetworkPowerDriver
+from labgrid.resource import NetworkPowerPort, YKUSHPowerPort
+from labgrid.driver.powerdriver import (
+    ExternalPowerDriver,
+    ManualPowerDriver,
+    NetworkPowerDriver,
+    YKUSHPowerDriver,
+)
+from labgrid.util.helper import processwrapper
 
 
 class TestManualPowerDriver:
@@ -269,3 +275,73 @@ def test_import_backend_siglent(self):
     def test_import_backend_poe_mib(self):
         pytest.importorskip("pysnmp")
         import labgrid.driver.power.poe_mib
+
+class TestYKUSHPowerDriver:
+    YKUSH_FAKE_SERIAL = "YK12345"
+    YKUSH_LIST_OUTPUT = f"Attached YKUSH Boards:\n1. Board found with serial number: {YKUSH_FAKE_SERIAL}".encode(
+        "utf-8"
+    )
+    YKUSH3_FAKE_SERIAL = "Y3N10673"
+    YKUSH3_LIST_OUTPUT = f"Attached YKUSH3 Boards:\n1. Board found with serial number: {YKUSH3_FAKE_SERIAL}".encode(
+        "utf-8"
+    )
+    YKUSHXS_LIST_OUTPUT = (
+        "Attached YKUSH XS Boards:\n1. Board found with serial number: YKU1234".encode(
+            "utf-8"
+        )
+    )
+
+    def test_create(self, target):
+        resource = YKUSHPowerPort(
+            target, "power", serial=self.YKUSH_FAKE_SERIAL, index=1
+        )
+        device = YKUSHPowerDriver(target, "power")
+        assert isinstance(device, YKUSHPowerDriver)
+
+    def test_default_off(self, target, mocker):
+        check_output_mock = mocker.patch(
+            "labgrid.util.helper.processwrapper.check_output"
+        )
+        check_output_mock.side_effect = [
+            self.YKUSH_LIST_OUTPUT,
+            self.YKUSHXS_LIST_OUTPUT,
+            self.YKUSH3_LIST_OUTPUT,
+            b"",
+        ]
+        resource = YKUSHPowerPort(
+            target, "power", serial=self.YKUSH_FAKE_SERIAL, index=2
+        )
+        resource.avail = True
+        device = YKUSHPowerDriver(target, "power")
+        target.activate(device)
+        device.off()
+
+        check_output_mock.assert_called_with(
+            ["ykushcmd", "ykush", "-s", self.YKUSH_FAKE_SERIAL, "-d", "2"]
+        )
+
+    def test_ykush3_on(self, target, mocker):
+        check_output_mock = mocker.patch(
+            "labgrid.util.helper.processwrapper.check_output"
+        )
+        check_output_mock.side_effect = [
+            self.YKUSH_LIST_OUTPUT,
+            self.YKUSHXS_LIST_OUTPUT,
+            self.YKUSH3_LIST_OUTPUT,
+            b"",
+        ]
+        resource = YKUSHPowerPort(
+            target, "power", serial=self.YKUSH3_FAKE_SERIAL, index=3
+        )
+        resource.avail = True
+        device = YKUSHPowerDriver(target, "power")
+        target.activate(device)
+        device.on()
+
+        check_output_mock.assert_called_with(
+            ["ykushcmd", "ykush3", "-s", self.YKUSH3_FAKE_SERIAL, "-u", "3"]
+        )
+
+    def test_import_backend_poe_mib(self):
+        pytest.importorskip("pysnmp")
+        import labgrid.driver.power.poe_mib