ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/ofproto/test_ofproto_parser.py

   1 # Copyright (C) 2012 Nippon Telegraph and Telephone Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 # vim: tabstop=4 shiftwidth=4 softtabstop=4
  17
  18 import six
  19
  20 import binascii
  21 import unittest
  22 from nose.tools import *
  23 import struct
  24 from ryu import exception
  25
  26 from ryu.ofproto import ofproto_common, ofproto_parser
  27 from ryu.ofproto import ofproto_v1_0, ofproto_v1_0_parser
  28
  29 import logging
  30 LOG = logging.getLogger(__name__)
  31
  32 if six.PY3:
  33     buffer = bytes
  34
  35
  36 class TestOfproto_Parser(unittest.TestCase):
  37     def setUp(self):
  38         LOG.debug('setUp')
  39         self.bufHello = binascii.unhexlify('0100000800000001')
  40
  41         fr = '010600b0000000020000000000000abc' \
  42             + '00000100010000000000008700000fff' \
  43             + '0002aefa39d2b9177472656d61302d30' \
  44             + '00000000000000000000000000000000' \
  45             + '000000c0000000000000000000000000' \
  46             + 'fffe723f9a764cc87673775f30786162' \
  47             + '63000000000000000000000100000001' \
  48             + '00000082000000000000000000000000' \
  49             + '00012200d6c5a1947472656d61312d30' \
  50             + '00000000000000000000000000000000' \
  51             + '000000c0000000000000000000000000'
  52         self.bufFeaturesReply = binascii.unhexlify(fr)
  53
  54         pi = '010a005200000000000001010040' \
  55             + '00020000000000000002000000000001' \
  56             + '080045000032000000004011f967c0a8' \
  57             + '0001c0a8000200010001001e00000000' \
  58             + '00000000000000000000000000000000' \
  59             + '00000000'
  60         self.bufPacketIn = binascii.unhexlify(pi)
  61
  62     def tearDown(self):
  63         LOG.debug('tearDown')
  64         pass
  65
  66     def testHello(self):
  67         (version,
  68          msg_type,
  69          msg_len,
  70          xid) = ofproto_parser.header(self.bufHello)
  71         eq_(version, 1)
  72         eq_(msg_type, 0)
  73         eq_(msg_len, 8)
  74         eq_(xid, 1)
  75
  76     def testFeaturesReply(self):
  77         (version,
  78          msg_type,
  79          msg_len,
  80          xid) = ofproto_parser.header(self.bufFeaturesReply)
  81
  82         msg = ofproto_parser.msg(self,
  83                                  version,
  84                                  msg_type,
  85                                  msg_len,
  86                                  xid,
  87                                  self.bufFeaturesReply)
  88         LOG.debug(msg)
  89
  90         ok_(isinstance(msg, ofproto_v1_0_parser.OFPSwitchFeatures))
  91         LOG.debug(msg.ports[65534])
  92         ok_(isinstance(msg.ports[1], ofproto_v1_0_parser.OFPPhyPort))
  93         ok_(isinstance(msg.ports[2], ofproto_v1_0_parser.OFPPhyPort))
  94         ok_(isinstance(msg.ports[65534], ofproto_v1_0_parser.OFPPhyPort))
  95
  96     def testPacketIn(self):
  97         (version,
  98          msg_type,
  99          msg_len,
 100          xid) = ofproto_parser.header(self.bufPacketIn)
 101
 102         msg = ofproto_parser.msg(self,
 103                                  version,
 104                                  msg_type,
 105                                  msg_len,
 106                                  xid,
 107                                  self.bufPacketIn)
 108         LOG.debug(msg)
 109         ok_(isinstance(msg, ofproto_v1_0_parser.OFPPacketIn))
 110
 111     @raises(AssertionError)
 112     def test_check_msg_len(self):
 113         (version,
 114          msg_type,
 115          msg_len,
 116          xid) = ofproto_parser.header(self.bufPacketIn)
 117
 118         msg_len = len(self.bufPacketIn) + 1
 119         ofproto_parser.msg(self,
 120                            version,
 121                            msg_type,
 122                            msg_len,
 123                            xid,
 124                            self.bufPacketIn)
 125
 126     @raises(exception.OFPUnknownVersion)
 127     def test_check_msg_parser(self):
 128         (version,
 129          msg_type,
 130          msg_len,
 131          xid) = ofproto_parser.header(self.bufPacketIn)
 132
 133         version = 0xff
 134         ofproto_parser.msg(self,
 135                            version,
 136                            msg_type,
 137                            msg_len,
 138                            xid,
 139                            self.bufPacketIn)
 140
 141
 142 class TestMsgBase(unittest.TestCase):
 143     """ Test case for ofproto_parser.MsgBase
 144     """
 145
 146     def setUp(self):
 147         pass
 148
 149     def tearDown(self):
 150         pass
 151
 152     def test_init(self):
 153         pass
 154
 155     def test_set_xid(self):
 156         xid = 3841413783
 157         c = ofproto_parser.MsgBase(object)
 158         c.set_xid(xid)
 159         eq_(xid, c.xid)
 160
 161     @raises(AssertionError)
 162     def test_set_xid_check_xid(self):
 163         xid = 2160492514
 164         c = ofproto_parser.MsgBase(object)
 165         c.xid = xid
 166         c.set_xid(xid)
 167
 168     def _test_parser(self, msg_type=ofproto_v1_0.OFPT_HELLO):
 169         version = ofproto_v1_0.OFP_VERSION
 170         msg_len = ofproto_v1_0.OFP_HEADER_SIZE
 171         xid = 2183948390
 172         data = b'\x00\x01\x02\x03'
 173
 174         fmt = ofproto_v1_0.OFP_HEADER_PACK_STR
 175         buf = struct.pack(fmt, version, msg_type, msg_len, xid) \
 176             + data
 177
 178         res = ofproto_v1_0_parser.OFPHello.parser(
 179             object, version, msg_type, msg_len, xid, bytearray(buf))
 180
 181         eq_(version, res.version)
 182         eq_(msg_type, res.msg_type)
 183         eq_(msg_len, res.msg_len)
 184         eq_(xid, res.xid)
 185         eq_(buffer(buf), res.buf)
 186
 187         # test __str__()
 188         list_ = ('version', 'msg_type', 'msg_len', 'xid')
 189         check = {}
 190         for s in str(res).rsplit(','):
 191             if '=' in s:
 192                 (k, v,) = s.rsplit('=')
 193                 if k in list_:
 194                     check[k] = v
 195
 196         eq_(hex(ofproto_v1_0.OFP_VERSION), check['version'])
 197         eq_(hex(ofproto_v1_0.OFPT_HELLO), check['msg_type'])
 198         eq_(hex(msg_len), check['msg_len'])
 199         eq_(hex(xid), check['xid'])
 200
 201         return True
 202
 203     def test_parser(self):
 204         ok_(self._test_parser())
 205
 206     @raises(AssertionError)
 207     def test_parser_check_msg_type(self):
 208         self._test_parser(ofproto_v1_0.OFPT_ERROR)
 209
 210     def _test_serialize(self):
 211
 212         class Datapath(object):
 213             ofproto = ofproto_v1_0
 214             ofproto_parser = ofproto_v1_0_parser
 215
 216         c = ofproto_v1_0_parser.OFPHello(Datapath)
 217
 218         c.serialize()
 219         eq_(ofproto_v1_0.OFP_VERSION, c.version)
 220         eq_(ofproto_v1_0.OFPT_HELLO, c.msg_type)
 221         eq_(0, c.xid)
 222
 223         return True
 224
 225     def test_serialize(self):
 226         ok_(self._test_serialize())
 227
 228
 229 class TestMsgStrAttr(unittest.TestCase):
 230     """ Test case for ofproto_parser.msg_str_attr
 231     """
 232
 233     def test_msg_str_attr(self):
 234         class Check(object):
 235             check = 'msg_str_attr_test'
 236
 237         c = Check()
 238         buf = ''
 239
 240         res = ofproto_parser.msg_str_attr(c, buf, ('check',))
 241         str_ = str(res)
 242         str_ = str_.rsplit()
 243         eq_('check', str_[0])
 244         eq_('msg_str_attr_test', str_[1])