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 *
23 from ryu.ofproto import ether, inet
24 from ryu.lib.packet.packet import Packet
25 from ryu.lib.packet.udp import udp
26 from ryu.lib.packet.ipv4 import ipv4
27 from ryu.lib.packet import packet_utils
28 from ryu.lib import addrconv
31 LOG = logging.getLogger('test_udp')
34 class Test_udp(unittest.TestCase):
41 u = udp(src_port, dst_port, total_length, csum)
42 buf = pack(udp._PACK_STR, src_port, dst_port, total_length, csum)
51 eq_(self.src_port, self.u.src_port)
52 eq_(self.dst_port, self.u.dst_port)
53 eq_(self.total_length, self.u.total_length)
54 eq_(self.csum, self.u.csum)
56 def test_parser(self):
57 r1, r2, _ = self.u.parser(self.buf)
59 eq_(self.src_port, r1.src_port)
60 eq_(self.dst_port, r1.dst_port)
61 eq_(self.total_length, r1.total_length)
62 eq_(self.csum, r1.csum)
65 def test_serialize(self):
71 src_ip = '192.168.10.1'
72 dst_ip = '192.168.100.1'
73 prev = ipv4(4, 5, 0, 0, 0, 0, 0, 64,
74 inet.IPPROTO_UDP, 0, src_ip, dst_ip)
76 u = udp(src_port, dst_port, total_length, csum)
77 buf = u.serialize(bytearray(), prev)
78 res = struct.unpack(udp._PACK_STR, buf)
82 eq_(res[2], struct.calcsize(udp._PACK_STR))
85 ph = struct.pack('!4s4sBBH',
86 addrconv.ipv4.text_to_bin(src_ip),
87 addrconv.ipv4.text_to_bin(dst_ip), 0, 17, res[2])
88 d = ph + buf + bytearray()
89 s = packet_utils.checksum(d)
93 def test_malformed_udp(self):
94 m_short_buf = self.buf[1:udp._MIN_LEN]
95 udp.parser(m_short_buf)
97 def test_default_args(self):
98 prev = ipv4(proto=inet.IPPROTO_UDP)
100 buf = u.serialize(bytearray(), prev)
101 res = struct.unpack(udp._PACK_STR, buf)
105 eq_(res[2], udp._MIN_LEN)
108 jsondict = self.u.to_jsondict()
109 u = udp.from_jsondict(jsondict['udp'])
110 eq_(str(self.u), str(u))