neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/openflow/native/test_br_phys.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 import neutron.plugins.ml2.drivers.openvswitch.agent.common.constants \
  20     as ovs_const
  21 from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent.openflow.native \
  22     import ovs_bridge_test_base
  23
  24
  25 call = mock.call  # short hand
  26
  27
  28 class OVSPhysicalBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase,
  29                             ovs_bridge_test_base.OVSDVRProcessTestMixin):
  30     dvr_process_table_id = ovs_const.DVR_PROCESS_VLAN
  31     dvr_process_next_table_id = ovs_const.LOCAL_VLAN_TRANSLATION
  32
  33     def setUp(self):
  34         super(OVSPhysicalBridgeTest, self).setUp()
  35         self.setup_bridge_mock('br-phys', self.br_phys_cls)
  36
  37     def test_setup_default_table(self):
  38         self.br.setup_default_table()
  39         (dp, ofp, ofpp) = self._get_dp()
  40         expected = [
  41             call.delete_flows(),
  42             call._send_msg(ofpp.OFPFlowMod(dp,
  43                 cookie=0,
  44                 instructions=[
  45                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  46                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
  47                     ]),
  48                 ],
  49                 match=ofpp.OFPMatch(),
  50                 priority=0,
  51                 table_id=0)),
  52         ]
  53         self.assertEqual(expected, self.mock.mock_calls)
  54
  55     def test_provision_local_vlan(self):
  56         port = 999
  57         lvid = 888
  58         segmentation_id = 777
  59         distributed = False
  60         self.br.provision_local_vlan(port=port, lvid=lvid,
  61                                      segmentation_id=segmentation_id,
  62                                      distributed=distributed)
  63         (dp, ofp, ofpp) = self._get_dp()
  64         expected = [
  65             call._send_msg(ofpp.OFPFlowMod(dp,
  66                 cookie=0,
  67                 instructions=[
  68                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  69                         ofpp.OFPActionSetField(
  70                             vlan_vid=segmentation_id | ofp.OFPVID_PRESENT),
  71                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
  72                     ]),
  73                 ],
  74                 match=ofpp.OFPMatch(
  75                     in_port=port,
  76                     vlan_vid=lvid | ofp.OFPVID_PRESENT),
  77                 priority=4,
  78                 table_id=0)),
  79         ]
  80         self.assertEqual(expected, self.mock.mock_calls)
  81
  82     def test_provision_local_vlan_novlan(self):
  83         port = 999
  84         lvid = 888
  85         segmentation_id = None
  86         distributed = False
  87         self.br.provision_local_vlan(port=port, lvid=lvid,
  88                                      segmentation_id=segmentation_id,
  89                                      distributed=distributed)
  90         (dp, ofp, ofpp) = self._get_dp()
  91         expected = [
  92             call._send_msg(ofpp.OFPFlowMod(dp,
  93                 cookie=0,
  94                 instructions=[
  95                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
  96                         ofpp.OFPActionPopVlan(),
  97                         ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0),
  98                     ]),
  99                 ],
 100                 match=ofpp.OFPMatch(
 101                     in_port=port,
 102                     vlan_vid=lvid | ofp.OFPVID_PRESENT),
 103                 priority=4,
 104                 table_id=0)),
 105         ]
 106         self.assertEqual(expected, self.mock.mock_calls)
 107
 108     def test_reclaim_local_vlan(self):
 109         port = 999
 110         lvid = 888
 111         self.br.reclaim_local_vlan(port=port, lvid=lvid)
 112         (dp, ofp, ofpp) = self._get_dp()
 113         expected = [
 114             call.delete_flows(
 115                 match=ofpp.OFPMatch(
 116                     in_port=port,
 117                     vlan_vid=lvid | ofp.OFPVID_PRESENT)),
 118         ]
 119         self.assertEqual(expected, self.mock.mock_calls)
 120
 121     def test_add_dvr_mac_vlan(self):
 122         mac = '00:02:b3:13:fe:3d'
 123         port = 8888
 124         self.br.add_dvr_mac_vlan(mac=mac, port=port)
 125         (dp, ofp, ofpp) = self._get_dp()
 126         expected = [
 127             call._send_msg(ofpp.OFPFlowMod(dp,
 128                 cookie=0,
 129                 instructions=[
 130                     ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
 131                         ofpp.OFPActionOutput(port, 0),
 132                     ]),
 133                 ],
 134                 match=ofpp.OFPMatch(eth_src=mac),
 135                 priority=2,
 136                 table_id=3)),
 137         ]
 138         self.assertEqual(expected, self.mock.mock_calls)
 139
 140     def test_remove_dvr_mac_vlan(self):
 141         mac = '00:02:b3:13:fe:3d'
 142         self.br.remove_dvr_mac_vlan(mac=mac)
 143         (dp, ofp, ofpp) = self._get_dp()
 144         expected = [
 145             call.delete_flows(eth_src=mac, table_id=3),
 146         ]
 147         self.assertEqual(expected, self.mock.mock_calls)