diff --git a/doc/configuration.rst b/doc/configuration.rst index 685aa26b6..17e1d3ceb 100644 --- a/doc/configuration.rst +++ b/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. diff --git a/labgrid/driver/power/sentry_sequential.py b/labgrid/driver/power/sentry_sequential.py new file mode 100644 index 000000000..b504a28af --- /dev/null +++ b/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]}") diff --git a/tests/test_powerdriver.py b/tests/test_powerdriver.py index 2ae8783b2..3d8dae9fb 100644 --- a/tests/test_powerdriver.py +++ b/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