ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/ofproto/test_oxm.py

   1 # Copyright (C) 2015 Nippon Telegraph and Telephone Corporation.
   2 # Copyright (C) 2015 YAMAMOTO Takashi <yamamoto at valinux co jp>
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 import unittest
  18
  19 import ryu.ofproto.ofproto_v1_3 as ofp
  20
  21
  22 class Test_OXM(unittest.TestCase):
  23     def _test_encode(self, user, on_wire):
  24         (f, uv) = user
  25         (n, v, m) = ofp.oxm_from_user(f, uv)
  26         buf = bytearray()
  27         ofp.oxm_serialize(n, v, m, buf, 0)
  28         self.assertEqual(on_wire, buf)
  29
  30     def _test_decode(self, user, on_wire):
  31         (n, v, m, l) = ofp.oxm_parse(on_wire, 0)
  32         self.assertEqual(len(on_wire), l)
  33         (f, uv) = ofp.oxm_to_user(n, v, m)
  34         self.assertEqual(user, (f, uv))
  35
  36     def _test_encode_header(self, user, on_wire):
  37         f = user
  38         n = ofp.oxm_from_user_header(f)
  39         buf = bytearray()
  40         ofp.oxm_serialize_header(n, buf, 0)
  41         self.assertEqual(on_wire, buf)
  42
  43     def _test_decode_header(self, user, on_wire):
  44         (n, l) = ofp.oxm_parse_header(on_wire, 0)
  45         self.assertEqual(len(on_wire), l)
  46         f = ofp.oxm_to_user_header(n)
  47         self.assertEqual(user, f)
  48
  49     def _test(self, user, on_wire, header_bytes):
  50         self._test_encode(user, on_wire)
  51         self._test_decode(user, on_wire)
  52         if isinstance(user[1], tuple):  # has mask?
  53             return
  54         user_header = user[0]
  55         on_wire_header = on_wire[:header_bytes]
  56         self._test_decode_header(user_header, on_wire_header)
  57         if user_header.startswith('field_'):
  58             return  # not supported
  59         self._test_encode_header(user_header, on_wire_header)
  60
  61     def test_basic_nomask(self):
  62         user = ('ipv4_src', '192.0.2.1')
  63         on_wire = (
  64             b'\x80\x00\x16\x04'
  65             b'\xc0\x00\x02\x01'
  66         )
  67         self._test(user, on_wire, 4)
  68
  69     def test_basic_mask(self):
  70         user = ('ipv4_src', ('192.0.2.1', '255.255.0.0'))
  71         on_wire = (
  72             b'\x80\x00\x17\x08'
  73             b'\xc0\x00\x02\x01'
  74             b'\xff\xff\x00\x00'
  75         )
  76         self._test(user, on_wire, 4)
  77
  78     def test_exp_nomask(self):
  79         user = ('_dp_hash', 0x12345678)
  80         on_wire = (
  81             b'\xff\xff\x00\x08'
  82             b'\x00\x00\x23\x20'  # Nicira
  83             b'\x12\x34\x56\x78'
  84         )
  85         self._test(user, on_wire, 8)
  86
  87     def test_exp_mask(self):
  88         user = ('_dp_hash', (0x12345678, 0x7fffffff))
  89         on_wire = (
  90             b'\xff\xff\x01\x0c'
  91             b'\x00\x00\x23\x20'  # Nicira
  92             b'\x12\x34\x56\x78'
  93             b'\x7f\xff\xff\xff'
  94         )
  95         self._test(user, on_wire, 8)
  96
  97     def test_exp_nomask_2(self):
  98         user = ('tcp_flags', 0x876)
  99         on_wire = (
 100             b'\xff\xff\x54\x06'
 101             b'\x4f\x4e\x46\x00'  # ONF
 102             b'\x08\x76'
 103         )
 104         self._test(user, on_wire, 8)
 105
 106     def test_exp_mask_2(self):
 107         user = ('tcp_flags', (0x876, 0x7ff))
 108         on_wire = (
 109             b'\xff\xff\x55\x08'
 110             b'\x4f\x4e\x46\x00'  # ONF
 111             b'\x08\x76'
 112             b'\x07\xff'
 113         )
 114         self._test(user, on_wire, 8)
 115
 116     def test_exp_nomask_3(self):
 117         user = ('actset_output', 0x98765432)
 118         on_wire = (
 119             b'\xff\xff\x56\x08'
 120             b'\x4f\x4e\x46\x00'  # ONF
 121             b'\x98\x76\x54\x32'
 122         )
 123         self._test(user, on_wire, 8)
 124
 125     def test_exp_mask_3(self):
 126         user = ('actset_output', (0x98765432, 0xfffffffe))
 127         on_wire = (
 128             b'\xff\xff\x57\x0c'
 129             b'\x4f\x4e\x46\x00'  # ONF
 130             b'\x98\x76\x54\x32'
 131             b'\xff\xff\xff\xfe'
 132         )
 133         self._test(user, on_wire, 8)
 134
 135     def test_nxm_1_nomask(self):
 136         user = ('tun_ipv4_src', '192.0.2.1')
 137         on_wire = (
 138             b'\x00\x01\x3e\x04'
 139             b'\xc0\x00\x02\x01'
 140         )
 141         self._test(user, on_wire, 4)
 142
 143     def test_nxm_1_mask(self):
 144         user = ('tun_ipv4_src', ('192.0.2.1', '255.255.0.0'))
 145         on_wire = (
 146             b'\x00\x01\x3f\x08'
 147             b'\xc0\x00\x02\x01'
 148             b'\xff\xff\x00\x00'
 149         )
 150         self._test(user, on_wire, 4)
 151
 152     def test_ext_256_nomask(self):
 153         user = ('pbb_uca', 50)
 154         on_wire = (
 155             b'\xff\xff\x00\x07'
 156             b'\x4f\x4e\x46\x00'  # ONF
 157             b'\x0a\x00'
 158             b'\x32'
 159         )
 160         self._test(user, on_wire, 10)
 161
 162     def test_ext_256_mask(self):
 163         user = ('pbb_uca', (50, 51))
 164         on_wire = (
 165             b'\xff\xff\x01\x08'
 166             b'\x4f\x4e\x46\x00'  # ONF
 167             b'\x0a\x00'
 168             b'\x32'
 169             b'\x33'
 170         )
 171         self._test(user, on_wire, 10)
 172
 173     def test_basic_unknown_nomask(self):
 174         user = ('field_100', 'aG9nZWhvZ2U=')
 175         on_wire = (
 176             b'\x00\x00\xc8\x08'
 177             b'hogehoge'
 178         )
 179         self._test(user, on_wire, 4)
 180
 181     def test_basic_unknown_mask(self):
 182         user = ('field_100', ('aG9nZWhvZ2U=', 'ZnVnYWZ1Z2E='))
 183         on_wire = (
 184             b'\x00\x00\xc9\x10'
 185             b'hogehoge'
 186             b'fugafuga'
 187         )
 188         self._test(user, on_wire, 4)