ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/packet/test_udp.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 unittest
  19 import logging
  20 import struct
  21 from struct import *
  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
  29
  30
  31 LOG = logging.getLogger('test_udp')
  32
  33
  34 class Test_udp(unittest.TestCase):
  35     """ Test case for udp
  36     """
  37     src_port = 6431
  38     dst_port = 8080
  39     total_length = 65507
  40     csum = 12345
  41     u = udp(src_port, dst_port, total_length, csum)
  42     buf = pack(udp._PACK_STR, src_port, dst_port, total_length, csum)
  43
  44     def setUp(self):
  45         pass
  46
  47     def tearDown(self):
  48         pass
  49
  50     def test_init(self):
  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)
  55
  56     def test_parser(self):
  57         r1, r2, _ = self.u.parser(self.buf)
  58
  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)
  63         eq_(None, r2)
  64
  65     def test_serialize(self):
  66         src_port = 6431
  67         dst_port = 8080
  68         total_length = 0
  69         csum = 0
  70
  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)
  75
  76         u = udp(src_port, dst_port, total_length, csum)
  77         buf = u.serialize(bytearray(), prev)
  78         res = struct.unpack(udp._PACK_STR, buf)
  79
  80         eq_(res[0], src_port)
  81         eq_(res[1], dst_port)
  82         eq_(res[2], struct.calcsize(udp._PACK_STR))
  83
  84         # checksum
  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)
  90         eq_(0, s)
  91
  92     @raises(Exception)
  93     def test_malformed_udp(self):
  94         m_short_buf = self.buf[1:udp._MIN_LEN]
  95         udp.parser(m_short_buf)
  96
  97     def test_default_args(self):
  98         prev = ipv4(proto=inet.IPPROTO_UDP)
  99         u = udp()
 100         buf = u.serialize(bytearray(), prev)
 101         res = struct.unpack(udp._PACK_STR, buf)
 102
 103         eq_(res[0], 1)
 104         eq_(res[1], 1)
 105         eq_(res[2], udp._MIN_LEN)
 106
 107     def test_json(self):
 108         jsondict = self.u.to_jsondict()
 109         u = udp.from_jsondict(jsondict['udp'])
 110         eq_(str(self.u), str(u))