ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/packet/test_dhcp.py

   1 # Copyright (C) 2013 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
  17 import inspect
  18 import logging
  19 import struct
  20 import unittest
  21
  22 import six
  23 from nose.tools import eq_
  24 from nose.tools import ok_
  25
  26 from ryu.lib import addrconv
  27 from ryu.lib.packet import dhcp
  28
  29
  30 LOG = logging.getLogger(__name__)
  31
  32
  33 class Test_dhcp_offer(unittest.TestCase):
  34
  35     op = dhcp.DHCP_BOOT_REPLY
  36     chaddr = 'aa:aa:aa:aa:aa:aa'
  37     htype = 1
  38     hlen = 6
  39     hops = 0
  40     xid = 1
  41     secs = 0
  42     flags = 1
  43     ciaddr = '192.168.10.10'
  44     yiaddr = '192.168.20.20'
  45     siaddr = '192.168.30.30'
  46     giaddr = '192.168.40.40'
  47     sname = 'abc'
  48     boot_file = ''
  49
  50     option_list = [
  51         dhcp.option(dhcp.DHCP_MESSAGE_TYPE_OPT, b'\x02', 1),
  52         dhcp.option(dhcp.DHCP_SUBNET_MASK_OPT, b'\xff\xff\xff\x00', 4),
  53         dhcp.option(dhcp.DHCP_GATEWAY_ADDR_OPT, b'\xc0\xa8\x0a\x09', 4),
  54         dhcp.option(dhcp.DHCP_DNS_SERVER_ADDR_OPT, b'\xc0\xa8\x0a\x09', 4),
  55         dhcp.option(dhcp.DHCP_IP_ADDR_LEASE_TIME_OPT, b'\x00\x03\xf4\x80', 4),
  56         dhcp.option(dhcp.DHCP_RENEWAL_TIME_OPT, b'\x00\x01\xfa\x40', 4),
  57         dhcp.option(dhcp.DHCP_REBINDING_TIME_OPT, b'\x00\x03\x75\xf0', 4),
  58         dhcp.option(dhcp.DHCP_SERVER_IDENTIFIER_OPT, b'\xc0\xa8\x0a\x09', 4)]
  59     magic_cookie = '99.130.83.99'
  60     options = dhcp.options(option_list=option_list, options_len=50,
  61                            magic_cookie=magic_cookie)
  62
  63     dh = dhcp.dhcp(op, chaddr, options, htype=htype, hlen=hlen,
  64                    hops=hops, xid=xid, secs=secs, flags=flags,
  65                    ciaddr=ciaddr, yiaddr=yiaddr, siaddr=siaddr,
  66                    giaddr=giaddr, sname=sname, boot_file=boot_file)
  67
  68     buf = (
  69         b"\x02\x01\x06\x00\x00\x00\x00\x01\x00\x00\x00\x01\xc0\xa8\x0a\x0a"
  70         b"\xc0\xa8\x14\x14\xc0\xa8\x1e\x1e\xc0\xa8\x28\x28\xaa\xaa\xaa\xaa"
  71         b"\xaa\xaa\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x61\x62\x63\x00"
  72         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  73         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  74         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  75         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  76         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  77         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  78         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  79         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  80         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  81         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  82         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
  83         b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x63\x82\x53\x63"
  84         b"\x35\x01\x02\x01\x04\xff\xff\xff\x00\x03\x04\xc0\xa8\x0a\x09\x06"
  85         b"\x04\xc0\xa8\x0a\x09\x33\x04\x00\x03\xf4\x80\x3a\x04\x00\x01\xfa"
  86         b"\x40\x3b\x04\x00\x03\x75\xf0\x36\x04\xc0\xa8\x0a\x09\xff")
  87
  88     def setUp(self):
  89         pass
  90
  91     def tearDown(self):
  92         pass
  93
  94     def test_init(self):
  95         eq_(self.op, self.dh.op)
  96         eq_(self.htype, self.dh.htype)
  97         eq_(self.hlen, self.dh.hlen)
  98         eq_(self.hops, self.dh.hops)
  99         eq_(self.xid, self.dh.xid)
 100         eq_(self.secs, self.dh.secs)
 101         eq_(self.flags, self.dh.flags)
 102         eq_(self.ciaddr, self.dh.ciaddr)
 103         eq_(self.yiaddr, self.dh.yiaddr)
 104         eq_(self.siaddr, self.dh.siaddr)
 105         eq_(self.giaddr, self.dh.giaddr)
 106         eq_(self.chaddr, self.dh.chaddr)
 107         eq_(self.sname, self.dh.sname)
 108         eq_(self.boot_file, self.dh.boot_file)
 109         eq_(str(self.options), str(self.dh.options))
 110
 111     def test_parser(self):
 112         res, _, rest = dhcp.dhcp.parser(self.buf)
 113
 114         eq_(self.op, res.op)
 115         eq_(self.htype, res.htype)
 116         eq_(self.hlen, res.hlen)
 117         eq_(self.hops, res.hops)
 118         eq_(self.xid, res.xid)
 119         eq_(self.secs, res.secs)
 120         eq_(self.flags, res.flags)
 121         eq_(self.ciaddr, res.ciaddr)
 122         eq_(self.yiaddr, res.yiaddr)
 123         eq_(self.siaddr, res.siaddr)
 124         eq_(self.giaddr, res.giaddr)
 125         eq_(self.chaddr, res.chaddr)
 126         # sname is 64 byte length. rest of data is filled by '\x00'.
 127         eq_(self.sname.ljust(64, '\x00'), res.sname)
 128         # boof_file is 128 byte length. rest of data is filled by '\x00'.
 129         eq_(self.boot_file.ljust(128, '\x00'), res.boot_file)
 130         eq_(str(self.options), str(res.options))
 131         eq_(b'', rest)
 132
 133     def test_parser_corrupted(self):
 134         corrupt_buf = self.buf[:-4]
 135         pkt, _, rest = dhcp.dhcp.parser(corrupt_buf)
 136
 137         ok_(isinstance(pkt, dhcp.dhcp))
 138         ok_(isinstance(pkt.options, dhcp.options))
 139         for opt in pkt.options.option_list[:-1]:
 140             ok_(isinstance(opt, dhcp.option))
 141         ok_(isinstance(pkt.options.option_list[-1], six.binary_type))
 142
 143         buf = pkt.serialize()
 144         eq_(str(buf), str(corrupt_buf))
 145         eq_(b'', rest)
 146
 147     def test_serialize(self):
 148         buf = self.dh.serialize()
 149
 150         res = struct.unpack_from(dhcp.dhcp._DHCP_PACK_STR,
 151                                  six.binary_type(buf))
 152
 153         eq_(self.op, res[0])
 154         eq_(self.htype, res[1])
 155         eq_(self.hlen, res[2])
 156         eq_(self.hops, res[3])
 157         eq_(self.xid, res[4])
 158         eq_(self.secs, res[5])
 159         eq_(self.flags, res[6])
 160         eq_(self.ciaddr, addrconv.ipv4.bin_to_text(res[7]))
 161         eq_(self.yiaddr, addrconv.ipv4.bin_to_text(res[8]))
 162         eq_(self.siaddr, addrconv.ipv4.bin_to_text(res[9]))
 163         eq_(self.giaddr, addrconv.ipv4.bin_to_text(res[10]))
 164         eq_(self.chaddr, addrconv.mac.bin_to_text(res[11][:6]))
 165         # sname is 64 byte length. rest of data is filled by '\x00'.
 166         eq_(self.sname.ljust(64, '\x00'), res[12].decode('ascii'))
 167         # boof_file is 128 byte length. rest of data is filled by '\x00'.
 168         eq_(self.boot_file.ljust(128, '\x00'), res[13].decode('ascii'))
 169         options = dhcp.options.parser(
 170             buf[struct.calcsize(dhcp.dhcp._DHCP_PACK_STR):])
 171         eq_(str(self.options), str(options))
 172
 173     def test_to_string(self):
 174         option_values = ['tag', 'length', 'value']
 175         opt_str_list = []
 176         for option in self.option_list:
 177             _opt_str = ','.join(['%s=%s' % (k, repr(getattr(option, k)))
 178                                  for k, v in inspect.getmembers(option)
 179                                  if k in option_values])
 180             opt_str = '%s(%s)' % (dhcp.option.__name__, _opt_str)
 181             opt_str_list.append(opt_str)
 182         option_str = '[%s]' % ', '.join(opt_str_list)
 183
 184         opts_vals = {'magic_cookie': repr(self.magic_cookie),
 185                      'option_list': option_str,
 186                      'options_len': repr(self.options.options_len)}
 187         _options_str = ','.join(['%s=%s' % (k, opts_vals[k])
 188                                  for k, v in inspect.getmembers(self.options)
 189                                  if k in opts_vals])
 190         options_str = '%s(%s)' % (dhcp.options.__name__, _options_str)
 191
 192         dhcp_values = {'op': repr(self.op),
 193                        'htype': repr(self.htype),
 194                        'hlen': repr(self.hlen),
 195                        'hops': repr(self.hops),
 196                        'xid': repr(self.xid),
 197                        'secs': repr(self.secs),
 198                        'flags': repr(self.flags),
 199                        'ciaddr': repr(self.ciaddr),
 200                        'yiaddr': repr(self.yiaddr),
 201                        'siaddr': repr(self.siaddr),
 202                        'giaddr': repr(self.giaddr),
 203                        'chaddr': repr(self.chaddr),
 204                        'sname': repr(self.sname),
 205                        'boot_file': repr(self.boot_file),
 206                        'options': options_str}
 207         _dh_str = ','.join(['%s=%s' % (k, dhcp_values[k])
 208                             for k, v in inspect.getmembers(self.dh)
 209                             if k in dhcp_values])
 210         dh_str = '%s(%s)' % (dhcp.dhcp.__name__, _dh_str)
 211
 212         eq_(str(self.dh), dh_str)
 213         eq_(repr(self.dh), dh_str)
 214
 215     def test_json(self):
 216         jsondict = self.dh.to_jsondict()
 217         dh = dhcp.dhcp.from_jsondict(jsondict['dhcp'])
 218         eq_(str(self.dh), str(dh))