driver/power: Add sentry_sequential backend for single-bank models

doc/configuration.rst

@@ -222,8 +222,15 @@ Currently available are:
   for details.
 
 ``sentry``
-  Controls *Sentry PDUs* via SNMP using Sentry3-MIB.
-  It was tested on *CW-24VDD* and *4805-XLS-16*.
+  Controls *Sentry PDUs* via SNMP using Sentry3-MIB with multi-bank OID mapping.
+  Supports up to 48 outlets organized as 6 banks of 8 outlets each.
+  Tested on *CW-24VDD* and *4805-XLS-16*.
+  For single-bank sequential models like *CW-16V1*, use ``sentry_sequential`` instead.
+
+``sentry_sequential``
+  Controls *Sentry PDUs* via SNMP using Sentry3-MIB with sequential OID mapping.
+  Supports up to 16 outlets numbered sequentially (1.1.1 through 1.1.16).
+  Suitable for single-bank models like *CW-16V1*.
 
 ``shelly_gen1``
   Controls relays of *Shelly* devices using the Gen 1 Device API.

labgrid/driver/power/sentry_sequential.py

@@ -0,0 +1,60 @@
+"""
+Sentry PDU driver with sequential OID mapping for single-bank models.
+
+This driver was tested on CW-16V1 but should work on all devices
+implementing Sentry3-MIB with sequential OID numbering.
+
+This driver uses sequential OID mapping with 16 outlets numbered
+1.1.1 through 1.1.16. For multi-bank models like CW-24VDD or
+4805-XLS-16, use 'sentry' instead.
+"""
+
+from ..exception import ExecutionError
+from ...util.helper import processwrapper
+
+INDEX_TO_OID = {
+    1: "1.1.1", 2: "1.1.2", 3: "1.1.3", 4: "1.1.4",
+    5: "1.1.5", 6: "1.1.6", 7: "1.1.7", 8: "1.1.8",
+    9: "1.1.9", 10: "1.1.10", 11: "1.1.11", 12: "1.1.12",
+    13: "1.1.13", 14: "1.1.14", 15: "1.1.15", 16: "1.1.16",
+}
+
+BASE_STATUS_OID = ".1.3.6.1.4.1.1718.3.2.3.1.10"
+BASE_CTRL_OID = ".1.3.6.1.4.1.1718.3.2.3.1.11"
+
+def _snmp_get(host, oid):
+    out = processwrapper.check_output(
+        f"snmpget -v1 -c private -O qn {host} {oid}".split()
+    ).decode('ascii')
+    out_oid, value = out.strip().split(' ', 1)
+    assert oid == out_oid
+    if value == "3" or value == "5":
+        return True
+    if value == "4":
+        return False
+
+def _snmp_set(host, oid, value):
+    try:
+        processwrapper.check_output(
+            f"snmpset -v1 -c private {host} {oid} {value}".split()
+        )
+    except Exception as e:
+        raise ExecutionError("failed to set SNMP value") from e
+
+def power_set(host, port, index, value):
+    assert port is None
+
+    index = int(index)
+    value = 1 if value else 2
+    assert 1 <= index <= 16
+
+    _snmp_set(host, f"{BASE_CTRL_OID}.{INDEX_TO_OID[index]}", f"int {value}")
+
+
+def power_get(host, port, index):
+    assert port is None
+
+    index = int(index)
+    assert 1 <= index <= 16
+
+    return _snmp_get(host, f"{BASE_STATUS_OID}.{INDEX_TO_OID[index]}")

tests/test_powerdriver.py

@@ -287,6 +287,7 @@ def test_import_backends(self):
         import labgrid.driver.power.netio_kshell
         import labgrid.driver.power.rest
         import labgrid.driver.power.sentry
+        import labgrid.driver.power.sentry_sequential
         import labgrid.driver.power.eg_pms2_network
         import labgrid.driver.power.shelly_gen1
         import labgrid.driver.power.ubus
@@ -307,3 +308,134 @@ def test_import_backend_siglent(self):
     def test_import_backend_poe_mib(self):
         pytest.importorskip("pysnmp")
         import labgrid.driver.power.poe_mib
+
+
+class TestSentrySequentialBackend:
+    def test_power_get_on_value_3(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+        mock_output.return_value = b'.1.3.6.1.4.1.1718.3.2.3.1.10.1.1.5 3'
+
+        result = sentry_sequential.power_get('host', None, 5)
+        assert result is True
+
+    def test_power_get_on_value_5(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+        mock_output.return_value = b'.1.3.6.1.4.1.1718.3.2.3.1.10.1.1.5 5'
+
+        result = sentry_sequential.power_get('host', None, 5)
+        assert result is True
+
+    def test_power_get_off(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+        mock_output.return_value = b'.1.3.6.1.4.1.1718.3.2.3.1.10.1.1.5 4'
+
+        result = sentry_sequential.power_get('host', None, 5)
+        assert result is False
+
+    def test_power_set_on(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+
+        sentry_sequential.power_set('host', None, 5, True)
+
+        mock_output.assert_called_once()
+        call_args = mock_output.call_args[0][0]
+        assert 'int' in call_args
+        assert '1' in call_args
+
+    def test_power_set_off(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+
+        sentry_sequential.power_set('host', None, 5, False)
+
+        call_args = mock_output.call_args[0][0]
+        assert 'int' in call_args
+        assert '2' in call_args
+
+    def test_power_set_index_bounds(self):
+        from labgrid.driver.power import sentry_sequential
+
+        with pytest.raises(AssertionError):
+            sentry_sequential.power_set('host', None, 0, True)
+
+        with pytest.raises(AssertionError):
+            sentry_sequential.power_set('host', None, 17, True)
+
+    def test_power_get_index_bounds(self):
+        from labgrid.driver.power import sentry_sequential
+
+        with pytest.raises(AssertionError):
+            sentry_sequential.power_get('host', None, 0)
+
+        with pytest.raises(AssertionError):
+            sentry_sequential.power_get('host', None, 17)
+
+    def test_power_set_port_must_be_none(self):
+        from labgrid.driver.power import sentry_sequential
+
+        with pytest.raises(AssertionError):
+            sentry_sequential.power_set('host', 'port', 5, True)
+
+    def test_power_get_port_must_be_none(self):
+        from labgrid.driver.power import sentry_sequential
+
+        with pytest.raises(AssertionError):
+            sentry_sequential.power_get('host', 'port', 5)
+
+    def test_power_set_error_handling(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+        from labgrid.driver.exception import ExecutionError
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+        mock_output.side_effect = Exception("SNMP failed")
+
+        with pytest.raises(ExecutionError):
+            sentry_sequential.power_set('host', None, 5, True)
+
+    def test_power_get_oid_mapping(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+        mock_output.return_value = b'.1.3.6.1.4.1.1718.3.2.3.1.10.1.1.1 3'
+
+        sentry_sequential.power_get('host', None, 1)
+
+        call_args = mock_output.call_args[0][0]
+        # Verify OID ends with .1.1.1 for index 1
+        assert '.1.3.6.1.4.1.1718.3.2.3.1.10.1.1.1' in call_args
+
+    def test_power_set_oid_mapping(self, mocker):
+        from labgrid.driver.power import sentry_sequential
+
+        mock_output = mocker.patch(
+            'labgrid.driver.power.sentry_sequential.processwrapper.check_output'
+        )
+
+        sentry_sequential.power_set('host', None, 16, True)
+
+        call_args = mock_output.call_args[0][0]
+        # Verify OID ends with .1.1.16 for index 16
+        assert '.1.3.6.1.4.1.1718.3.2.3.1.11.1.1.16' in call_args