IPMI device will live here from now on.

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)

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