neutron/tests/unit/cmd/test_ovs_cleanup.py

   1 # Copyright (c) 2012 OpenStack Foundation.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import itertools
  17
  18 import mock
  19 from oslo_utils import uuidutils
  20
  21 from neutron.agent.common import ovs_lib
  22 from neutron.agent.linux import ip_lib
  23 from neutron.cmd import ovs_cleanup as util
  24 from neutron.tests import base
  25
  26
  27 class TestOVSCleanup(base.BaseTestCase):
  28
  29     @mock.patch('neutron.common.config.setup_logging')
  30     @mock.patch('neutron.cmd.ovs_cleanup.setup_conf')
  31     @mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges')
  32     @mock.patch('neutron.agent.common.ovs_lib.OVSBridge')
  33     @mock.patch.object(util, 'collect_neutron_ports')
  34     @mock.patch.object(util, 'delete_neutron_ports')
  35     def test_main(self, mock_delete, mock_collect, mock_ovs,
  36                   mock_get_bridges, mock_conf, mock_logging):
  37         bridges = ['br-int', 'br-ex']
  38         ports = ['p1', 'p2', 'p3']
  39         conf = mock.Mock()
  40         conf.ovs_all_ports = False
  41         conf.ovs_integration_bridge = 'br-int'
  42         conf.external_network_bridge = 'br-ex'
  43         mock_conf.return_value = conf
  44         mock_get_bridges.return_value = bridges
  45         mock_collect.return_value = ports
  46
  47         util.main()
  48         mock_ovs.assert_has_calls([mock.call().delete_ports(
  49             all_ports=False)])
  50         mock_collect.assert_called_once_with(set(bridges))
  51         mock_delete.assert_called_once_with(ports)
  52
  53     def test_collect_neutron_ports(self):
  54         port1 = ovs_lib.VifPort('tap1234', 1, uuidutils.generate_uuid(),
  55                                 '11:22:33:44:55:66', 'br')
  56         port2 = ovs_lib.VifPort('tap5678', 2, uuidutils.generate_uuid(),
  57                                 '77:88:99:aa:bb:cc', 'br')
  58         port3 = ovs_lib.VifPort('tap90ab', 3, uuidutils.generate_uuid(),
  59                                 '99:00:aa:bb:cc:dd', 'br')
  60         ports = [[port1, port2], [port3]]
  61         portnames = [p.port_name for p in itertools.chain(*ports)]
  62         with mock.patch('neutron.agent.common.ovs_lib.OVSBridge') as ovs:
  63             ovs.return_value.get_vif_ports.side_effect = ports
  64             bridges = ['br-int', 'br-ex']
  65             ret = util.collect_neutron_ports(bridges)
  66             self.assertEqual(ret, portnames)
  67
  68     @mock.patch.object(ip_lib, 'IPDevice')
  69     def test_delete_neutron_ports(self, mock_ip):
  70         ports = ['tap1234', 'tap5678', 'tap09ab']
  71         port_found = [True, False, True]
  72
  73         mock_ip.return_value.exists.side_effect = port_found
  74         util.delete_neutron_ports(ports)
  75         mock_ip.assert_has_calls(
  76             [mock.call('tap1234'),
  77              mock.call().exists(),
  78              mock.call().link.delete(),
  79              mock.call('tap5678'),
  80              mock.call().exists(),
  81              mock.call('tap09ab'),
  82              mock.call().exists(),
  83              mock.call().link.delete()])