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
23 from nose.tools import *
24 from ryu.ofproto import ether, inet
25 from ryu.lib.packet import packet_utils
26 from ryu.lib.packet.ethernet import ethernet
27 from ryu.lib.packet.packet import Packet
28 from ryu.lib.packet.ipv4 import ipv4
29 from ryu.lib.packet.tcp import tcp
30 from ryu.lib import addrconv
33 LOG = logging.getLogger('test_ipv4')
36 class Test_ipv4(unittest.TestCase):
37 """ Test case for ipv4
41 header_length = 5 + 10
42 ver_hlen = version << 4 | header_length
44 total_length = header_length + 64
45 identification = 30774
48 flg_off = flags << 13 | offset
50 proto = inet.IPPROTO_TCP
53 dst = '131.151.32.129'
54 length = header_length * 4
55 option = b'\x86\x28\x00\x00\x00\x01\x01\x22' \
56 + b'\x00\x01\xae\x00\x00\x00\x00\x00' \
57 + b'\x00\x00\x00\x00\x00\x00\x00\x00' \
58 + b'\x00\x00\x00\x00\x00\x00\x00\x00' \
59 + b'\x00\x00\x00\x00\x00\x00\x00\x01'
61 buf = pack(ipv4._PACK_STR, ver_hlen, tos, total_length, identification,
62 flg_off, ttl, proto, csum,
63 addrconv.ipv4.text_to_bin(src),
64 addrconv.ipv4.text_to_bin(dst)) \
67 ip = ipv4(version, header_length, tos, total_length, identification,
68 flags, offset, ttl, proto, csum, src, dst, option)
77 eq_(self.version, self.ip.version)
78 eq_(self.header_length, self.ip.header_length)
79 eq_(self.tos, self.ip.tos)
80 eq_(self.total_length, self.ip.total_length)
81 eq_(self.identification, self.ip.identification)
82 eq_(self.flags, self.ip.flags)
83 eq_(self.offset, self.ip.offset)
84 eq_(self.ttl, self.ip.ttl)
85 eq_(self.proto, self.ip.proto)
86 eq_(self.csum, self.ip.csum)
87 eq_(self.src, self.ip.src)
88 eq_(self.dst, self.ip.dst)
89 eq_(self.length, len(self.ip))
90 eq_(self.option, self.ip.option)
92 def test_parser(self):
93 res, ptype, _ = self.ip.parser(self.buf)
95 eq_(res.version, self.version)
96 eq_(res.header_length, self.header_length)
97 eq_(res.tos, self.tos)
98 eq_(res.total_length, self.total_length)
99 eq_(res.identification, self.identification)
100 eq_(res.flags, self.flags)
101 eq_(res.offset, self.offset)
102 eq_(res.ttl, self.ttl)
103 eq_(res.proto, self.proto)
104 eq_(res.csum, self.csum)
105 eq_(res.src, self.src)
106 eq_(res.dst, self.dst)
109 def test_serialize(self):
110 buf = self.ip.serialize(bytearray(), None)
111 res = struct.unpack_from(ipv4._PACK_STR, six.binary_type(buf))
112 option = buf[ipv4._MIN_LEN:ipv4._MIN_LEN + len(self.option)]
114 eq_(res[0], self.ver_hlen)
115 eq_(res[1], self.tos)
116 eq_(res[2], self.total_length)
117 eq_(res[3], self.identification)
118 eq_(res[4], self.flg_off)
119 eq_(res[5], self.ttl)
120 eq_(res[6], self.proto)
121 eq_(res[8], addrconv.ipv4.text_to_bin(self.src))
122 eq_(res[9], addrconv.ipv4.text_to_bin(self.dst))
123 eq_(option, self.option)
126 csum = packet_utils.checksum(buf)
130 def test_malformed_ipv4(self):
131 m_short_buf = self.buf[1:ipv4._MIN_LEN]
132 ipv4.parser(m_short_buf)
135 jsondict = self.ip.to_jsondict()
136 ip = ipv4.from_jsondict(jsondict['ipv4'])
137 eq_(str(self.ip), str(ip))