1 # Copyright (C) 2015 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.
19 import ryu.ofproto.ofproto_v1_5 as ofp
22 LOG = logging.getLogger(__name__)
25 class Test_OXS(unittest.TestCase):
26 def _test_encode(self, user, on_wire):
27 """ test encording user value into on-wire bytes.
32 v: on-wire bytes value
35 (t, v, _) = ofp.oxs_from_user(n, uv)
37 ofp.oxs_serialize(t, v, None, buf, 0)
38 self.assertEqual(on_wire, buf)
40 def _test_decode(self, user, on_wire):
41 """ test decording user value from on-wire bytes.
44 v: on-wire bytes value
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))
54 def _test_encode_header(self, user, on_wire):
55 """ test encording header.
59 t = ofp.oxs_from_user_header(user)
61 ofp.oxs_serialize_header(t, buf, 0)
62 self.assertEqual(on_wire, buf)
64 def _test_decode_header(self, user, on_wire):
65 """ test decording header.
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)
76 def _test(self, user, on_wire, header_bytes):
79 user: user specified value.
80 eg. user = ('duration', (100, 100))
81 on_wire: on-wire bytes
82 header_bytes: header length
84 self._test_encode(user, on_wire)
85 self._test_decode(user, on_wire)
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)
93 def test_basic_single(self):
94 user = ('flow_count', 100)
99 self._test(user, on_wire, 4)
101 def test_basic_double(self):
102 user = ('duration', (100, 200))
108 self._test(user, on_wire, 4)
110 def test_basic_unknown(self):
111 user = ('field_100', 'aG9nZWhvZ2U=')
116 self._test(user, on_wire, 4)