1 # Copyright (C) 2012 Nippon Telegraph and Telephone Corporation.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 # vim: tabstop=4 shiftwidth=4 softtabstop=4
22 from nose.tools import *
24 from ryu import exception
26 from ryu.ofproto import ofproto_common, ofproto_parser
27 from ryu.ofproto import ofproto_v1_0, ofproto_v1_0_parser
30 LOG = logging.getLogger(__name__)
36 class TestOfproto_Parser(unittest.TestCase):
39 self.bufHello = binascii.unhexlify('0100000800000001')
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)
54 pi = '010a005200000000000001010040' \
55 + '00020000000000000002000000000001' \
56 + '080045000032000000004011f967c0a8' \
57 + '0001c0a8000200010001001e00000000' \
58 + '00000000000000000000000000000000' \
60 self.bufPacketIn = binascii.unhexlify(pi)
70 xid) = ofproto_parser.header(self.bufHello)
76 def testFeaturesReply(self):
80 xid) = ofproto_parser.header(self.bufFeaturesReply)
82 msg = ofproto_parser.msg(self,
87 self.bufFeaturesReply)
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))
96 def testPacketIn(self):
100 xid) = ofproto_parser.header(self.bufPacketIn)
102 msg = ofproto_parser.msg(self,
109 ok_(isinstance(msg, ofproto_v1_0_parser.OFPPacketIn))
111 @raises(AssertionError)
112 def test_check_msg_len(self):
116 xid) = ofproto_parser.header(self.bufPacketIn)
118 msg_len = len(self.bufPacketIn) + 1
119 ofproto_parser.msg(self,
126 @raises(exception.OFPUnknownVersion)
127 def test_check_msg_parser(self):
131 xid) = ofproto_parser.header(self.bufPacketIn)
134 ofproto_parser.msg(self,
142 class TestMsgBase(unittest.TestCase):
143 """ Test case for ofproto_parser.MsgBase
155 def test_set_xid(self):
157 c = ofproto_parser.MsgBase(object)
161 @raises(AssertionError)
162 def test_set_xid_check_xid(self):
164 c = ofproto_parser.MsgBase(object)
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
172 data = b'\x00\x01\x02\x03'
174 fmt = ofproto_v1_0.OFP_HEADER_PACK_STR
175 buf = struct.pack(fmt, version, msg_type, msg_len, xid) \
178 res = ofproto_v1_0_parser.OFPHello.parser(
179 object, version, msg_type, msg_len, xid, bytearray(buf))
181 eq_(version, res.version)
182 eq_(msg_type, res.msg_type)
183 eq_(msg_len, res.msg_len)
185 eq_(buffer(buf), res.buf)
188 list_ = ('version', 'msg_type', 'msg_len', 'xid')
190 for s in str(res).rsplit(','):
192 (k, v,) = s.rsplit('=')
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'])
203 def test_parser(self):
204 ok_(self._test_parser())
206 @raises(AssertionError)
207 def test_parser_check_msg_type(self):
208 self._test_parser(ofproto_v1_0.OFPT_ERROR)
210 def _test_serialize(self):
212 class Datapath(object):
213 ofproto = ofproto_v1_0
214 ofproto_parser = ofproto_v1_0_parser
216 c = ofproto_v1_0_parser.OFPHello(Datapath)
219 eq_(ofproto_v1_0.OFP_VERSION, c.version)
220 eq_(ofproto_v1_0.OFPT_HELLO, c.msg_type)
225 def test_serialize(self):
226 ok_(self._test_serialize())
229 class TestMsgStrAttr(unittest.TestCase):
230 """ Test case for ofproto_parser.msg_str_attr
233 def test_msg_str_attr(self):
235 check = 'msg_str_attr_test'
240 res = ofproto_parser.msg_str_attr(c, buf, ('check',))
243 eq_('check', str_[0])
244 eq_('msg_str_attr_test', str_[1])