"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/openflow/native/ovs_bridge_test_base.py" (22 Oct 2019, 11511 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "ovs_bridge_test_base.py": 14.0.2_vs_14.0.3.

    1 # Copyright (C) 2014,2015 VA Linux Systems Japan K.K.
    2 # Copyright (C) 2014,2015 YAMAMOTO Takashi <yamamoto at valinux co jp>
    3 # All Rights Reserved.
    4 #
    5 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    6 #    not use this file except in compliance with the License. You may obtain
    7 #    a copy of the License at
    8 #
    9 #         http://www.apache.org/licenses/LICENSE-2.0
   10 #
   11 #    Unless required by applicable law or agreed to in writing, software
   12 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   13 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   14 #    License for the specific language governing permissions and limitations
   15 #    under the License.
   16 
   17 import mock
   18 from oslo_utils import importutils
   19 
   20 from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
   21 from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent \
   22     import ovs_test_base
   23 
   24 
   25 call = mock.call  # short hand
   26 
   27 
   28 class OVSBridgeTestBase(ovs_test_base.OVSOSKenTestBase):
   29     _ARP_MODULE = 'os_ken.lib.packet.arp'
   30     _ETHER_TYPES_MODULE = 'os_ken.lib.packet.ether_types'
   31     _ICMPV6_MODULE = 'os_ken.lib.packet.icmpv6'
   32     _IN_PROTO_MODULE = 'os_ken.lib.packet.in_proto'
   33     _OFP_MODULE = 'os_ken.ofproto.ofproto_v1_3'
   34     _OFPP_MODULE = 'os_ken.ofproto.ofproto_v1_3_parser'
   35 
   36     def setup_bridge_mock(self, name, cls):
   37         self.br = cls(name)
   38         self.stamp = self.br.default_cookie
   39         self.dp = mock.Mock()
   40         self.ofp = importutils.import_module(self._OFP_MODULE)
   41         self.ofpp = importutils.import_module(self._OFPP_MODULE)
   42         self.arp = importutils.import_module(self._ARP_MODULE)
   43         self.ether_types = importutils.import_module(self._ETHER_TYPES_MODULE)
   44         self.icmpv6 = importutils.import_module(self._ICMPV6_MODULE)
   45         self.in_proto = importutils.import_module(self._IN_PROTO_MODULE)
   46         mock.patch.object(self.br, '_get_dp', autospec=True,
   47                           return_value=self._get_dp()).start()
   48         mock__send_msg = mock.patch.object(self.br, '_send_msg').start()
   49         mock_delete_flows = mock.patch.object(self.br,
   50                                               'uninstall_flows').start()
   51         self.mock = mock.Mock()
   52         self.mock.attach_mock(mock__send_msg, '_send_msg')
   53         self.mock.attach_mock(mock_delete_flows, 'uninstall_flows')
   54 
   55     def _get_dp(self):
   56         return self.dp, self.ofp, self.ofpp
   57 
   58     def test_drop_port(self):
   59         in_port = 2345
   60         self.br.drop_port(in_port=in_port)
   61         (dp, ofp, ofpp) = self._get_dp()
   62         expected = [
   63             call._send_msg(
   64                 ofpp.OFPFlowMod(dp,
   65                     cookie=self.stamp,
   66                     instructions=[],
   67                     match=ofpp.OFPMatch(in_port=in_port),
   68                     priority=2,
   69                     table_id=0),
   70                 active_bundle=None),
   71         ]
   72         self.assertEqual(expected, self.mock.mock_calls)
   73 
   74     def test_install_goto(self):
   75         dest_table_id = 123
   76         priority = 99
   77         in_port = 666
   78         self.br.install_goto(dest_table_id=dest_table_id,
   79                              priority=priority, in_port=in_port)
   80         (dp, ofp, ofpp) = self._get_dp()
   81         expected = [
   82             call._send_msg(
   83                 ofpp.OFPFlowMod(dp,
   84                     cookie=self.stamp,
   85                     instructions=[
   86                         ofpp.OFPInstructionGotoTable(table_id=dest_table_id),
   87                     ],
   88                     match=ofpp.OFPMatch(in_port=in_port),
   89                     priority=priority,
   90                     table_id=0),
   91                 active_bundle=None),
   92         ]
   93         self.assertEqual(expected, self.mock.mock_calls)
   94 
   95     def test_install_drop(self):
   96         priority = 99
   97         in_port = 666
   98         self.br.install_drop(priority=priority, in_port=in_port)
   99         (dp, ofp, ofpp) = self._get_dp()
  100         expected = [
  101             call._send_msg(
  102                 ofpp.OFPFlowMod(dp,
  103                     cookie=self.stamp,
  104                     instructions=[],
  105                     match=ofpp.OFPMatch(in_port=in_port),
  106                     priority=priority,
  107                     table_id=0),
  108                 active_bundle=None),
  109         ]
  110         self.assertEqual(expected, self.mock.mock_calls)
  111 
  112     def test_install_normal(self):
  113         priority = 99
  114         in_port = 666
  115         self.br.install_normal(priority=priority, in_port=in_port)
  116         (dp, ofp, ofpp) = self._get_dp()
  117         expected = [
  118             call._send_msg(
  119                 ofpp.OFPFlowMod(dp,
  120                     cookie=self.stamp,
  121                     instructions=[
  122                         ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  123                             ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0)
  124                         ]),
  125                     ],
  126                     match=ofpp.OFPMatch(in_port=in_port),
  127                     priority=priority,
  128                     table_id=0),
  129                 active_bundle=None),
  130         ]
  131         self.assertEqual(expected, self.mock.mock_calls)
  132 
  133     def test__cidr_to_os_ken(self):
  134         f = self.br._cidr_to_os_ken
  135         self.assertEqual('192.168.0.1', f('192.168.0.1'))
  136         self.assertEqual('192.168.0.1', f('192.168.0.1/32'))
  137         self.assertEqual(('192.168.0.0', '255.255.255.0'), f('192.168.0.0/24'))
  138 
  139     def test__setup_controllers__out_of_band(self):
  140         cfg = mock.MagicMock()
  141         cfg.OVS.of_listen_address = ""
  142         cfg.OVS.of_listen_port = ""
  143 
  144         m_add_protocols = mock.patch.object(self.br, 'add_protocols')
  145         m_set_controller = mock.patch.object(self.br, 'set_controller')
  146         m_set_probe = mock.patch.object(self.br,
  147                                         'set_controllers_inactivity_probe')
  148         m_set_ccm = mock.patch.object(self.br,
  149                                       'set_controllers_connection_mode')
  150 
  151         with m_set_ccm as set_ccm:
  152             with m_set_controller, m_add_protocols, m_set_probe:
  153                 self.br.setup_controllers(cfg)
  154                 set_ccm.assert_called_once_with("out-of-band")
  155 
  156 
  157 class OVSDVRProcessTestMixin(object):
  158     def test_install_dvr_process_ipv4(self):
  159         vlan_tag = 999
  160         gateway_ip = '192.0.2.1'
  161         self.br.install_dvr_process_ipv4(vlan_tag=vlan_tag,
  162                                          gateway_ip=gateway_ip)
  163         (dp, ofp, ofpp) = self._get_dp()
  164         expected = [
  165             call._send_msg(ofpp.OFPFlowMod(dp,
  166                 cookie=self.stamp,
  167                 instructions=[],
  168                 match=ofpp.OFPMatch(
  169                     eth_type=self.ether_types.ETH_TYPE_ARP,
  170                     arp_tpa=gateway_ip,
  171                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
  172                 priority=3,
  173                 table_id=constants.FLOOD_TO_TUN),
  174                            active_bundle=None),
  175         ]
  176         self.assertEqual(expected, self.mock.mock_calls)
  177 
  178     def test_delete_dvr_process_ipv4(self):
  179         vlan_tag = 999
  180         gateway_ip = '192.0.2.1'
  181         self.br.delete_dvr_process_ipv4(vlan_tag=vlan_tag,
  182                                         gateway_ip=gateway_ip)
  183         (dp, ofp, ofpp) = self._get_dp()
  184         expected = [
  185             call.uninstall_flows(table_id=constants.FLOOD_TO_TUN,
  186                 match=ofpp.OFPMatch(
  187                     eth_type=self.ether_types.ETH_TYPE_ARP,
  188                     arp_tpa=gateway_ip,
  189                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
  190         ]
  191         self.assertEqual(expected, self.mock.mock_calls)
  192 
  193     def test_install_dvr_process_ipv6(self):
  194         vlan_tag = 999
  195         gateway_mac = '08:60:6e:7f:74:e7'
  196         self.br.install_dvr_process_ipv6(vlan_tag=vlan_tag,
  197                                          gateway_mac=gateway_mac)
  198         (dp, ofp, ofpp) = self._get_dp()
  199         expected = [
  200             call._send_msg(ofpp.OFPFlowMod(dp,
  201                 cookie=self.stamp,
  202                 instructions=[],
  203                 match=ofpp.OFPMatch(
  204                     eth_src=gateway_mac,
  205                     eth_type=self.ether_types.ETH_TYPE_IPV6,
  206                     icmpv6_type=self.icmpv6.ND_ROUTER_ADVERT,
  207                     ip_proto=self.in_proto.IPPROTO_ICMPV6,
  208                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
  209                 priority=3,
  210                 table_id=constants.FLOOD_TO_TUN),
  211                            active_bundle=None),
  212         ]
  213         self.assertEqual(expected, self.mock.mock_calls)
  214 
  215     def test_delete_dvr_process_ipv6(self):
  216         vlan_tag = 999
  217         gateway_mac = '08:60:6e:7f:74:e7'
  218         self.br.delete_dvr_process_ipv6(vlan_tag=vlan_tag,
  219                                         gateway_mac=gateway_mac)
  220         (dp, ofp, ofpp) = self._get_dp()
  221         expected = [
  222             call.uninstall_flows(table_id=constants.FLOOD_TO_TUN,
  223                 match=ofpp.OFPMatch(
  224                     eth_src=gateway_mac,
  225                     eth_type=self.ether_types.ETH_TYPE_IPV6,
  226                     icmpv6_type=self.icmpv6.ND_ROUTER_ADVERT,
  227                     ip_proto=self.in_proto.IPPROTO_ICMPV6,
  228                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
  229         ]
  230         self.assertEqual(expected, self.mock.mock_calls)
  231 
  232     def test_install_dvr_process(self):
  233         vlan_tag = 999
  234         vif_mac = '00:0e:0c:5e:95:d0'
  235         dvr_mac_address = 'f2:0b:a4:5b:b2:ab'
  236         self.br.install_dvr_process(vlan_tag=vlan_tag,
  237                                     vif_mac=vif_mac,
  238                                     dvr_mac_address=dvr_mac_address)
  239         (dp, ofp, ofpp) = self._get_dp()
  240         expected = [
  241             call._send_msg(ofpp.OFPFlowMod(dp,
  242                 cookie=self.stamp,
  243                 instructions=[],
  244                 match=ofpp.OFPMatch(
  245                     eth_dst=vif_mac,
  246                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
  247                 priority=2,
  248                 table_id=self.dvr_process_table_id),
  249                            active_bundle=None),
  250             call._send_msg(ofpp.OFPFlowMod(dp,
  251                 cookie=self.stamp,
  252                 instructions=[
  253                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  254                         ofpp.OFPActionSetField(eth_src=dvr_mac_address),
  255                     ]),
  256                     ofpp.OFPInstructionGotoTable(
  257                         table_id=self.dvr_process_next_table_id),
  258                 ],
  259                 match=ofpp.OFPMatch(
  260                     eth_src=vif_mac,
  261                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
  262                 priority=1,
  263                 table_id=self.dvr_process_table_id),
  264                            active_bundle=None),
  265         ]
  266         self.assertEqual(expected, self.mock.mock_calls)
  267 
  268     def test_delete_dvr_process(self):
  269         vlan_tag = 999
  270         vif_mac = '00:0e:0c:5e:95:d0'
  271         self.br.delete_dvr_process(vlan_tag=vlan_tag,
  272                                    vif_mac=vif_mac)
  273         (dp, ofp, ofpp) = self._get_dp()
  274         expected = [
  275             call.uninstall_flows(table_id=self.dvr_process_table_id,
  276                 match=ofpp.OFPMatch(
  277                     eth_dst=vif_mac,
  278                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
  279             call.uninstall_flows(table_id=self.dvr_process_table_id,
  280                 match=ofpp.OFPMatch(
  281                     eth_src=vif_mac,
  282                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
  283         ]
  284         self.assertEqual(expected, self.mock.mock_calls)