1# Copyright (C) 2012 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# vim: tabstop=4 shiftwidth=4 softtabstop=4
17
18import logging
19
20from ryu.tests.integrated import tester
21from ryu.ofproto import ofproto_v1_0
22from ryu.ofproto import ether
23from ryu.ofproto import nx_match
24
25LOG = logging.getLogger(__name__)
26
27
28class RunTest(tester.TestFlowBase):
29    """ Test case for add flows of1.0
30    """
31    OFP_VERSIONS = [ofproto_v1_0.OFP_VERSION]
32
33    def __init__(self, *args, **kwargs):
34        super(RunTest, self).__init__(*args, **kwargs)
35        self._verify = []
36
37    def add_action(self, dp, action):
38        rule = nx_match.ClsRule()
39        self.send_flow_mod(
40            dp, rule, 0, dp.ofproto.OFPFC_ADD, 0, 0, None,
41            0xffffffff, None, dp.ofproto.OFPFF_SEND_FLOW_REM, action)
42
43    def add_rule(self, dp, rule):
44        self.send_flow_mod(
45            dp, rule, 0, dp.ofproto.OFPFC_ADD, 0, 0, None,
46            0xffffffff, None, dp.ofproto.OFPFF_SEND_FLOW_REM, [])
47
48    def send_flow_mod(self, dp, rule, cookie, command, idle_timeout,
49                      hard_timeout, priority=None, buffer_id=0xffffffff,
50                      out_port=None, flags=0, actions=None):
51
52        if priority is None:
53            priority = dp.ofproto.OFP_DEFAULT_PRIORITY
54        if out_port is None:
55            out_port = dp.ofproto.OFPP_NONE
56
57        match_tuple = rule.match_tuple()
58        match = dp.ofproto_parser.OFPMatch(*match_tuple)
59
60        m = dp.ofproto_parser.OFPFlowMod(
61            dp, match, cookie, command, idle_timeout, hard_timeout,
62            priority, buffer_id, out_port, flags, actions)
63
64        dp.send_msg(m)
65
66    def _verify_action(self, actions, type_, name, value):
67        try:
68            action = actions[0]
69            if action.cls_action_type != type_:
70                return "Action type error. send:%s, val:%s" \
71                    % (type_, action.cls_action_type)
72        except IndexError:
73            return "Action is not setting."
74
75        f_value = None
76        if name:
77            try:
78                if isinstance(name, list):
79                    f_value = [getattr(action, n) for n in name]
80                else:
81                    f_value = getattr(action, name)
82            except AttributeError:
83                pass
84
85        if f_value != value:
86            return "Value error. send:%s=%s val:%s" \
87                % (name, value, f_value)
88        return True
89
90    def _verify_rule(self, rule, name, value):
91        f_value = getattr(rule, name)
92        if f_value != value:
93            return "Value error. send:%s=%s val:%s" \
94                % (name, value, f_value)
95        return True
96
97    def verify_default(self, dp, stats):
98        verify = self._verify
99        self._verify = []
100        match = stats[0].match
101        actions = stats[0].actions
102
103        if len(verify) == 2:
104            return self._verify_rule(match, *verify)
105        elif len(verify) == 3:
106            return self._verify_action(actions, *verify)
107        else:
108            return "self._verify is invalid."
109
110    # Test of Actions
111    def test_action_output(self, dp):
112        out_port = 2
113        self._verify = [dp.ofproto.OFPAT_OUTPUT,
114                        'port', out_port]
115        action = dp.ofproto_parser.OFPActionOutput(out_port)
116        self.add_action(dp, [action, ])
117
118    def test_action_vlan_vid(self, dp):
119        vlan_vid = 2
120        self._verify = [dp.ofproto.OFPAT_SET_VLAN_VID,
121                        'vlan_vid', vlan_vid]
122        action = dp.ofproto_parser.OFPActionVlanVid(vlan_vid)
123        self.add_action(dp, [action, ])
124
125    def test_action_vlan_pcp(self, dp):
126        vlan_pcp = 4
127        self._verify = [dp.ofproto.OFPAT_SET_VLAN_PCP,
128                        'vlan_pcp', vlan_pcp]
129        action = dp.ofproto_parser.OFPActionVlanPcp(vlan_pcp)
130        self.add_action(dp, [action, ])
131
132    def test_action_strip_vlan(self, dp):
133        vlan_pcp = 4
134        self._verify = [dp.ofproto.OFPAT_STRIP_VLAN,
135                        None, None]
136        action = dp.ofproto_parser.OFPActionStripVlan()
137        self.add_action(dp, [action, ])
138
139    def test_action_set_dl_src(self, dp):
140        dl_src = '56:b3:42:04:b2:7a'
141        dl_src_bin = self.haddr_to_bin(dl_src)
142        self._verify = [dp.ofproto.OFPAT_SET_DL_SRC,
143                        'dl_addr', dl_src_bin]
144        action = dp.ofproto_parser.OFPActionSetDlSrc(dl_src_bin)
145        self.add_action(dp, [action, ])
146
147    def test_action_set_dl_dst(self, dp):
148        dl_dst = 'c2:93:a2:fb:d0:f4'
149        dl_dst_bin = self.haddr_to_bin(dl_dst)
150        self._verify = [dp.ofproto.OFPAT_SET_DL_DST,
151                        'dl_addr', dl_dst_bin]
152        action = dp.ofproto_parser.OFPActionSetDlDst(dl_dst_bin)
153        self.add_action(dp, [action, ])
154
155    def test_action_set_nw_src(self, dp):
156        nw_src = '216.132.81.105'
157        nw_src_int = self.ipv4_to_int(nw_src)
158        self._verify = [dp.ofproto.OFPAT_SET_NW_SRC,
159                        'nw_addr', nw_src_int]
160        action = dp.ofproto_parser.OFPActionSetNwSrc(nw_src_int)
161        self.add_action(dp, [action, ])
162
163    def test_action_set_nw_dst(self, dp):
164        nw_dst = '223.201.206.3'
165        nw_dst_int = self.ipv4_to_int(nw_dst)
166        self._verify = [dp.ofproto.OFPAT_SET_NW_DST,
167                        'nw_addr', nw_dst_int]
168        action = dp.ofproto_parser.OFPActionSetNwDst(nw_dst_int)
169        self.add_action(dp, [action, ])
170
171    def test_action_set_nw_tos(self, dp):
172        # lowest two bits must be zero
173        nw_tos = 1 << 2
174        self._verify = [dp.ofproto.OFPAT_SET_NW_TOS,
175                        'tos', nw_tos]
176        action = dp.ofproto_parser.OFPActionSetNwTos(nw_tos)
177        self.add_action(dp, [action, ])
178
179    def test_action_set_tp_src(self, dp):
180        tp_src = 55420
181        self._verify = [dp.ofproto.OFPAT_SET_TP_SRC,
182                        'tp', tp_src]
183        action = dp.ofproto_parser.OFPActionSetTpSrc(tp_src)
184        self.add_action(dp, [action, ])
185
186    def test_action_set_tp_dst(self, dp):
187        tp_dst = 15430
188        self._verify = [dp.ofproto.OFPAT_SET_TP_DST,
189                        'tp', tp_dst]
190        action = dp.ofproto_parser.OFPActionSetTpDst(tp_dst)
191        self.add_action(dp, [action, ])
192
193    def test_action_enqueue(self, dp):
194        port = 207
195        queue_id = 4287508753
196        self._verify = [dp.ofproto.OFPAT_ENQUEUE,
197                        ['port', 'queue_id'], [port, queue_id]]
198        action = dp.ofproto_parser.OFPActionEnqueue(port, queue_id)
199        self.add_action(dp, [action, ])
200
201    # Test of Rules
202    def test_rule_set_in_port(self, dp):
203        in_port = 32
204        self._verify = ['in_port', in_port]
205
206        rule = nx_match.ClsRule()
207        rule.set_in_port(in_port)
208        self.add_rule(dp, rule)
209
210    def test_rule_set_dl_src(self, dp):
211        dl_src = 'b8:a1:94:51:78:83'
212        dl_src_bin = self.haddr_to_bin(dl_src)
213        self._verify = ['dl_src', dl_src_bin]
214
215        rule = nx_match.ClsRule()
216        rule.set_dl_src(dl_src_bin)
217        self.add_rule(dp, rule)
218
219    def test_rule_set_dl_type_ip(self, dp):
220        dl_type = ether.ETH_TYPE_IP
221        self._verify = ['dl_type', dl_type]
222
223        rule = nx_match.ClsRule()
224        rule.set_dl_type(dl_type)
225        self.add_rule(dp, rule)
226
227    def test_rule_set_dl_type_arp(self, dp):
228        dl_type = ether.ETH_TYPE_ARP
229        self._verify = ['dl_type', dl_type]
230
231        rule = nx_match.ClsRule()
232        rule.set_dl_type(dl_type)
233        self.add_rule(dp, rule)
234
235    def test_rule_set_dl_type_vlan(self, dp):
236        dl_type = ether.ETH_TYPE_8021Q
237        self._verify = ['dl_type', dl_type]
238
239        rule = nx_match.ClsRule()
240        rule.set_dl_type(dl_type)
241        self.add_rule(dp, rule)
242
243    def test_rule_set_dl_type_ipv6(self, dp):
244        dl_type = ether.ETH_TYPE_IPV6
245        self._verify = ['dl_type', dl_type]
246
247        rule = nx_match.ClsRule()
248        rule.set_dl_type(dl_type)
249        self.add_rule(dp, rule)
250
251    def test_rule_set_dl_type_lacp(self, dp):
252        dl_type = ether.ETH_TYPE_SLOW
253        self._verify = ['dl_type', dl_type]
254
255        rule = nx_match.ClsRule()
256        rule.set_dl_type(dl_type)
257        self.add_rule(dp, rule)
258