← Back to branch summary

~ubuntu-branches/ubuntu/wily/ryu/wily-proposed

« back to all changes in this revision

Viewing changes to ryu/tests/unit/ofproto/test_parser_compat.py

  • Committer: Package Import Robot
  • Author(s): Dariusz Dwornikowski
  • Date: 2014-08-18 16:58:52 UTC
  • Revision ID: package-import@ubuntu.com-20140818165852-i0qck3g5mw7rtxt0
Tags: upstream-3.12
Import upstream version 3.12

Show diffs side-by-side

added added

removed removed

Lines of Context:
ryu/tests/unit/ofproto/test_parser_compat.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
import sys
 
18
import unittest
 
19
from nose.tools import eq_
 
20
from nose.tools import ok_
 
21
 
 
22
from ryu.ofproto import ofproto_v1_2
 
23
from ryu.ofproto import ofproto_v1_3
 
24
from ryu.ofproto import ofproto_v1_2_parser
 
25
from ryu.ofproto import ofproto_v1_3_parser
 
26
 
 
27
from ryu.lib import addrconv
 
28
from struct import unpack
 
29
 
 
30
 
 
31
class Test_Parser_Compat(unittest.TestCase):
 
32
    def __init__(self, methodName):
 
33
        print 'init', methodName
 
34
        super(Test_Parser_Compat, self).__init__(methodName)
 
35
 
 
36
    def setUp(self):
 
37
        pass
 
38
 
 
39
    def tearDown(self):
 
40
        pass
 
41
 
 
42
    def _test(self, name, ofpp):
 
43
        ofp = {
 
44
            ofproto_v1_2_parser: ofproto_v1_2,
 
45
            ofproto_v1_3_parser: ofproto_v1_3,
 
46
        }[ofpp]
 
47
 
 
48
        in_port = 987654321
 
49
        eth_src = 'aa:bb:cc:dd:ee:ff'
 
50
        ipv4_src = '192.0.2.9'
 
51
        ipv6_src = 'fe80::f00b:a4ff:feef:5d8f'
 
52
 
 
53
        old_in_port = in_port
 
54
        old_eth_src = addrconv.mac.text_to_bin(eth_src)
 
55
        old_ipv4_src = unpack('!I', addrconv.ipv4.text_to_bin(ipv4_src))[0]
 
56
        old_ipv6_src = list(unpack('!8H',
 
57
                            addrconv.ipv6.text_to_bin(ipv6_src)))
 
58
 
 
59
        def check(o):
 
60
            check_old(o)
 
61
            check_new(o)
 
62
 
 
63
        def check_old(o):
 
64
            # old api
 
65
            def get_field(m, t):
 
66
                for f in m.fields:
 
67
                    if isinstance(f, t):
 
68
                        return f
 
69
            get_value = lambda m, t: get_field(m, t).value
 
70
 
 
71
            eq_(get_value(o, ofpp.MTInPort), old_in_port)
 
72
            eq_(get_value(o, ofpp.MTEthSrc), old_eth_src)
 
73
            eq_(get_value(o, ofpp.MTIPV4Src), old_ipv4_src)
 
74
            eq_(get_value(o, ofpp.MTIPv6Src), old_ipv6_src)
 
75
 
 
76
        def check_new(o):
 
77
            # new api
 
78
            eq_(o['in_port'], in_port)
 
79
            eq_(o['eth_src'], eth_src)
 
80
            eq_(o['ipv4_src'], ipv4_src)
 
81
            eq_(o['ipv6_src'], ipv6_src)
 
82
 
 
83
        # ensure that old and new api produces the same thing
 
84
 
 
85
        # old api
 
86
        old = ofpp.OFPMatch()
 
87
        old.set_in_port(old_in_port)
 
88
        old.set_dl_src(old_eth_src)
 
89
        old.set_ipv4_src(old_ipv4_src)
 
90
        old.set_ipv6_src(old_ipv6_src)
 
91
 
 
92
        old_buf = bytearray()
 
93
        old.serialize(old_buf, 0)
 
94
 
 
95
        # note: you can't inspect an object composed with the old set_XXX api
 
96
        # before serialize().
 
97
        check_old(old)
 
98
 
 
99
        # another variant of old api; originally it was intended to be
 
100
        # internal but actually used in the field.  eg. LINC l2_switch_v1_3.py
 
101
        old2 = ofpp.OFPMatch()
 
102
        old2.append_field(ofp.OXM_OF_IN_PORT, old_in_port)
 
103
        old2.append_field(ofp.OXM_OF_ETH_SRC, old_eth_src)
 
104
        old2.append_field(ofp.OXM_OF_IPV4_SRC, old_ipv4_src)
 
105
        old2.append_field(ofp.OXM_OF_IPV6_SRC, old_ipv6_src)
 
106
        check_old(old2)
 
107
 
 
108
        old2_buf = bytearray()
 
109
        old2.serialize(old2_buf, 0)
 
110
 
 
111
        # new api
 
112
        new = ofpp.OFPMatch(in_port=in_port, eth_src=eth_src,
 
113
                            ipv4_src=ipv4_src, ipv6_src=ipv6_src)
 
114
        check_new(new)
 
115
 
 
116
        new_buf = bytearray()
 
117
        new.serialize(new_buf, 0)
 
118
        eq_(new_buf, old_buf)
 
119
        eq_(new_buf, old2_buf)
 
120
 
 
121
        old_jsondict = old.to_jsondict()
 
122
        old2_jsondict = old2.to_jsondict()
 
123
        new_jsondict = new.to_jsondict()
 
124
        eq_(new_jsondict, old_jsondict)
 
125
        eq_(new_jsondict, old2_jsondict)
 
126
 
 
127
        eq_(str(new), str(old))
 
128
        eq_(str(new), str(old2))
 
129
 
 
130
        # a parsed object can be inspected by old and new api
 
131
 
 
132
        check(ofpp.OFPMatch.parser(buffer(new_buf), 0))
 
133
        check(ofpp.OFPMatch.from_jsondict(new_jsondict.values()[0]))
 
134
 
 
135
 
 
136
def _add_tests():
 
137
    import new
 
138
    import functools
 
139
    import itertools
 
140
 
 
141
    ofpps = [ofproto_v1_2_parser, ofproto_v1_3_parser]
 
142
    for ofpp in ofpps:
 
143
                        mod = ofpp.__name__.split('.')[-1]
 
144
                        method_name = 'test_' + mod + '_ofpmatch_compat'
 
145
 
 
146
                        def _run(self, name, ofpp):
 
147
                            print ('processing %s ...' % name)
 
148
                            self._test(name, ofpp)
 
149
                        print ('adding %s ...' % method_name)
 
150
                        f = functools.partial(_run, name=method_name,
 
151
                                              ofpp=ofpp)
 
152
                        f.func_name = method_name
 
153
                        f.__name__ = method_name
 
154
                        cls = Test_Parser_Compat
 
155
                        im = new.instancemethod(f, None, cls)
 
156
                        setattr(cls, method_name, im)
 
157
 
 
158
_add_tests()