neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/openflow/native/test_br_int.py

   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
  19 from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.openflow.native \
  20     import ovs_bridge_test_base
  21
  22
  23 call = mock.call  # short hand
  24
  25
  26 class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
  27     def setUp(self):
  28         super(OVSIntegrationBridgeTest, self).setUp()
  29         self.setup_bridge_mock('br-int', self.br_int_cls)
  30
  31     def test_setup_default_table(self):
  32         self.br.setup_default_table()
  33         (dp, ofp, ofpp) = self._get_dp()
  34         expected = [
  35             call._send_msg(ofpp.OFPFlowMod(dp,
  36                 cookie=0,
  37                 instructions=[
  38                     ofpp.OFPInstructionActions(
  39                         ofp.OFPIT_APPLY_ACTIONS, [
  40                             ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0)
  41                         ]),
  42                 ],
  43                 match=ofpp.OFPMatch(),
  44                 priority=0,
  45                 table_id=0)),
  46             call._send_msg(ofpp.OFPFlowMod(dp,
  47                 cookie=0,
  48                 instructions=[],
  49                 match=ofpp.OFPMatch(),
  50                 priority=0,
  51                 table_id=23)),
  52             call._send_msg(ofpp.OFPFlowMod(dp,
  53                 cookie=0,
  54                 instructions=[],
  55                 match=ofpp.OFPMatch(),
  56                 priority=0,
  57                 table_id=24)),
  58         ]
  59         self.assertEqual(expected, self.mock.mock_calls)
  60
  61     def test_provision_local_vlan(self):
  62         port = 999
  63         lvid = 888
  64         segmentation_id = 777
  65         self.br.provision_local_vlan(port=port, lvid=lvid,
  66                                      segmentation_id=segmentation_id)
  67         (dp, ofp, ofpp) = self._get_dp()
  68         expected = [
  69             call._send_msg(ofpp.OFPFlowMod(dp,
  70                 cookie=0,
  71                 instructions=[
  72                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  73                         ofpp.OFPActionSetField(
  74                             vlan_vid=lvid | ofp.OFPVID_PRESENT),
  75                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0)
  76                     ]),
  77                 ],
  78                 match=ofpp.OFPMatch(
  79                     in_port=port,
  80                     vlan_vid=segmentation_id | ofp.OFPVID_PRESENT),
  81                 priority=3,
  82                 table_id=0)),
  83         ]
  84         self.assertEqual(expected, self.mock.mock_calls)
  85
  86     def test_provision_local_vlan_novlan(self):
  87         port = 999
  88         lvid = 888
  89         segmentation_id = None
  90         self.br.provision_local_vlan(port=port, lvid=lvid,
  91                                      segmentation_id=segmentation_id)
  92         (dp, ofp, ofpp) = self._get_dp()
  93         expected = [
  94             call._send_msg(ofpp.OFPFlowMod(dp,
  95                 cookie=0,
  96                 instructions=[
  97                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  98                         ofpp.OFPActionPushVlan(),
  99                         ofpp.OFPActionSetField(
 100                             vlan_vid=lvid | ofp.OFPVID_PRESENT),
 101                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
 102                     ]),
 103                 ],
 104                 match=ofpp.OFPMatch(
 105                     in_port=port,
 106                     vlan_vid=ofp.OFPVID_NONE),
 107                 priority=3,
 108                 table_id=0)),
 109         ]
 110         self.assertEqual(expected, self.mock.mock_calls)
 111
 112     def test_reclaim_local_vlan(self):
 113         port = 999
 114         segmentation_id = 777
 115         self.br.reclaim_local_vlan(port=port, segmentation_id=segmentation_id)
 116         (dp, ofp, ofpp) = self._get_dp()
 117         expected = [
 118             call.delete_flows(
 119                 match=ofpp.OFPMatch(
 120                     in_port=port,
 121                     vlan_vid=segmentation_id | ofp.OFPVID_PRESENT)),
 122         ]
 123         self.assertEqual(expected, self.mock.mock_calls)
 124
 125     def test_reclaim_local_vlan_novlan(self):
 126         port = 999
 127         segmentation_id = None
 128         self.br.reclaim_local_vlan(port=port, segmentation_id=segmentation_id)
 129         (dp, ofp, ofpp) = self._get_dp()
 130         expected = [
 131             call.delete_flows(
 132                 match=ofpp.OFPMatch(
 133                     in_port=port,
 134                     vlan_vid=ofp.OFPVID_NONE)),
 135         ]
 136         self.assertEqual(expected, self.mock.mock_calls)
 137
 138     def test_install_dvr_to_src_mac(self):
 139         network_type = 'vxlan'
 140         vlan_tag = 1111
 141         gateway_mac = '08:60:6e:7f:74:e7'
 142         dst_mac = '00:02:b3:13:fe:3d'
 143         dst_port = 6666
 144         self.br.install_dvr_to_src_mac(network_type=network_type,
 145                                        vlan_tag=vlan_tag,
 146                                        gateway_mac=gateway_mac,
 147                                        dst_mac=dst_mac,
 148                                        dst_port=dst_port)
 149         (dp, ofp, ofpp) = self._get_dp()
 150         expected = [
 151             call._send_msg(ofpp.OFPFlowMod(dp,
 152                 cookie=0,
 153                 instructions=[
 154                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 155                         ofpp.OFPActionPopVlan(),
 156                         ofpp.OFPActionSetField(eth_src=gateway_mac),
 157                         ofpp.OFPActionOutput(6666, 0),
 158                     ]),
 159                 ],
 160                 match=ofpp.OFPMatch(
 161                     eth_dst=dst_mac,
 162                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
 163                 priority=4,
 164                 table_id=1)),
 165         ]
 166         self.assertEqual(expected, self.mock.mock_calls)
 167
 168     def test_delete_dvr_to_src_mac(self):
 169         network_type = 'vxlan'
 170         vlan_tag = 1111
 171         dst_mac = '00:02:b3:13:fe:3d'
 172         self.br.delete_dvr_to_src_mac(network_type=network_type,
 173                                       vlan_tag=vlan_tag,
 174                                       dst_mac=dst_mac)
 175         (dp, ofp, ofpp) = self._get_dp()
 176         expected = [
 177             call.delete_flows(table_id=1,
 178                 match=ofpp.OFPMatch(
 179                     eth_dst=dst_mac,
 180                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
 181         ]
 182         self.assertEqual(expected, self.mock.mock_calls)
 183
 184     def test_install_dvr_to_src_mac_vlan(self):
 185         network_type = 'vlan'
 186         vlan_tag = 1111
 187         gateway_mac = '08:60:6e:7f:74:e7'
 188         dst_mac = '00:02:b3:13:fe:3d'
 189         dst_port = 6666
 190         self.br.install_dvr_to_src_mac(network_type=network_type,
 191                                        vlan_tag=vlan_tag,
 192                                        gateway_mac=gateway_mac,
 193                                        dst_mac=dst_mac,
 194                                        dst_port=dst_port)
 195         (dp, ofp, ofpp) = self._get_dp()
 196         expected = [
 197             call._send_msg(ofpp.OFPFlowMod(dp,
 198                 cookie=0,
 199                 instructions=[
 200                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 201                         ofpp.OFPActionPopVlan(),
 202                         ofpp.OFPActionSetField(eth_src=gateway_mac),
 203                         ofpp.OFPActionOutput(dst_port, 0),
 204                     ]),
 205                 ],
 206                 match=ofpp.OFPMatch(
 207                     eth_dst=dst_mac,
 208                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
 209                 priority=4,
 210                 table_id=2)),
 211         ]
 212         self.assertEqual(expected, self.mock.mock_calls)
 213
 214     def test_delete_dvr_to_src_mac_vlan(self):
 215         network_type = 'vlan'
 216         vlan_tag = 1111
 217         dst_mac = '00:02:b3:13:fe:3d'
 218         self.br.delete_dvr_to_src_mac(network_type=network_type,
 219                                       vlan_tag=vlan_tag,
 220                                       dst_mac=dst_mac)
 221         (dp, ofp, ofpp) = self._get_dp()
 222         expected = [
 223             call.delete_flows(table_id=2,
 224                 match=ofpp.OFPMatch(
 225                     eth_dst=dst_mac,
 226                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
 227         ]
 228         self.assertEqual(expected, self.mock.mock_calls)
 229
 230     def test_add_dvr_mac_vlan(self):
 231         mac = '00:02:b3:13:fe:3d'
 232         port = 8888
 233         self.br.add_dvr_mac_vlan(mac=mac, port=port)
 234         (dp, ofp, ofpp) = self._get_dp()
 235         expected = [
 236             call._send_msg(ofpp.OFPFlowMod(dp,
 237                 cookie=0,
 238                 instructions=[
 239                     ofpp.OFPInstructionGotoTable(table_id=2),
 240                 ],
 241                 match=ofpp.OFPMatch(
 242                     eth_src=mac,
 243                     in_port=port),
 244                 priority=4,
 245                 table_id=0))
 246         ]
 247         self.assertEqual(expected, self.mock.mock_calls)
 248
 249     def test_remove_dvr_mac_vlan(self):
 250         mac = '00:02:b3:13:fe:3d'
 251         self.br.remove_dvr_mac_vlan(mac=mac)
 252         (dp, ofp, ofpp) = self._get_dp()
 253         expected = [
 254             call.delete_flows(eth_src=mac, table_id=0),
 255         ]
 256         self.assertEqual(expected, self.mock.mock_calls)
 257
 258     def test_add_dvr_mac_tun(self):
 259         mac = '00:02:b3:13:fe:3d'
 260         port = 8888
 261         self.br.add_dvr_mac_tun(mac=mac, port=port)
 262         (dp, ofp, ofpp) = self._get_dp()
 263         expected = [
 264             call._send_msg(ofpp.OFPFlowMod(dp,
 265                 cookie=0,
 266                 instructions=[
 267                     ofpp.OFPInstructionGotoTable(table_id=1),
 268                 ],
 269                 match=ofpp.OFPMatch(
 270                     eth_src=mac,
 271                     in_port=port),
 272                 priority=2,
 273                 table_id=0))
 274         ]
 275         self.assertEqual(expected, self.mock.mock_calls)
 276
 277     def test_remove_dvr_mac_tun(self):
 278         mac = '00:02:b3:13:fe:3d'
 279         port = 8888
 280         self.br.remove_dvr_mac_tun(mac=mac, port=port)
 281         expected = [
 282             call.delete_flows(eth_src=mac, in_port=port, table_id=0),
 283         ]
 284         self.assertEqual(expected, self.mock.mock_calls)
 285
 286     def test_install_icmpv6_na_spoofing_protection(self):
 287         port = 8888
 288         ip_addresses = ['2001:db8::1', 'fdf8:f53b:82e4::1/128']
 289         self.br.install_icmpv6_na_spoofing_protection(port, ip_addresses)
 290         (dp, ofp, ofpp) = self._get_dp()
 291         expected = [
 292             call._send_msg(ofpp.OFPFlowMod(dp,
 293                 cookie=0,
 294                 instructions=[
 295                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 296                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
 297                     ]),
 298                 ],
 299                 match=ofpp.OFPMatch(
 300                     eth_type=self.ether_types.ETH_TYPE_IPV6,
 301                     icmpv6_type=self.icmpv6.ND_NEIGHBOR_ADVERT,
 302                     ip_proto=self.in_proto.IPPROTO_ICMPV6,
 303                     ipv6_nd_target='2001:db8::1',
 304                     in_port=8888,
 305                 ),
 306                 priority=2,
 307                 table_id=24)),
 308             call._send_msg(ofpp.OFPFlowMod(dp,
 309                 cookie=0,
 310                 instructions=[
 311                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 312                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
 313                     ]),
 314                 ],
 315                 match=ofpp.OFPMatch(
 316                     eth_type=self.ether_types.ETH_TYPE_IPV6,
 317                     icmpv6_type=self.icmpv6.ND_NEIGHBOR_ADVERT,
 318                     ip_proto=self.in_proto.IPPROTO_ICMPV6,
 319                     ipv6_nd_target='fdf8:f53b:82e4::1',
 320                     in_port=8888,
 321                 ),
 322                 priority=2,
 323                 table_id=24)),
 324             call._send_msg(ofpp.OFPFlowMod(dp,
 325                 cookie=0,
 326                 instructions=[
 327                     ofpp.OFPInstructionGotoTable(table_id=24),
 328                 ],
 329                 match=ofpp.OFPMatch(
 330                     eth_type=self.ether_types.ETH_TYPE_IPV6,
 331                     icmpv6_type=self.icmpv6.ND_NEIGHBOR_ADVERT,
 332                     ip_proto=self.in_proto.IPPROTO_ICMPV6,
 333                     in_port=8888,
 334                 ),
 335                 priority=10,
 336                 table_id=0)),
 337         ]
 338         self.assertEqual(expected, self.mock.mock_calls)
 339
 340     def test_install_arp_spoofing_protection(self):
 341         port = 8888
 342         ip_addresses = ['192.0.2.1', '192.0.2.2/32']
 343         self.br.install_arp_spoofing_protection(port, ip_addresses)
 344         (dp, ofp, ofpp) = self._get_dp()
 345         expected = [
 346             call._send_msg(ofpp.OFPFlowMod(dp,
 347                 cookie=0,
 348                 instructions=[
 349                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 350                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
 351                     ]),
 352                 ],
 353                 match=ofpp.OFPMatch(
 354                     eth_type=self.ether_types.ETH_TYPE_ARP,
 355                     arp_spa='192.0.2.1',
 356                     in_port=8888,
 357                 ),
 358                 priority=2,
 359                 table_id=24)),
 360             call._send_msg(ofpp.OFPFlowMod(dp,
 361                 cookie=0,
 362                 instructions=[
 363                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 364                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
 365                     ]),
 366                 ],
 367                 match=ofpp.OFPMatch(
 368                     eth_type=self.ether_types.ETH_TYPE_ARP,
 369                     arp_spa='192.0.2.2',
 370                     in_port=8888
 371                 ),
 372                 priority=2,
 373                 table_id=24)),
 374             call._send_msg(ofpp.OFPFlowMod(dp,
 375                 cookie=0,
 376                 instructions=[
 377                     ofpp.OFPInstructionGotoTable(table_id=24),
 378                 ],
 379                 match=ofpp.OFPMatch(
 380                     eth_type=self.ether_types.ETH_TYPE_ARP,
 381                     in_port=8888,
 382                 ),
 383                 priority=10,
 384                 table_id=0)),
 385         ]
 386         self.assertEqual(expected, self.mock.mock_calls)
 387
 388     def test_delete_arp_spoofing_protection(self):
 389         port = 8888
 390         self.br.delete_arp_spoofing_protection(port)
 391         (dp, ofp, ofpp) = self._get_dp()
 392         expected = [
 393             call.delete_flows(table_id=0, match=ofpp.OFPMatch(
 394                 eth_type=self.ether_types.ETH_TYPE_ARP,
 395                 in_port=8888)),
 396             call.delete_flows(table_id=0, match=ofpp.OFPMatch(
 397                 eth_type=self.ether_types.ETH_TYPE_IPV6,
 398                 icmpv6_type=self.icmpv6.ND_NEIGHBOR_ADVERT,
 399                 in_port=8888,
 400                 ip_proto=self.in_proto.IPPROTO_ICMPV6)),
 401             call.delete_flows(table_id=24, in_port=port),
 402         ]
 403         self.assertEqual(expected, self.mock.mock_calls)