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

   1 # Copyright (C) 2015 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 import logging
  17 import unittest
  18
  19 import ryu.ofproto.ofproto_v1_5 as ofp
  20
  21
  22 LOG = logging.getLogger(__name__)
  23
  24
  25 class Test_OXS(unittest.TestCase):
  26     def _test_encode(self, user, on_wire):
  27         """ test encording user value into on-wire bytes.
  28
  29         n: name of OXS field
  30         uv: user vale
  31         t: oxs_type
  32         v: on-wire bytes value
  33         """
  34         (n, uv) = user
  35         (t, v, _) = ofp.oxs_from_user(n, uv)
  36         buf = bytearray()
  37         ofp.oxs_serialize(t, v, None, buf, 0)
  38         self.assertEqual(on_wire, buf)
  39
  40     def _test_decode(self, user, on_wire):
  41         """ test decording user value from on-wire bytes.
  42
  43         t: oxs_type
  44         v: on-wire bytes value
  45         l: length of field
  46         n: name of OXS field
  47         uv: user vale
  48         """
  49         (t, v, _, l) = ofp.oxs_parse(on_wire, 0)
  50         self.assertEqual(len(on_wire), l)
  51         (n, uv) = ofp.oxs_to_user(t, v, None)
  52         self.assertEqual(user, (n, uv))
  53
  54     def _test_encode_header(self, user, on_wire):
  55         """ test encording header.
  56
  57         t: oxs_type
  58         """
  59         t = ofp.oxs_from_user_header(user)
  60         buf = bytearray()
  61         ofp.oxs_serialize_header(t, buf, 0)
  62         self.assertEqual(on_wire, buf)
  63
  64     def _test_decode_header(self, user, on_wire):
  65         """ test decording header.
  66
  67         t: oxs_type
  68         l: length of header
  69         n: name of OXS field
  70         """
  71         (t, l) = ofp.oxs_parse_header(on_wire, 0)
  72         self.assertEqual(len(on_wire), l)
  73         n = ofp.oxs_to_user_header(t)
  74         self.assertEqual(user, n)
  75
  76     def _test(self, user, on_wire, header_bytes):
  77         """ execute tests.
  78
  79         user: user specified value.
  80               eg. user = ('duration', (100, 100))
  81         on_wire: on-wire bytes
  82         header_bytes: header length
  83         """
  84         self._test_encode(user, on_wire)
  85         self._test_decode(user, on_wire)
  86         user_header = user[0]
  87         on_wire_header = on_wire[:header_bytes]
  88         self._test_decode_header(user_header, on_wire_header)
  89         if user_header.startswith('field_'):
  90             return  # not supported
  91         self._test_encode_header(user_header, on_wire_header)
  92
  93     def test_basic_single(self):
  94         user = ('flow_count', 100)
  95         on_wire = (
  96             b'\x80\x02\x06\x04'
  97             b'\x00\x00\x00\x64'
  98         )
  99         self._test(user, on_wire, 4)
 100
 101     def test_basic_double(self):
 102         user = ('duration', (100, 200))
 103         on_wire = (
 104             b'\x80\x02\x00\x08'
 105             b'\x00\x00\x00\x64'
 106             b'\x00\x00\x00\xc8'
 107         )
 108         self._test(user, on_wire, 4)
 109
 110     def test_basic_unknown(self):
 111         user = ('field_100', 'aG9nZWhvZ2U=')
 112         on_wire = (
 113             b'\x00\x00\xc8\x08'
 114             b'hogehoge'
 115         )
 116         self._test(user, on_wire, 4)