ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/ofproto/test_parser_ofpstats.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 try:
  17     # Python 3
  18     from functools import reduce
  19 except ImportError:
  20     # Python 2
  21     pass
  22
  23 import six
  24 import sys
  25 import unittest
  26 from nose.tools import eq_
  27 from nose.tools import ok_
  28
  29 from ryu.ofproto import ofproto_v1_5
  30 from ryu.ofproto import ofproto_v1_5_parser
  31 from ryu.tests import test_lib
  32
  33
  34 class Test_Parser_OFPStats(unittest.TestCase):
  35     _ofp = {ofproto_v1_5_parser: ofproto_v1_5}
  36
  37     def __init__(self, methodName):
  38         print('init %s' % methodName)
  39         super(Test_Parser_OFPStats, self).__init__(methodName)
  40
  41     def setUp(self):
  42         pass
  43
  44     def tearDown(self):
  45         pass
  46
  47     def _test(self, name, ofpp, d):
  48         stats = ofpp.OFPStats(**d)
  49         b = bytearray()
  50         stats.serialize(b, 0)
  51         stats2 = stats.parser(six.binary_type(b), 0)
  52         for k, v in d.items():
  53             ok_(k in stats)
  54             ok_(k in stats2)
  55             eq_(stats[k], v)
  56             eq_(stats2[k], v)
  57         for k, v in stats.iteritems():
  58             ok_(k in d)
  59             eq_(d[k], v)
  60         for k, v in stats2.iteritems():
  61             ok_(k in d)
  62             eq_(d[k], v)
  63
  64
  65 def _add_tests():
  66     import functools
  67     import itertools
  68
  69     class Field(object):
  70         pass
  71
  72     class Int1(Field):
  73         @staticmethod
  74         def generate():
  75             yield 0
  76             yield 0xff
  77
  78     class Int2(Field):
  79         @staticmethod
  80         def generate():
  81             yield 0
  82             yield 0x1234
  83             yield 0xffff
  84
  85     class Int3(Field):
  86         @staticmethod
  87         def generate():
  88             yield 0
  89             yield 0x123456
  90             yield 0xffffff
  91
  92     class Int4(Field):
  93         @staticmethod
  94         def generate():
  95             yield 0
  96             yield 0x12345678
  97             yield 0xffffffff
  98
  99     class Int4double(Field):
 100         @staticmethod
 101         def generate():
 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.
 105             yield [0, 1]
 106             yield [0x12345678, 0x23456789]
 107             yield [0xffffffff, 0xfffffffe]
 108
 109     class Int8(Field):
 110         @staticmethod
 111         def generate():
 112             yield 0
 113             yield 0x123456789abcdef0
 114             yield 0xffffffffffffffff
 115
 116     class Mac(Field):
 117         @staticmethod
 118         def generate():
 119             yield '00:00:00:00:00:00'
 120             yield 'f2:0b:a4:7d:f8:ea'
 121             yield 'ff:ff:ff:ff:ff:ff'
 122
 123     class IPv4(Field):
 124         @staticmethod
 125         def generate():
 126             yield '0.0.0.0'
 127             yield '192.0.2.1'
 128             yield '255.255.255.255'
 129
 130     class IPv6(Field):
 131         @staticmethod
 132         def generate():
 133             yield '::'
 134             yield 'fe80::f00b:a4ff:fed0:3f70'
 135             yield 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff'
 136
 137     class B64(Field):
 138         @staticmethod
 139         def generate():
 140             yield 'aG9nZWhvZ2U='
 141             yield 'ZnVnYWZ1Z2E='
 142
 143     ofpps = [ofproto_v1_5_parser]
 144     common = [
 145         ('duration', Int4double),
 146         ('idle_time', Int4double),
 147         ('flow_count', Int4),
 148         ('packet_count', Int8),
 149         ('byte_count', Int8),
 150         ('field_100', B64),
 151     ]
 152     L = {}
 153     L[ofproto_v1_5_parser] = common
 154
 155     def flatten_one(l, i):
 156         if isinstance(i, tuple):
 157             return l + flatten(i)
 158         else:
 159             return l + [i]
 160     flatten = lambda l: reduce(flatten_one, l, [])
 161
 162     for ofpp in ofpps:
 163         for n in range(1, 3):
 164             for C in itertools.combinations(L[ofpp], n):
 165                 l = [1]
 166                 keys = []
 167                 clss = []
 168                 for (k, cls) in C:
 169                     l = itertools.product(l, cls.generate())
 170                     keys.append(k)
 171                     clss.append(cls)
 172                 l = [flatten(x)[1:] for x in l]
 173                 for values in l:
 174                     d = dict(zip(keys, values))
 175                     for n, uv in d.items():
 176                         if isinstance(uv, list):
 177                             # XXX
 178                             # OFPStats returns value as tuple when field is
 179                             # 'duration' or 'idle_time'. Then convert list
 180                             # value into tuple here.
 181                             d[n] = tuple(uv)
 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(' ', '_')
 194
 195                     def _run(self, name, ofpp, d):
 196                         print('processing %s ...' % name)
 197                         if six.PY3:
 198                             self._test(self, name, ofpp, d)
 199                         else:
 200                             self._test(name, ofpp, d)
 201                     print('adding %s ...' % method_name)
 202                     f = functools.partial(_run, name=method_name,
 203                                           ofpp=ofpp, d=d)
 204                     test_lib.add_method(Test_Parser_OFPStats,
 205                                         method_name, f)
 206
 207
 208 _add_tests()