neutron/tests/functional/agent/linux/test_dhcp.py

   1 # Copyright (c) 2015 Mirantis, Inc.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    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, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 import mock
  16
  17 from oslo_config import cfg
  18
  19 from neutron.agent.common import config
  20 from neutron.agent.dhcp import config as dhcp_conf
  21 from neutron.agent.linux import dhcp
  22 from neutron.agent.linux import interface
  23 from neutron.agent.linux import ip_lib
  24 from neutron.common import config as common_conf
  25 from neutron.tests import base as tests_base
  26 from neutron.tests.common import net_helpers
  27 from neutron.tests.functional import base as functional_base
  28
  29
  30 class TestDhcp(functional_base.BaseSudoTestCase):
  31     def setUp(self):
  32         super(TestDhcp, self).setUp()
  33         conf = cfg.ConfigOpts()
  34         conf.register_opts(config.INTERFACE_DRIVER_OPTS)
  35         conf.register_opts(interface.OPTS)
  36         conf.register_opts(common_conf.core_opts)
  37         conf.register_opts(dhcp_conf.DHCP_AGENT_OPTS)
  38         conf.set_override('interface_driver', 'openvswitch')
  39         conf.set_override('host', 'foo_host')
  40         self.conf = conf
  41         br_int = self.useFixture(net_helpers.OVSBridgeFixture()).bridge
  42         self.conf.set_override('ovs_integration_bridge', br_int.br_name)
  43
  44     def test_cleanup_stale_devices(self):
  45         plugin = mock.MagicMock()
  46         dev_mgr = dhcp.DeviceManager(self.conf, plugin)
  47         network = {
  48             'id': 'foo_id',
  49             'tenant_id': 'foo_tenant',
  50             'namespace': 'qdhcp-foo_id',
  51             'ports': [],
  52             'subnets': [tests_base.AttributeDict({'id': 'subnet_foo_id',
  53                                                   'enable_dhcp': True,
  54                                                   'ipv6_address_mode': None,
  55                                                   'ipv6_ra_mode': None,
  56                                                   'cidr': '10.0.0.0/24',
  57                                                   'ip_version': 4,
  58                                                   'gateway_ip': '10.0.0.1'})]}
  59         dhcp_port = {
  60             'id': 'foo_port_id',
  61             'mac_address': '10:22:33:44:55:67',
  62             'fixed_ips': [tests_base.AttributeDict(
  63                 {'subnet_id': 'subnet_foo_id', 'ip_address': '10.0.0.1'})]
  64         }
  65         plugin.create_dhcp_port.return_value = tests_base.AttributeDict(
  66             dhcp_port)
  67         dev_mgr.driver.plug("foo_id",
  68                             "foo_id2",
  69                             "tapfoo_id2",
  70                             "10:22:33:44:55:68",
  71                             namespace="qdhcp-foo_id")
  72         dev_mgr.driver.plug("foo_id",
  73                             "foo_id3",
  74                             "tapfoo_id3",
  75                             "10:22:33:44:55:69",
  76                             namespace="qdhcp-foo_id")
  77         ipw = ip_lib.IPWrapper(namespace="qdhcp-foo_id")
  78         devices = ipw.get_devices(exclude_loopback=True)
  79         self.addCleanup(ipw.netns.delete, 'qdhcp-foo_id')
  80         self.assertEqual(2, len(devices))
  81         # setting up dhcp for the network
  82         dev_mgr.setup(tests_base.AttributeDict(network))
  83         devices = ipw.get_devices(exclude_loopback=True)
  84         # only one non-loopback device should remain
  85         self.assertEqual(1, len(devices))
  86         self.assertEqual("tapfoo_port_id", devices[0].name)