interface: ipmb: add rx_filter test

Also introduce IpmbHeaderRsp/Req

Signed-off-by: Heiko Thiery <heiko.thiery@kontron.com>

pyipmi/interfaces/aardvark.py

@@ -20,7 +20,7 @@
 from ..msgs import create_message, encode_message, decode_message
 from ..errors import IpmiTimeoutError
 from ..logger import log
-from ..interfaces.ipmb import IpmbHeader, checksum, rx_filter, encode_ipmb_msg
+from ..interfaces.ipmb import IpmbHeaderReq, checksum, rx_filter, encode_ipmb_msg
 
 try:
     import pyaardvark
@@ -63,7 +63,7 @@ def close_session(self):
         self._dev.close()
 
     def is_ipmc_accessible(self, target):
-        header = IpmbHeader()
+        header = IpmbHeaderReq()
         header.netfn = 6
         header.rs_lun = 0
         header.rs_sa = target.ipmb_address
@@ -136,7 +136,7 @@ def _send_and_receive(self, target, lun, netfn, cmdid, payload):
         self._inc_sequence_number()
 
         # assemble IPMB header
-        header = IpmbHeader()
+        header = IpmbHeaderReq()
         header.netfn = netfn
         header.rs_lun = lun
         header.rs_sa = target.ipmb_address

pyipmi/interfaces/ipmb.py

@@ -30,15 +30,30 @@ def checksum(data):
 
 
 class IpmbHeader(object):
-    def __init__(self):
-        self.rs_sa = None
-        self.rs_lun = None
-        self.rq_sa = None
-        self.rq_lun = None
-        self.rq_seq = None
-        self.netfn = None
-        self.cmd_id = None
+    """Representation of the IPMI message header.
 
+    Request:
+    *-------*--------------*----------*-------*---------------*--------*
+    | rs_sa | netfn/rs_lun | checksum | rq_sa | rq_seq/rq_lun | cmd_id |
+    *-------*--------------*----------*-------*---------------*--------*
+
+    Response:
+    *-------*--------------*----------*-------*---------------*--------*
+    | rq_sa | netfn/rq_lun | checksum | rs_sa | rq_seq/rs_lun | cmd_id |
+    *-------*--------------*----------*-------*---------------*--------*
+
+    """
+    rs_sa = None
+    rs_lun = None
+    rq_sa = None
+    rq_lun = None
+    rq_seq = None
+    netfn = None
+    cmd_id = None
+    checksum = None
+
+
+class IpmbHeaderReq(IpmbHeader):
     def encode(self):
         data = array('B')
         data.append(self.rs_sa)
@@ -49,6 +64,25 @@ def encode(self):
         data.append(self.cmd_id)
         return data.tostring()
 
+    def decode(self):
+        raise NotImplementedError()
+
+
+class IpmbHeaderRsp(IpmbHeader):
+    def encode(self):
+        raise NotImplementedError()
+
+    def decode(self, data):
+        data = array('B', data)
+        self.rq_sa = data[0]
+        self.netfn = data[1] >> 2
+        self.rq_lun = data[1] & 3
+        self.checksum =  data[2]
+        self.rs_sa = data[3]
+        self.rq_seq = data[4] >> 2
+        self.rs_lun = data[4] & 3
+        self.cmd_id = data[5]
+
 
 def encode_ipmb_msg(header, data):
     """Encode an IPMB message.
@@ -86,7 +120,7 @@ def encode_send_message(payload, rq_sa, rs_sa, channel, seq, tracking=1):
     req.channel.tracking = tracking
     data = encode_message(req)
 
-    header = IpmbHeader()
+    header = IpmbHeaderReq()
     header.netfn = req.__netfn__
     header.rs_lun = 0
     header.rs_sa = rs_sa
@@ -158,18 +192,21 @@ def rx_filter(header, data):
     data: the received message as bytestring
     """
 
+    rsp_header = IpmbHeaderRsp()
+    rsp_header.decode(data)
+
     data = array('B', data)
 
     checks = [
         (checksum(data[0:3]), 0, 'Header checksum failed'),
         (checksum(data[3:]), 0, 'payload checksum failed'),
-        # (data[0], header.rq_sa, 'slave address mismatch'),
-        (data[1] & ~3, header.netfn << 2 | 4, 'NetFn mismatch'),
-        # (data[3], header.rs_sa, 'target address mismatch'),
-        # (data[1] & 3, header.rq_lun, 'request LUN mismatch'),
-        (data[4] & 3, header.rs_lun & 3, 'responder LUN mismatch'),
-        (data[4] >> 2, header.rq_seq, 'sequence number mismatch'),
-        (data[5], header.cmd_id, 'command id mismatch'),
+        # rsp_header.rq_sa, header.rq_sa, 'slave address mismatch'),
+        (rsp_header.netfn, header.netfn | 1, 'NetFn mismatch'),
+        # rsp_header.rs_sa, header.rs_sa, 'target address mismatch'),
+        # rsp_header.rq_lun, header.rq_lun, 'request LUN mismatch'),
+        (rsp_header.rs_lun, header.rs_lun, 'responder LUN mismatch'),
+        (rsp_header.rq_seq, header.rq_seq, 'sequence number mismatch'),
+        (rsp_header.cmd_id, header.cmd_id, 'command id mismatch'),
     ]
 
     match = True

pyipmi/interfaces/rmcp.py

@@ -29,7 +29,7 @@
 from ..messaging import ChannelAuthenticationCapabilities
 from ..errors import DecodingError, NotSupportedError
 from ..logger import log
-from ..interfaces.ipmb import (IpmbHeader, encode_ipmb_msg,
+from ..interfaces.ipmb import (IpmbHeaderReq, encode_ipmb_msg,
                                encode_bridged_message, decode_bridged_message,
                                rx_filter)
 from ..utils import check_completion_code
@@ -503,7 +503,7 @@ def _send_and_receive(self, target, lun, netfn, cmdid, payload):
         """
         self._inc_sequence_number()
 
-        header = IpmbHeader()
+        header = IpmbHeaderReq()
         header.netfn = netfn
         header.rs_lun = lun
         header.rs_sa = target.ipmb_address

tests/interfaces/test_ipmb.py

@@ -3,33 +3,44 @@
 
 from array import array
 
-from nose.tools import eq_
+from nose.tools import eq_, ok_
 
 from pyipmi import Target
-from pyipmi.interfaces.ipmb import (checksum, IpmbHeader, encode_send_message,
-                                    encode_bridged_message, encode_ipmb_msg,
-                                    decode_bridged_message)
+from pyipmi.interfaces.ipmb import (checksum, IpmbHeaderReq, IpmbHeaderRsp,
+                                    encode_send_message, encode_bridged_message,
+                                    encode_ipmb_msg, decode_bridged_message,
+                                    rx_filter)
 
 
 def test_checksum():
     eq_(checksum([1, 2, 3, 4, 5]), 256-15)
 
 
 def test_header_encode():
-    header = IpmbHeader()
+    header = IpmbHeaderReq()
     header.rs_lun = 0
     header.rs_sa = 0x72
     header.rq_seq = 2
-    header.rq_lun = 0
+    header.rq_lun = 1
     header.rq_sa = 0x20
     header.netfn = 6
     header.cmd_id = 1
     data = header.encode()
-    eq_(data, b'\x72\x18\x76\x20\x08\x01')
+    eq_(data, b'\x72\x18\x76\x20\x09\x01')
+
+
+def test_header_rsp_decode():
+    header = IpmbHeaderRsp()
+    data = header.decode(b'\x72\x19\x76\x20\x08\x01')
+    eq_(header.rq_sa, 0x72)
+    eq_(header.rq_lun, 1)
+    eq_(header.rs_sa, 0x20)
+    eq_(header.rs_lun, 0)
+    eq_(header.netfn, 6)
 
 
 def test_encode_ipmb_msg():
-    header = IpmbHeader()
+    header = IpmbHeaderReq()
     header.rs_lun = 0
     header.rs_sa = 0x72
     header.rq_seq = 2
@@ -51,7 +62,7 @@ def test_encode_bridged_message():
     payload = array('B', b'\xaa\xbb')
     t = Target(0)
     t.set_routing([(0x81, 0x20, 7), (0x20, 0x72, None)])
-    header = IpmbHeader()
+    header = IpmbHeaderReq()
     header.netfn = 6
     header.rs_lun = 0
     header.rq_seq = 0x11
@@ -71,4 +82,25 @@ def test_decode_bridged_message():
 
 
 def test_rx_filter():
-    pass
+    header_req = IpmbHeaderReq()
+    header_req.rs_lun = 1
+    header_req.rs_sa = 0x72
+    header_req.rq_seq = 2
+    header_req.rq_lun = 0
+    header_req.rq_sa = 0x20
+    header_req.netfn = 6
+    header_req.cmd_id = 1
+
+    # requester and responder fields are twisted ... (sa and lun)
+    header_rsp = IpmbHeaderReq()
+    header_rsp.rs_lun = 0
+    header_rsp.rs_sa = 0x20
+    header_rsp.rq_seq = 2
+    header_rsp.rq_lun = 1
+    header_rsp.rq_sa = 0x72
+    header_rsp.netfn = 6 + 1
+    header_rsp.cmd_id = 1
+
+    rx_data = encode_ipmb_msg(header_rsp, b'\xaa\xbb\xcc')
+
+    ok_(rx_filter(header_req, rx_data))