Now PDU and Router classes can raise exceptions + tests for PDU class.

ololobster.py

@@ -33,8 +33,8 @@ def get_router(self, router_id):
 
 ololobster = Ololobster()
 ololobster.update()
-pdu_1 = ololobster.get_pdu("pdu_1")
-hp2530_1 = ololobster.get_router("hp2530_1")
+pdu_2 = ololobster.get_pdu("pdu_2")
+hp2530_2 = ololobster.get_router("hp2530_2")
 
 #hp2530_1.enable_port(pdu_1)
-#pdu_1.on(hp2530_1)
+pdu_2.on(hp2530_2)

pdu.py

@@ -1,8 +1,10 @@
+from exceptionz import DeviceError
 import netsnmp
 
 
 class PDU(object):
     CMD = ".1.3.6.1.4.1.318.1.1.12.3.3.1.1.4.%i"
+    SUCCESS_CODE = 1
     class action(object):
         ON = 1
         OFF = 2
@@ -22,35 +24,27 @@ def __init__(self, id, settings):
 
     def _perform_cmd(self, action, outlet):
         cmd = netsnmp.Varbind(PDU.CMD % (outlet), val=action, type="INTEGER")
-        netsnmp.snmpset(cmd, Version=1, DestHost=self._ip, Community="private")
+        result = netsnmp.snmpset(cmd, Version=1, DestHost=self._ip, Community="private")
+        if (PDU.SUCCESS_CODE != result):
+            raise DeviceError(self.id)
 
 
     def off(self, device):
         outlet = self._outlets.get(device.id, None)
-        assert(outlet is not None)
+        if (outlet is None):
+            raise DeviceError(self.id)
         self._perform_cmd(PDU.action.OFF, outlet)
 
 
     def on(self, device):
         outlet = self._outlets.get(device.id, None)
-        assert(outlet is not None)
+        if (outlet is None):
+            raise DeviceError(self.id)
         self._perform_cmd(PDU.action.ON, outlet)
 
 
     def reboot(self, device):
         outlet = self._outlets.get(device.id, None)
-        assert(outlet is not None)
+        if (outlet is None):
+            raise DeviceError(self.id)
         self._perform_cmd(PDU.action.REBOOT, outlet)
-
-
-class Device(object):
-    def __init__(self, id):
-        self.id = id
-
-
-#kvg1 = Device("kvg1")
-#pdu1 = PDU("pdu1", "192.168.50.111", {"kvg1": 1})
-#pdu2 = PDU("pdu2", "192.168.50.110", {"kvg1": 1})
-
-#pdu1.on(kvg1)
-#pdu2.on(kvg1)

router.py

@@ -1,8 +1,10 @@
+from exceptionz import DeviceError
 import netsnmp
 
 
 class Router(object):
     CMD = ".1.3.6.1.2.1.2.2.1.7.%i"
+    SUCCESS_CODE = 1
     class action(object):
         ENABLE = 1
         DISABLE = 2
@@ -21,26 +23,20 @@ def __init__(self, id, settings):
 
     def _perform_cmd(self, action, port):
         cmd = netsnmp.Varbind(Router.CMD % (port), val=action, type="INTEGER")
-        netsnmp.snmpset(cmd, Version=1, DestHost=self._ip, Community="private")
+        result = netsnmp.snmpset(cmd, Version=1, DestHost=self._ip, Community="private")
+        if (Router.SUCCESS_CODE != result):
+            raise DeviceError(self.id)
 
 
     def disable_port(self, device):
         port = self._ports.get(device.id, None)
-        assert(port is not None)
+        if (port is None):
+            raise DeviceError(self.id)
         self._perform_cmd(Router.action.DISABLE, port)
 
 
     def enable_port(self, device):
         port = self._ports.get(device.id, None)
-        assert(port is not None)
+        if (port is None):
+            raise DeviceError(self.id)
         self._perform_cmd(Router.action.ENABLE, port)
-
-
-class Device(object):
-    def __init__(self, id):
-        self.id = id
-
-
-#kvg1 = Device("kvg1")
-#router = Router("router", "192.168.50.174", {"kvg1": 20})
-#router.enable_port(kvg1)

tests/pdu_tests.py

@@ -0,0 +1,40 @@
+import sys
+sys.path.append("/home/user/_cluster/cntrl_hac")
+sys.path.append("/home/user/_cluster/testutil_hac")
+from cluster import Cluster
+from exceptionz import DeviceError
+from pdu import PDU
+
+import unittest
+
+
+# Working PDU.
+PDU_IP = "192.168.50.110"
+# Some device connected to PDU.
+DUMMY_DEVICE = PDU("dummy", {"ip": None})
+DUMMY_DEVICE_OUTLET = 8
+
+
+class TestsForPDU(unittest.TestCase):
+    def test_correct_settings(self):
+        pdu = PDU("my_pdu", {"ip": PDU_IP,
+                             "outlet%i" % (DUMMY_DEVICE_OUTLET): DUMMY_DEVICE.id})
+        pdu.off(DUMMY_DEVICE)
+        pdu.on(DUMMY_DEVICE)
+
+
+    def test_wrong_ip(self):
+        pdu = PDU("my_pdu", {"ip": "127.0.0.1",
+                             "outlet%i" % (DUMMY_DEVICE_OUTLET): DUMMY_DEVICE.id})
+        with self.assertRaises(DeviceError):
+            pdu.off(DUMMY_DEVICE)
+
+
+    def test_wrong_outlets_config(self):
+        pdu = PDU("my_pdu", {"ip": PDU_IP})
+        with self.assertRaises(DeviceError):
+            pdu.off(DUMMY_DEVICE)
+
+
+if (__name__ == '__main__'):
+    unittest.main()