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

   1 # Copyright (C) 2013 Nippon Telegraph and Telephone Corporation.
   2 # Copyright (C) 2013 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 from __future__ import print_function
  18
  19 try:
  20     # Python 3
  21     from functools import reduce
  22 except ImportError:
  23     # Python 2
  24     pass
  25
  26 import six
  27 import unittest
  28 from nose.tools import eq_
  29 from nose.tools import ok_
  30
  31 from ryu.ofproto import ofproto_v1_2
  32 from ryu.ofproto import ofproto_v1_3
  33 from ryu.ofproto import ofproto_v1_4
  34 from ryu.ofproto import ofproto_v1_5
  35 from ryu.ofproto import ofproto_v1_2_parser
  36 from ryu.ofproto import ofproto_v1_3_parser
  37 from ryu.ofproto import ofproto_v1_4_parser
  38 from ryu.ofproto import ofproto_v1_5_parser
  39 from ryu.tests import test_lib
  40
  41
  42 class Test_Parser_OFPMatch(unittest.TestCase):
  43     _ofp = {ofproto_v1_2_parser: ofproto_v1_2,
  44             ofproto_v1_3_parser: ofproto_v1_3,
  45             ofproto_v1_4_parser: ofproto_v1_4,
  46             ofproto_v1_5_parser: ofproto_v1_5}
  47
  48     def __init__(self, methodName):
  49         print('init %s' % methodName)
  50         super(Test_Parser_OFPMatch, self).__init__(methodName)
  51
  52     def setUp(self):
  53         pass
  54
  55     def tearDown(self):
  56         pass
  57
  58     def _test(self, name, ofpp, d, domask):
  59         if domask:
  60             d = dict(self._ofp[ofpp].oxm_normalize_user(k, uv)
  61                      for (k, uv)
  62                      in d.items())
  63         match = ofpp.OFPMatch(**d)
  64         b = bytearray()
  65         match.serialize(b, 0)
  66         match2 = match.parser(six.binary_type(b), 0)
  67         for k, v in d.items():
  68             ok_(k in match)
  69             ok_(k in match2)
  70             eq_(match[k], v)
  71             eq_(match2[k], v)
  72         for k, v in match.iteritems():
  73             ok_(k in d)
  74             eq_(d[k], v)
  75         for k, v in match2.iteritems():
  76             ok_(k in d)
  77             eq_(d[k], v)
  78
  79
  80 def _add_tests():
  81     import functools
  82     import itertools
  83
  84     class Field(object):
  85         @classmethod
  86         def generate_mask(cls):
  87             return list(cls.generate())[1]
  88
  89     class Int1(Field):
  90         @staticmethod
  91         def generate():
  92             yield 0
  93             yield 0xff
  94
  95     class Int2(Field):
  96         @staticmethod
  97         def generate():
  98             yield 0
  99             yield 0x1234
 100             yield 0xffff
 101
 102     class Int3(Field):
 103         @staticmethod
 104         def generate():
 105             yield 0
 106             yield 0x123456
 107             yield 0xffffff
 108
 109     class Int4(Field):
 110         @staticmethod
 111         def generate():
 112             yield 0
 113             yield 0x12345678
 114             yield 0xffffffff
 115
 116     class Int8(Field):
 117         @staticmethod
 118         def generate():
 119             yield 0
 120             yield 0x123456789abcdef0
 121             yield 0xffffffffffffffff
 122
 123     class Mac(Field):
 124         @staticmethod
 125         def generate():
 126             yield '00:00:00:00:00:00'
 127             yield 'f2:0b:a4:7d:f8:ea'
 128             yield 'ff:ff:ff:ff:ff:ff'
 129
 130     class IPv4(Field):
 131         @staticmethod
 132         def generate():
 133             yield '0.0.0.0'
 134             yield '192.0.2.1'
 135             yield '255.255.255.255'
 136
 137     class IPv6(Field):
 138         @staticmethod
 139         def generate():
 140             yield '::'
 141             yield 'fe80::f00b:a4ff:fed0:3f70'
 142             yield 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff'
 143
 144     class B64(Field):
 145         @staticmethod
 146         def generate():
 147             yield 'aG9nZWhvZ2U='
 148             yield 'ZnVnYWZ1Z2E='
 149
 150     ofpps = [ofproto_v1_2_parser, ofproto_v1_3_parser,
 151              ofproto_v1_4_parser, ofproto_v1_5_parser]
 152     common = [
 153         # OpenFlow Basic
 154         ('in_port', Int4),
 155         ('in_phy_port', Int4),
 156         ('metadata', Int8),
 157         ('eth_dst', Mac),
 158         ('eth_src', Mac),
 159         ('eth_type', Int2),
 160         ('vlan_vid', Int2),
 161         ('vlan_pcp', Int1),
 162         ('ip_dscp', Int1),
 163         ('ip_ecn', Int1),
 164         ('ip_proto', Int1),
 165         ('ipv4_src', IPv4),
 166         ('ipv4_dst', IPv4),
 167         ('tcp_src', Int2),
 168         ('tcp_dst', Int2),
 169         ('udp_src', Int2),
 170         ('udp_dst', Int2),
 171         ('sctp_src', Int2),
 172         ('sctp_dst', Int2),
 173         ('icmpv4_type', Int1),
 174         ('icmpv4_code', Int1),
 175         ('arp_op', Int2),
 176         ('arp_spa', IPv4),
 177         ('arp_tpa', IPv4),
 178         ('arp_sha', Mac),
 179         ('arp_tha', Mac),
 180         ('ipv6_src', IPv6),
 181         ('ipv6_dst', IPv6),
 182         ('ipv6_flabel', Int4),
 183         ('icmpv6_type', Int1),
 184         ('icmpv6_code', Int1),
 185         ('ipv6_nd_target', IPv6),
 186         ('ipv6_nd_sll', Mac),
 187         ('ipv6_nd_tll', Mac),
 188         ('mpls_label', Int4),
 189         ('mpls_tc', Int1),
 190         # Old ONF Experimenter --> OpenFlow Basic (OF1.4+)
 191         ('pbb_uca', Int1),
 192         # ONF Experimenter --> OpenFlow Basic (OF1.5+)
 193         ('tcp_flags', Int2),
 194         ('actset_output', Int4),
 195         # Nicira Experimenter
 196         ('eth_dst_nxm', Mac),
 197         ('eth_src_nxm', Mac),
 198         ('tunnel_id_nxm', Int8),
 199         ('tun_ipv4_src', IPv4),
 200         ('tun_ipv4_dst', IPv4),
 201         ('pkt_mark', Int4),
 202         ('conj_id', Int4),
 203         ('tun_ipv6_src', IPv6),
 204         ('tun_ipv6_dst', IPv6),
 205         ('_dp_hash', Int4),
 206         ('reg0', Int4),
 207         ('reg1', Int4),
 208         ('reg2', Int4),
 209         ('reg3', Int4),
 210         ('reg4', Int4),
 211         ('reg5', Int4),
 212         ('reg6', Int4),
 213         ('reg7', Int4),
 214         # Common Experimenter
 215         ('field_100', B64),
 216     ]
 217     L = {}
 218     L[ofproto_v1_2_parser] = common + [
 219         # OF1.2 doesn't have OXM_OF_PBB_ISID.
 220         #    OFPXMC_OPENFLOW_BASIC = 0x8000
 221         #    OXM_OF_PBB_ISID = 37
 222         #    (OFPXMC_OPENFLOW_BASIC << 7) + OXM_OF_PBB_ISID == 4194341
 223         ('field_4194341', B64),
 224     ]
 225     L[ofproto_v1_3_parser] = common + [
 226         # OpenFlow Basic (OF1.3+)
 227         ('mpls_bos', Int1),
 228         ('pbb_isid', Int3),
 229         ('tunnel_id', Int8),
 230         ('ipv6_exthdr', Int2),
 231     ]
 232     L[ofproto_v1_4_parser] = L[ofproto_v1_3_parser]
 233     L[ofproto_v1_5_parser] = L[ofproto_v1_4_parser] + [
 234         # OpenFlow Basic (OF1.5+)
 235         ('packet_type', Int4),
 236     ]
 237
 238     def flatten_one(l, i):
 239         if isinstance(i, tuple):
 240             return l + flatten(i)
 241         else:
 242             return l + [i]
 243     flatten = lambda l: reduce(flatten_one, l, [])
 244
 245     for ofpp in ofpps:
 246         for n in range(1, 3):
 247             for C in itertools.combinations(L[ofpp], n):
 248                 l = [1]
 249                 keys = []
 250                 clss = []
 251                 for (k, cls) in C:
 252                     l = itertools.product(l, cls.generate())
 253                     keys.append(k)
 254                     clss.append(cls)
 255                 l = [flatten(x)[1:] for x in l]
 256                 for domask in [True, False]:
 257                     for values in l:
 258                         if domask:
 259                             values = [(value, cls.generate_mask())
 260                                       for (cls, value)
 261                                       in zip(clss, values)]
 262                         d = dict(zip(keys, values))
 263                         mod = ofpp.__name__.split('.')[-1]
 264                         method_name = 'test_' + mod
 265                         if domask:
 266                             method_name += '_mask'
 267                         for k in sorted(dict(d).keys()):
 268                             method_name += '_' + str(k)
 269                             method_name += '_' + str(d[k])
 270                         method_name = method_name.replace(':', '_')
 271                         method_name = method_name.replace('.', '_')
 272                         method_name = method_name.replace('(', '_')
 273                         method_name = method_name.replace(')', '_')
 274                         method_name = method_name.replace(',', '_')
 275                         method_name = method_name.replace("'", '_')
 276                         method_name = method_name.replace(' ', '_')
 277
 278                         def _run(self, name, ofpp, d, domask):
 279                             print('processing %s ...' % name)
 280                             if six.PY3:
 281                                 self._test(self, name, ofpp, d, domask)
 282                             else:
 283                                 self._test(name, ofpp, d, domask)
 284                         print('adding %s ...' % method_name)
 285                         f = functools.partial(_run, name=method_name,
 286                                               ofpp=ofpp, d=d, domask=domask)
 287                         test_lib.add_method(Test_Parser_OFPMatch,
 288                                             method_name, f)
 289
 290
 291 _add_tests()