diff --git a/cluster_devices/ipmi.py b/cluster_devices/ipmi.py new file mode 100644 index 0000000..a38f2bd --- /dev/null +++ b/cluster_devices/ipmi.py @@ -0,0 +1,46 @@ +""" +IPMI device. +Just a wrap for ipmitool. +""" + + +from exceptionz import DeviceError +import os +import subprocess + + +class IPMI(object): + CMD = "ipmitool -H %s -I lanplus -U %s -P %s chassis power %s -R 1 -N 1" + SUCCESS_CODE = 0 + + class action: + ON = "on" + OFF = "soft" + REBOOT = "cycle" + + + def __init__(self, id, settings): + self.id = id + self.node_id = settings["node_id"] + self._ip = settings["ip"] + self._login = settings["login"] + self._password = settings["password"] + + + def _perform_cmd(self, action): + cmd_str = IPMI.CMD % (self._ip, self._login, self._password, action) + devnull = open(os.devnull, "w") + result = subprocess.call(cmd_str.split(" "), stdout=devnull, stderr=devnull) + devnull.close() + if (IPMI.SUCCESS_CODE != result): + raise DeviceError(self.id) + + + def on(self): + self._perform_cmd(IPMI.action.ON) + + def off(self): + self._perform_cmd(IPMI.action.OFF) + + def reboot(self): + self._perform_cmd(IPMI.action.REBOOT) diff --git a/tests/ipmi_tests.py b/tests/ipmi_tests.py new file mode 100644 index 0000000..dd0a3b6 --- /dev/null +++ b/tests/ipmi_tests.py @@ -0,0 +1,49 @@ +import sys +sys.path.append("/home/user/_cluster/cntrl_hac") +from exceptionz import DeviceError +from ipmi import IPMI + +import unittest + + +# Working IPMI device. +IPMI_IP = "192.168.50.220" +IPMI_LOGIN = "ADMIN" +IPMI_PASSWORD = "ADMIN" + + +class TestsForIPMI(unittest.TestCase): + def test_correct_settings(self): + ipmi = IPMI("my_ipmi", {"ip": IPMI_IP, + "login": IPMI_LOGIN, + "password": IPMI_PASSWORD}) + ipmi.off() + ipmi.on() + + + def test_wrong_ip(self): + ipmi = IPMI("my_ipmi", {"ip": "127.0.0.1", + "login": IPMI_LOGIN, + "password": IPMI_PASSWORD}) + with self.assertRaises(DeviceError): + ipmi.off() + + + def test_wrong_login(self): + ipmi = IPMI("my_ipmi", {"ip": IPMI_IP, + "login": "OLOLOLOLOLOLOLOLOLOLOLOLOLO", + "password": IPMI_PASSWORD}) + with self.assertRaises(DeviceError): + ipmi.off() + + + def test_wrong_password(self): + ipmi = IPMI("my_ipmi", {"ip": IPMI_IP, + "login": IPMI_LOGIN, + "password": "OLOLOLOLOLOLOLOLOLOLOLOLOLO"}) + with self.assertRaises(DeviceError): + ipmi.off() + + +if (__name__ == '__main__'): + unittest.main()