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.
18 from functools import reduce
26 from nose.tools import eq_
27 from nose.tools import ok_
29 from ryu.ofproto import ofproto_v1_5
30 from ryu.ofproto import ofproto_v1_5_parser
31 from ryu.tests import test_lib
34 class Test_Parser_OFPStats(unittest.TestCase):
35 _ofp = {ofproto_v1_5_parser: ofproto_v1_5}
37 def __init__(self, methodName):
38 print('init %s' % methodName)
39 super(Test_Parser_OFPStats, self).__init__(methodName)
47 def _test(self, name, ofpp, d):
48 stats = ofpp.OFPStats(**d)
51 stats2 = stats.parser(six.binary_type(b), 0)
52 for k, v in d.items():
57 for k, v in stats.iteritems():
60 for k, v in stats2.iteritems():
99 class Int4double(Field):
102 # Note: If yield value as a tuple, flatten_one() will reduce it
103 # into a single value. So the followings avoid this problem by
104 # using a list value.
106 yield [0x12345678, 0x23456789]
107 yield [0xffffffff, 0xfffffffe]
113 yield 0x123456789abcdef0
114 yield 0xffffffffffffffff
119 yield '00:00:00:00:00:00'
120 yield 'f2:0b:a4:7d:f8:ea'
121 yield 'ff:ff:ff:ff:ff:ff'
128 yield '255.255.255.255'
134 yield 'fe80::f00b:a4ff:fed0:3f70'
135 yield 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff'
143 ofpps = [ofproto_v1_5_parser]
145 ('duration', Int4double),
146 ('idle_time', Int4double),
147 ('flow_count', Int4),
148 ('packet_count', Int8),
149 ('byte_count', Int8),
153 L[ofproto_v1_5_parser] = common
155 def flatten_one(l, i):
156 if isinstance(i, tuple):
157 return l + flatten(i)
160 flatten = lambda l: reduce(flatten_one, l, [])
163 for n in range(1, 3):
164 for C in itertools.combinations(L[ofpp], n):
169 l = itertools.product(l, cls.generate())
172 l = [flatten(x)[1:] for x in l]
174 d = dict(zip(keys, values))
175 for n, uv in d.items():
176 if isinstance(uv, list):
178 # OFPStats returns value as tuple when field is
179 # 'duration' or 'idle_time'. Then convert list
180 # value into tuple here.
182 mod = ofpp.__name__.split('.')[-1]
183 method_name = 'test_' + mod
184 for k in sorted(dict(d).keys()):
185 method_name += '_' + str(k)
186 method_name += '_' + str(d[k])
187 method_name = method_name.replace(':', '_')
188 method_name = method_name.replace('.', '_')
189 method_name = method_name.replace('(', '_')
190 method_name = method_name.replace(')', '_')
191 method_name = method_name.replace(',', '_')
192 method_name = method_name.replace("'", '_')
193 method_name = method_name.replace(' ', '_')
195 def _run(self, name, ofpp, d):
196 print('processing %s ...' % name)
198 self._test(self, name, ofpp, d)
200 self._test(name, ofpp, d)
201 print('adding %s ...' % method_name)
202 f = functools.partial(_run, name=method_name,
204 test_lib.add_method(Test_Parser_OFPStats,