Merge 3bd795d into 0869507

pyipmi/interfaces/ipmb.py

@@ -176,7 +176,7 @@ def rx_filter(header, data):
 
     for left, right, msg in checks:
         if left != right:
-            log().debug('%s: %s %s' % (msg, left, right))
+            log().debug('{:s}: {:s} {:s}'.format(msg, left, right))
             match = False
 
     return match

pyipmi/interfaces/ipmitool.py

@@ -53,6 +53,8 @@ def __init__(self, interface_type='lan'):
         self.re_timeout = re.compile(
                 b"Unable to send RAW command \(.*cmd=0x[0-9a-f]+\)")
 
+        self._session = None
+
     def establish_session(self, session):
         # just remember session parameters here
         self._session = session
@@ -75,7 +77,7 @@ def rmcp_ping(self):
             cmd += (' -P "%s"' % self._session.auth_password)
         cmd += (' session info all')
 
-        output, rc = self._run_ipmitool(cmd)
+        _, rc = self._run_ipmitool(cmd)
         if rc:
             raise IpmiTimeoutError()
 
@@ -93,8 +95,8 @@ def send_and_receive_raw(self, target, lun, netfn, raw_bytes):
         if self._interface_type in ['lan', 'lanplus']:
             cmd = self._build_ipmitool_cmd(target, lun, netfn, raw_bytes)
         elif self._interface_type in ['serial-terminal']:
-            cmd = self._build_serial_ipmitool_cmd(
-                        target, lun, netfn, raw_bytes)
+            cmd = self._build_serial_ipmitool_cmd(target, lun, netfn,
+                                                  raw_bytes)
         else:
             raise RuntimeError('interface type %s not supported' %
                                self._interface_type)
@@ -125,8 +127,8 @@ def send_and_receive_raw(self, target, lun, netfn, raw_bytes):
                                 'Close Session command failed')]
             output = ''.join(output_lines).replace('\r', '').strip()
             if len(output):
-                for x in output.split(' '):
-                    data.append(int(x, 16))
+                for value in output.split(' '):
+                    data.append(int(value, 16))
 
         log().debug('IPMI RX: {:s}'.format(
             ''.join('%02x ' % b for b in array('B', data))))
@@ -152,7 +154,7 @@ def send_and_receive(self, req):
     def _build_ipmitool_raw_data(lun, netfn, raw):
         cmd = ' -l {:d} raw '.format(lun)
         cmd += ' '.join(['0x%02x' % (d)
-                        for d in [netfn] + array('B', raw).tolist()])
+                         for d in [netfn] + array('B', raw).tolist()])
         return cmd
 
     @staticmethod

pyipmi/interfaces/rmcp.py

@@ -45,7 +45,7 @@ def loop():
     return stopped.set
 
 
-class RmcpMsg:
+class RmcpMsg(object):
     RMCP_HEADER_FORMAT = '!BxBB'
     ASF_RMCP_V_1_0 = 6
 
@@ -73,7 +73,7 @@ def unpack(self, pdu):
         return sdu
 
 
-class AsfMsg:
+class AsfMsg(object):
     ASF_HEADER_FORMAT = '!IBBxB'
 
     ASF_TYPE_PRESENCE_PONG = 0x40
@@ -176,7 +176,7 @@ def check_header(self):
             raise DecodingError('Data length mismatch')
 
 
-class IpmiMsg():
+class IpmiMsg(object):
     HEADER_FORMAT_NO_AUTH = '!BIIB'
     HEADER_FORMAT_AUTH = '!BII16BB'
 
@@ -274,7 +274,8 @@ def unpack(self, pdu):
         if len(pdu) < header_len + data_len:
             raise DecodingError('short SDU')
         elif len(pdu) > header_len + data_len:
-            raise DecodingError('SDU has extra bytes')
+            raise DecodingError('SDU has extra bytes ({:d},{:d},{:d} )'.format(
+                len(pdu), header_len, data_len))
 
         if hasattr(self, 'check_header'):
             self.check_header()
@@ -293,7 +294,7 @@ def check_header(self):
         pass
 
 
-class Rmcp:
+class Rmcp(object):
     NAME = 'rmcp'
 
     _session = None

pyipmi/utils.py

@@ -52,7 +52,7 @@ def chunks(data, count):
         yield data[i:i+count]
 
 
-class ByteBuffer:
+class ByteBuffer(object):
     def __init__(self, data=None):
 
         if data is not None:

tests/interfaces/test_ipmitool.py

@@ -19,6 +19,36 @@ def setup(self):
         self.session.set_auth_type_user('admin', 'secret')
         self._interface.establish_session(self.session)
 
+    def test_build_ipmitool_target_ipmb_address(self):
+        target = Target(0xb0)
+        cmd = self._interface._build_ipmitool_target(target)
+        eq_(cmd, ' -t 0xb0')
+
+    def test_build_ipmitool_target_routing_2(self):
+        target = Target(routing=[(0x81, 0x20, 7), (0x20, 0x82, 0)])
+        cmd = self._interface._build_ipmitool_target(target)
+        eq_(cmd, ' -t 0x82 -b 7')
+
+    def test_build_ipmitool_target_routing_3(self):
+        target = Target(routing=[(0x81, 0x20, 0),
+                                 (0x20, 0x82, 7),
+                                 (0x20, 0x72, None)])
+        cmd = self._interface._build_ipmitool_target(target)
+        eq_(cmd, ' -T 0x82 -B 0 -t 0x72 -b 7')
+
+    def test_send_and_receive(self):
+        pass
+
+    def test_rmcp_ping(self):
+        mock = MagicMock()
+        mock.return_value = (b'', 0)
+        self._interface._run_ipmitool = mock
+
+        self._interface.rmcp_ping()
+        mock.assert_called_once_with('ipmitool -I lan -H 10.0.1.1 -p 623 '
+                                     '-U "admin" -P "secret" '
+                                     'session info all')
+
     def test_send_and_receive_raw_valid(self):
         mock = MagicMock()
         mock.return_value = (b'', 0)
@@ -27,7 +57,9 @@ def test_send_and_receive_raw_valid(self):
         target = Target(0x20)
         self._interface.send_and_receive_raw(target, 0, 0x6, b'\x01')
 
-        mock.assert_called_once_with('ipmitool -I lan -H 10.0.1.1 -p 623 -U "admin" -P "secret" -t 0x20 -l 0 raw 0x06 0x01 2>&1')
+        mock.assert_called_once_with('ipmitool -I lan -H 10.0.1.1 -p 623 '
+                                     '-U "admin" -P "secret" -t 0x20 -l 0 '
+                                     'raw 0x06 0x01 2>&1')
 
     def test_send_and_receive_raw_lanplus(self):
         interface = Ipmitool(interface_type='lanplus')
@@ -40,7 +72,9 @@ def test_send_and_receive_raw_lanplus(self):
         target = Target(0x20)
         interface.send_and_receive_raw(target, 0, 0x6, b'\x01')
 
-        mock.assert_called_once_with('ipmitool -I lanplus -H 10.0.1.1 -p 623 -U "admin" -P "secret" -t 0x20 -l 0 raw 0x06 0x01 2>&1')
+        mock.assert_called_once_with('ipmitool -I lanplus -H 10.0.1.1 -p 623 '
+                                     '-U "admin" -P "secret" -t 0x20 -l 0 '
+                                     'raw 0x06 0x01 2>&1')
 
     def test_send_and_receive_raw_no_auth(self):
         mock = MagicMock()
@@ -52,21 +86,26 @@ def test_send_and_receive_raw_no_auth(self):
         target = Target(0x20)
         self._interface.send_and_receive_raw(target, 0, 0x6, b'\x01')
 
-        mock.assert_called_once_with('ipmitool -I lan -H 10.0.1.1 -p 623 -P "" -t 0x20 -l 0 raw 0x06 0x01 2>&1')
+        mock.assert_called_once_with('ipmitool -I lan -H 10.0.1.1 -p 623 '
+                                     '-P "" -t 0x20 -l 0 raw 0x06 0x01 2>&1')
 
     def test_send_and_receive_raw_return_value(self):
         mock = MagicMock()
-        mock.return_value = (b' 10 80 01 02 51 bd 98 3a 00 a8 06 00 03 00 00\n', 0)
+        mock.return_value = (b' 10 80 01 02 51 bd 98 3a 00 a8 '
+                             b'06 00 03 00 00\n', 0)
         self._interface._run_ipmitool = mock
 
         target = Target(0x20)
         data = self._interface.send_and_receive_raw(target, 0, 0x6, b'\x01')
 
-        eq_(data, b'\x00\x10\x80\x01\x02\x51\xbd\x98\x3a\x00\xa8\x06\x00\x03\x00\x00')
+        eq_(data, b'\x00\x10\x80\x01\x02\x51\xbd\x98\x3a\x00\xa8\x06'
+                  b'\x00\x03\x00\x00')
 
     def test_send_and_receive_raw_completion_code_timeout(self):
         mock = MagicMock()
-        mock.return_value = (b'Unable to send RAW command (channel=0x0 netfn=0x6 lun=0x0 cmd=0x1 rsp=0xc3): Ignore Me\n', 1)
+        mock.return_value = (b'Unable to send RAW command (channel=0x0 '
+                             b'netfn=0x6 lun=0x0 cmd=0x1 rsp=0xc3): '
+                             b'Ignore Me\n', 1)
 
         target = Target(0x20)
         self._interface._run_ipmitool = mock
@@ -76,7 +115,9 @@ def test_send_and_receive_raw_completion_code_timeout(self):
 
     def test_send_and_receive_raw_completion_code_not_ok(self):
         mock = MagicMock()
-        mock.return_value = (b'Unable to send RAW command (channel=0x0 netfn=0x6 lun=0x0 cmd=0x1 rsp=0xcc): Ignore Me\n', 1)
+        mock.return_value = (b'Unable to send RAW command (channel=0x0 '
+                             b'netfn=0x6 lun=0x0 cmd=0x1 rsp=0xcc): '
+                             b'Ignore Me\n', 1)
 
         target = Target(0x20)
         self._interface._run_ipmitool = mock
@@ -87,7 +128,8 @@ def test_send_and_receive_raw_completion_code_not_ok(self):
     @raises(IpmiTimeoutError)
     def test_send_and_receive_raw_timeout_without_response(self):
         mock = MagicMock()
-        mock.return_value = (b'Unable to send RAW command (channel=0x0 netfn=0x6 lun=0x0 cmd=0x1)\n', 1)
+        mock.return_value = (b'Unable to send RAW command '
+                             b'(channel=0x0 netfn=0x6 lun=0x0 cmd=0x1)\n', 1)
 
         target = Target(0x20)
         self._interface._run_ipmitool = mock
@@ -105,4 +147,6 @@ def test_send_and_receive_raw_serial(self):
         target = Target(0x20)
         interface.send_and_receive_raw(target, 0, 0x6, b'\x01')
 
-        mock.assert_called_once_with('ipmitool -I serial-terminal -D /dev/tty2:115200 -t 0x20 -l 0 raw 0x06 0x01')
+        mock.assert_called_once_with('ipmitool -I serial-terminal '
+                                     '-D /dev/tty2:115200 -t 0x20 -l 0 '
+                                     'raw 0x06 0x01')

tests/interfaces/test_rmcp.py

@@ -102,19 +102,25 @@ def test_ipmimsg_unpack_auth(self):
                            11, 12, 13, 14, 15, 16])
 
 
-#class TestRmcp:
+class TestRmcp:
 #    def test_send_and_receive_raw(self):
-#        mock = MagicMock()
+#        mock_send = MagicMock()
+#        mock_recv = MagicMock()
+#        mock_recv.return_value = (b'\x06\x00\xee\x07\x00\x00\x00\x00\x00\x00'
+#                                  b'\x00\x00\x00\x06'
+#                                  b'\x01\x02\x03\x04\x05\x06', 0)
+#
 #        target = Target()
 #        target.ipmb_address = 0x20
 #        rmcp = Rmcp()
 #        rmcp.host = '10.10.10.10'
 #        rmcp.port = 637
 #
-#        rmcp._sock = mock
+#        rmcp._sock.sendto = mock_send
+#        rmcp._sock.recvfrom = mock_recv
 #
 #        rmcp.send_and_receive_raw(target, 0, 0, b'\x00')
 #        rmcp._send_ipmi_msg.assert_called_with(1)
-#
-#    def test_send_and_receive(self):
-#        pass
+
+    def test_send_and_receive(self):
+        pass

tests/test_ipmi.py

@@ -20,6 +20,27 @@ def test_target():
     eq_(target.ipmb_address, 0xa0)
     eq_(target.routing, None)
 
+    target = Target(ipmb_address=0xb0)
+    eq_(target.ipmb_address, 0xb0)
+    eq_(target.routing, None)
+
+
+def test_target_routing():
+    target = Target(routing=[(0x82, 0x20, 0)])
+    eq_(len(target.routing), 1)
+    eq_(target.routing[0].rq_sa, 0x82)
+    eq_(target.routing[0].rs_sa, 0x20)
+    eq_(target.routing[0].channel, 0)
+
+    target = Target(routing=[(0x82, 0x20, 0), (0x20, 0x82, 7)])
+    eq_(len(target.routing), 2)
+    eq_(target.routing[0].rq_sa, 0x82)
+    eq_(target.routing[0].rs_sa, 0x20)
+    eq_(target.routing[0].channel, 0)
+    eq_(target.routing[1].rq_sa, 0x20)
+    eq_(target.routing[1].rs_sa, 0x82)
+    eq_(target.routing[1].channel, 7)
+
 
 def test_routing():
     routing = Routing(0x82, 0x20, 7)