ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/services/protocols/bgp/utils/test_bgp.py

   1 # Copyright (C) 2017 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
  17 import logging
  18 import unittest
  19
  20 from nose.tools import eq_, raises
  21
  22 from ryu.lib.packet.bgp import (
  23     BGPFlowSpecTrafficRateCommunity,
  24     BGPFlowSpecTrafficActionCommunity,
  25     BGPFlowSpecRedirectCommunity,
  26     BGPFlowSpecTrafficMarkingCommunity,
  27     BGPFlowSpecVlanActionCommunity,
  28     BGPFlowSpecTPIDActionCommunity,
  29 )
  30
  31 from ryu.services.protocols.bgp.core import BgpCoreError
  32 from ryu.services.protocols.bgp.utils.bgp import create_v4flowspec_actions
  33 from ryu.services.protocols.bgp.utils.bgp import create_v6flowspec_actions
  34 from ryu.services.protocols.bgp.utils.bgp import create_l2vpnflowspec_actions
  35
  36
  37 LOG = logging.getLogger(__name__)
  38
  39
  40 class Test_Utils_BGP(unittest.TestCase):
  41     """
  42     Test case for ryu.services.protocols.bgp.utils.bgp
  43     """
  44
  45     def _test_create_v4flowspec_actions(self, actions, expected_communities):
  46         communities = create_v4flowspec_actions(actions)
  47         expected_communities.sort(key=lambda x: x.subtype)
  48         communities.sort(key=lambda x: x.subtype)
  49         eq_(str(expected_communities), str(communities))
  50
  51     def test_create_v4flowspec_actions_all_actions(self):
  52         actions = {
  53             'traffic_rate': {
  54                 'as_number': 0,
  55                 'rate_info': 100.0,
  56             },
  57             'traffic_action': {
  58                 'action': 3,
  59             },
  60             'redirect': {
  61                 'as_number': 10,
  62                 'local_administrator': 10,
  63             },
  64             'traffic_marking': {
  65                 'dscp': 24,
  66             }
  67         }
  68         expected_communities = [
  69             BGPFlowSpecTrafficRateCommunity(as_number=0, rate_info=100.0),
  70             BGPFlowSpecTrafficActionCommunity(action=3),
  71             BGPFlowSpecRedirectCommunity(as_number=10, local_administrator=10),
  72             BGPFlowSpecTrafficMarkingCommunity(dscp=24),
  73         ]
  74         self._test_create_v4flowspec_actions(actions, expected_communities)
  75
  76     def test_create_v4flowspec_actions_without_actions(self):
  77         actions = None
  78         expected_communities = []
  79         self._test_create_v4flowspec_actions(actions, expected_communities)
  80
  81     @raises(ValueError)
  82     def test_create_v4flowspec_actions_not_exist_actions(self):
  83         actions = {
  84             'traffic_test': {
  85                 'test': 10,
  86             },
  87         }
  88         expected_communities = []
  89         self._test_create_v4flowspec_actions(actions, expected_communities)
  90
  91     def _test_create_v6flowspec_actions(self, actions, expected_communities):
  92         communities = create_v6flowspec_actions(actions)
  93         expected_communities.sort(key=lambda x: x.subtype)
  94         communities.sort(key=lambda x: x.subtype)
  95         eq_(str(expected_communities), str(communities))
  96
  97     def test_create_v6flowspec_actions_all_actions(self):
  98         actions = {
  99             'traffic_rate': {
 100                 'as_number': 0,
 101                 'rate_info': 100.0,
 102             },
 103             'traffic_action': {
 104                 'action': 3,
 105             },
 106             'redirect': {
 107                 'as_number': 10,
 108                 'local_administrator': 10,
 109             },
 110             'traffic_marking': {
 111                 'dscp': 24,
 112             }
 113         }
 114         expected_communities = [
 115             BGPFlowSpecTrafficRateCommunity(as_number=0, rate_info=100.0),
 116             BGPFlowSpecTrafficActionCommunity(action=3),
 117             BGPFlowSpecRedirectCommunity(as_number=10, local_administrator=10),
 118             BGPFlowSpecTrafficMarkingCommunity(dscp=24),
 119         ]
 120         self._test_create_v6flowspec_actions(actions, expected_communities)
 121
 122     def test_create_v6flowspec_actions_without_actions(self):
 123         actions = None
 124         expected_communities = []
 125         self._test_create_v6flowspec_actions(actions, expected_communities)
 126
 127     @raises(ValueError)
 128     def test_create_v6flowspec_actions_not_exist_actions(self):
 129         actions = {
 130             'traffic_test': {
 131                 'test': 10,
 132             },
 133         }
 134         expected_communities = []
 135         self._test_create_v6flowspec_actions(actions, expected_communities)
 136
 137     def _test_create_l2vpnflowspec_actions(self, actions, expected_communities):
 138         communities = create_l2vpnflowspec_actions(actions)
 139         expected_communities.sort(key=lambda x: x.subtype)
 140         communities.sort(key=lambda x: x.subtype)
 141         eq_(str(expected_communities), str(communities))
 142
 143     def test_create_l2vpnflowspec_actions_all_actions(self):
 144         actions = {
 145             'traffic_rate': {
 146                 'as_number': 0,
 147                 'rate_info': 100.0,
 148             },
 149             'traffic_action': {
 150                 'action': 3,
 151             },
 152             'redirect': {
 153                 'as_number': 10,
 154                 'local_administrator': 10,
 155             },
 156             'traffic_marking': {
 157                 'dscp': 24,
 158             },
 159             'vlan_action': {
 160                 'actions_1': (BGPFlowSpecVlanActionCommunity.POP |
 161                               BGPFlowSpecVlanActionCommunity.SWAP),
 162                 'vlan_1': 3000,
 163                 'cos_1': 3,
 164                 'actions_2': BGPFlowSpecVlanActionCommunity.PUSH,
 165                 'vlan_2': 4000,
 166                 'cos_2': 2,
 167             },
 168             'tpid_action': {
 169                 'actions': (BGPFlowSpecTPIDActionCommunity.TI |
 170                             BGPFlowSpecTPIDActionCommunity.TO),
 171                 'tpid_1': 5,
 172                 'tpid_2': 6,
 173             }
 174         }
 175         expected_communities = [
 176             BGPFlowSpecTrafficRateCommunity(as_number=0, rate_info=100.0),
 177             BGPFlowSpecTrafficActionCommunity(action=3),
 178             BGPFlowSpecRedirectCommunity(as_number=10, local_administrator=10),
 179             BGPFlowSpecTrafficMarkingCommunity(dscp=24),
 180             BGPFlowSpecVlanActionCommunity(
 181                 actions_1=(BGPFlowSpecVlanActionCommunity.POP |
 182                            BGPFlowSpecVlanActionCommunity.SWAP),
 183                 vlan_1=3000,
 184                 cos_1=3,
 185                 actions_2=BGPFlowSpecVlanActionCommunity.PUSH,
 186                 vlan_2=4000,
 187                 cos_2=2,
 188             ),
 189             BGPFlowSpecTPIDActionCommunity(
 190                 actions=(BGPFlowSpecTPIDActionCommunity.TI |
 191                          BGPFlowSpecTPIDActionCommunity.TO),
 192                 tpid_1=5,
 193                 tpid_2=6,
 194             ),
 195         ]
 196         self._test_create_l2vpnflowspec_actions(actions, expected_communities)
 197
 198     def test_create_l2vpnflowspec_actions_without_actions(self):
 199         actions = None
 200         expected_communities = []
 201         self._test_create_l2vpnflowspec_actions(actions, expected_communities)
 202
 203     @raises(ValueError)
 204     def test_create_l2vpnflowspec_actions_not_exist_actions(self):
 205         actions = {
 206             'traffic_test': {
 207                 'test': 10,
 208             },
 209         }
 210         expected_communities = []
 211         self._test_create_l2vpnflowspec_actions(actions, expected_communities)