# under the License.
import os
-import socket
-import uuid
import eventlet
import netaddr
from neutron.agent.linux import dhcp
from neutron.agent.linux import external_process
from neutron.agent.linux import interface
-from neutron.agent.linux import ip_lib
from neutron.agent import rpc as agent_rpc
from neutron.common import constants
-from neutron.common import exceptions
from neutron.common import legacy
from neutron.common import topics
from neutron.common import utils
from neutron import service as neutron_service
LOG = logging.getLogger(__name__)
-NS_PREFIX = 'qdhcp-'
-METADATA_DEFAULT_PREFIX = 16
-METADATA_DEFAULT_IP = '169.254.169.254/%d' % METADATA_DEFAULT_PREFIX
-METADATA_PORT = 80
class DhcpAgent(manager.Manager):
self.root_helper = config.get_root_helper(self.conf)
self.dhcp_driver_cls = importutils.import_class(self.conf.dhcp_driver)
ctx = context.get_admin_context_without_session()
- self.plugin_rpc = DhcpPluginApi(topics.PLUGIN, ctx)
- self.device_manager = DeviceManager(self.conf, self.plugin_rpc)
+ self.plugin_rpc = DhcpPluginApi(topics.PLUGIN,
+ ctx, self.conf.use_namespaces)
# create dhcp dir to store dhcp info
dhcp_dir = os.path.dirname("/%s/dhcp/" % self.conf.state_path)
if not os.path.isdir(dhcp_dir):
def _populate_networks_cache(self):
"""Populate the networks cache when the DHCP-agent starts."""
-
try:
existing_networks = self.dhcp_driver_cls.existing_dhcp_networks(
self.conf,
self.root_helper
)
-
for net_id in existing_networks:
- net = DictModel({"id": net_id, "subnets": [], "ports": []})
+ net = dhcp.NetModel(self.conf.use_namespaces,
+ {"id": net_id,
+ "subnets": [],
+ "ports": []})
self.cache.put(net)
except NotImplementedError:
# just go ahead with an empty networks cache
self.sync_state()
self.periodic_resync()
- def _ns_name(self, network):
- if self.conf.use_namespaces:
- return NS_PREFIX + network.id
-
def call_driver(self, action, network, **action_kwargs):
"""Invoke an action on a DHCP driver instance."""
try:
driver = self.dhcp_driver_cls(self.conf,
network,
self.root_helper,
- self.device_manager,
- self._ns_name(network),
- self.dhcp_version)
+ self.dhcp_version,
+ self.plugin_rpc)
+
getattr(driver, action)(**action_kwargs)
return True
else:
self.disable_dhcp_helper(network.id)
- if new_cidrs:
- self.device_manager.update(network)
-
def release_lease_for_removed_ips(self, port, network):
"""Releases the dhcp lease for ips removed from a port."""
prev_port = self.cache.get_port_by_id(port.id)
@utils.synchronized('dhcp-agent')
def port_update_end(self, context, payload):
"""Handle the port.update.end notification event."""
- port = DictModel(payload['port'])
+ port = dhcp.DictModel(payload['port'])
network = self.cache.get_network_by_id(port.network_id)
if network:
self.release_lease_for_removed_ips(port, network)
# or all the networks connected via a router
# to the one passed as a parameter
neutron_lookup_param = '--network_id=%s' % network.id
- meta_cidr = netaddr.IPNetwork(METADATA_DEFAULT_IP)
+ meta_cidr = netaddr.IPNetwork(dhcp.METADATA_DEFAULT_CIDR)
has_metadata_subnet = any(netaddr.IPNetwork(s.cidr) in meta_cidr
for s in network.subnets)
if (self.conf.enable_metadata_network and has_metadata_subnet):
'--metadata_proxy_socket=%s' % metadata_proxy_socket,
neutron_lookup_param,
'--state_path=%s' % self.conf.state_path,
- '--metadata_port=%d' % METADATA_PORT]
+ '--metadata_port=%d' % dhcp.METADATA_PORT]
proxy_cmd.extend(config.get_log_args(
cfg.CONF, 'neutron-ns-metadata-proxy-%s.log' % network.id))
return proxy_cmd
self.conf,
network.id,
self.root_helper,
- self._ns_name(network))
+ network.namespace)
pm.enable(callback)
def disable_isolated_metadata_proxy(self, network):
self.conf,
network.id,
self.root_helper,
- self._ns_name(network))
+ network.namespace)
pm.disable()
BASE_RPC_API_VERSION = '1.1'
- def __init__(self, topic, context):
+ def __init__(self, topic, context, use_namespaces):
super(DhcpPluginApi, self).__init__(
topic=topic, default_version=self.BASE_RPC_API_VERSION)
self.context = context
self.host = cfg.CONF.host
+ self.use_namespaces = use_namespaces
def get_active_networks_info(self):
"""Make a remote process call to retrieve all network info."""
self.make_msg('get_active_networks_info',
host=self.host),
topic=self.topic)
- return [DictModel(n) for n in networks]
+ return [dhcp.NetModel(self.use_namespaces, n) for n in networks]
def get_network_info(self, network_id):
"""Make a remote process call to retrieve network info."""
- return DictModel(self.call(self.context,
- self.make_msg('get_network_info',
- network_id=network_id,
- host=self.host),
- topic=self.topic))
+ return dhcp.NetModel(self.use_namespaces,
+ self.call(self.context,
+ self.make_msg('get_network_info',
+ network_id=network_id,
+ host=self.host),
+ topic=self.topic))
def get_dhcp_port(self, network_id, device_id):
"""Make a remote process call to get the dhcp port."""
- return DictModel(self.call(self.context,
- self.make_msg('get_dhcp_port',
- network_id=network_id,
- device_id=device_id,
- host=self.host),
- topic=self.topic))
+ return dhcp.DictModel(self.call(self.context,
+ self.make_msg('get_dhcp_port',
+ network_id=network_id,
+ device_id=device_id,
+ host=self.host),
+ topic=self.topic))
def create_dhcp_port(self, port):
"""Make a remote process call to create the dhcp port."""
- return DictModel(self.call(self.context,
- self.make_msg('create_dhcp_port',
- port=port,
- host=self.host),
- topic=self.topic))
+ return dhcp.DictModel(self.call(self.context,
+ self.make_msg('create_dhcp_port',
+ port=port,
+ host=self.host),
+ topic=self.topic))
def update_dhcp_port(self, port_id, port):
"""Make a remote process call to update the dhcp port."""
- return DictModel(self.call(self.context,
- self.make_msg('update_dhcp_port',
- port_id=port_id,
- port=port,
- host=self.host),
- topic=self.topic))
+ return dhcp.DictModel(self.call(self.context,
+ self.make_msg('update_dhcp_port',
+ port_id=port_id,
+ port=port,
+ host=self.host),
+ topic=self.topic))
def release_dhcp_port(self, network_id, device_id):
"""Make a remote process call to release the dhcp port."""
'ports': num_ports}
-class DeviceManager(object):
- OPTS = [
- cfg.StrOpt('interface_driver',
- help=_("The driver used to manage the virtual interface."))
- ]
-
- def __init__(self, conf, plugin):
- self.conf = conf
- self.root_helper = config.get_root_helper(conf)
- self.plugin = plugin
- if not conf.interface_driver:
- raise SystemExit(_('You must specify an interface driver'))
- try:
- self.driver = importutils.import_object(
- conf.interface_driver, conf
- )
- except Exception:
- msg = _("Error importing interface driver "
- "'%s'") % conf.interface_driver
- raise SystemExit(msg)
-
- def get_interface_name(self, network, port):
- """Return interface(device) name for use by the DHCP process."""
- return self.driver.get_device_name(port)
-
- def get_device_id(self, network):
- """Return a unique DHCP device ID for this host on the network."""
- # There could be more than one dhcp server per network, so create
- # a device id that combines host and network ids
-
- host_uuid = uuid.uuid5(uuid.NAMESPACE_DNS, socket.gethostname())
- return 'dhcp%s-%s' % (host_uuid, network.id)
-
- def _get_device(self, network):
- """Return DHCP ip_lib device for this host on the network."""
- device_id = self.get_device_id(network)
- port = self.plugin.get_dhcp_port(network.id, device_id)
- interface_name = self.get_interface_name(network, port)
- namespace = NS_PREFIX + network.id
- return ip_lib.IPDevice(interface_name,
- self.root_helper,
- namespace)
-
- def _set_default_route(self, network):
- """Sets the default gateway for this dhcp namespace.
-
- This method is idempotent and will only adjust the route if adjusting
- it would change it from what it already is. This makes it safe to call
- and avoids unnecessary perturbation of the system.
- """
- device = self._get_device(network)
- gateway = device.route.get_gateway()
- if gateway:
- gateway = gateway['gateway']
-
- for subnet in network.subnets:
- skip_subnet = (
- subnet.ip_version != 4
- or not subnet.enable_dhcp
- or subnet.gateway_ip is None)
-
- if skip_subnet:
- continue
-
- if gateway != subnet.gateway_ip:
- m = _('Setting gateway for dhcp netns on net %(n)s to %(ip)s')
- LOG.debug(m, {'n': network.id, 'ip': subnet.gateway_ip})
-
- device.route.add_gateway(subnet.gateway_ip)
-
- return
-
- # No subnets on the network have a valid gateway. Clean it up to avoid
- # confusion from seeing an invalid gateway here.
- if gateway is not None:
- msg = _('Removing gateway for dhcp netns on net %s')
- LOG.debug(msg, network.id)
-
- device.route.delete_gateway(gateway)
-
- def setup_dhcp_port(self, network):
- """Create/update DHCP port for the host if needed and return port."""
-
- device_id = self.get_device_id(network)
- subnets = {}
- dhcp_enabled_subnet_ids = []
- for subnet in network.subnets:
- if subnet.enable_dhcp:
- dhcp_enabled_subnet_ids.append(subnet.id)
- subnets[subnet.id] = subnet
-
- dhcp_port = None
- for port in network.ports:
- port_device_id = getattr(port, 'device_id', None)
- if port_device_id == device_id:
- port_fixed_ips = []
- for fixed_ip in port.fixed_ips:
- port_fixed_ips.append({'subnet_id': fixed_ip.subnet_id,
- 'ip_address': fixed_ip.ip_address})
- if fixed_ip.subnet_id in dhcp_enabled_subnet_ids:
- dhcp_enabled_subnet_ids.remove(fixed_ip.subnet_id)
-
- # If there are dhcp_enabled_subnet_ids here that means that
- # we need to add those to the port and call update.
- if dhcp_enabled_subnet_ids:
- port_fixed_ips.extend(
- [dict(subnet_id=s) for s in dhcp_enabled_subnet_ids])
- dhcp_port = self.plugin.update_dhcp_port(
- port.id, {'port': {'fixed_ips': port_fixed_ips}})
- else:
- dhcp_port = port
- # break since we found port that matches device_id
- break
-
- # DHCP port has not yet been created.
- if dhcp_port is None:
- LOG.debug(_('DHCP port %(device_id)s on network %(network_id)s'
- ' does not yet exist.'), {'device_id': device_id,
- 'network_id': network.id})
- port_dict = dict(
- name='',
- admin_state_up=True,
- device_id=device_id,
- network_id=network.id,
- tenant_id=network.tenant_id,
- fixed_ips=[dict(subnet_id=s) for s in dhcp_enabled_subnet_ids])
- dhcp_port = self.plugin.create_dhcp_port({'port': port_dict})
-
- # Convert subnet_id to subnet dict
- fixed_ips = [dict(subnet_id=fixed_ip.subnet_id,
- ip_address=fixed_ip.ip_address,
- subnet=subnets[fixed_ip.subnet_id])
- for fixed_ip in dhcp_port.fixed_ips]
-
- ips = [DictModel(item) if isinstance(item, dict) else item
- for item in fixed_ips]
- dhcp_port.fixed_ips = ips
-
- return dhcp_port
-
- def setup(self, network, reuse_existing=False):
- """Create and initialize a device for network's DHCP on this host."""
- port = self.setup_dhcp_port(network)
- interface_name = self.get_interface_name(network, port)
-
- if self.conf.use_namespaces:
- namespace = NS_PREFIX + network.id
- else:
- namespace = None
-
- if ip_lib.device_exists(interface_name,
- self.root_helper,
- namespace):
- if not reuse_existing:
- raise exceptions.PreexistingDeviceFailure(
- dev_name=interface_name)
-
- LOG.debug(_('Reusing existing device: %s.'), interface_name)
- else:
- self.driver.plug(network.id,
- port.id,
- interface_name,
- port.mac_address,
- namespace=namespace)
- ip_cidrs = []
- for fixed_ip in port.fixed_ips:
- subnet = fixed_ip.subnet
- net = netaddr.IPNetwork(subnet.cidr)
- ip_cidr = '%s/%s' % (fixed_ip.ip_address, net.prefixlen)
- ip_cidrs.append(ip_cidr)
-
- if (self.conf.enable_isolated_metadata and
- self.conf.use_namespaces):
- ip_cidrs.append(METADATA_DEFAULT_IP)
-
- self.driver.init_l3(interface_name, ip_cidrs,
- namespace=namespace)
-
- # ensure that the dhcp interface is first in the list
- if namespace is None:
- device = ip_lib.IPDevice(interface_name,
- self.root_helper)
- device.route.pullup_route(interface_name)
-
- if self.conf.use_namespaces:
- self._set_default_route(network)
-
- return interface_name
-
- def update(self, network):
- """Update device settings for the network's DHCP on this host."""
- if self.conf.use_namespaces:
- self._set_default_route(network)
-
- def destroy(self, network, device_name):
- """Destroy the device used for the network's DHCP on this host."""
- if self.conf.use_namespaces:
- namespace = NS_PREFIX + network.id
- else:
- namespace = None
-
- self.driver.unplug(device_name, namespace=namespace)
-
- self.plugin.release_dhcp_port(network.id,
- self.get_device_id(network))
-
-
-class DictModel(object):
- """Convert dict into an object that provides attribute access to values."""
- def __init__(self, d):
- for key, value in d.iteritems():
- if isinstance(value, list):
- value = [DictModel(item) if isinstance(item, dict) else item
- for item in value]
- elif isinstance(value, dict):
- value = DictModel(value)
-
- setattr(self, key, value)
-
-
class DhcpAgentWithStateReport(DhcpAgent):
def __init__(self, host=None):
super(DhcpAgentWithStateReport, self).__init__(host=host)
cfg.CONF.register_opts(DhcpAgent.OPTS)
config.register_agent_state_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)
- cfg.CONF.register_opts(DeviceManager.OPTS)
cfg.CONF.register_opts(dhcp.OPTS)
cfg.CONF.register_opts(interface.OPTS)
import socket
import StringIO
import sys
+import uuid
import netaddr
from oslo.config import cfg
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
+from neutron.common import exceptions
+from neutron.openstack.common import importutils
from neutron.openstack.common import jsonutils
from neutron.openstack.common import log as logging
from neutron.openstack.common import uuidutils
cfg.StrOpt('dnsmasq_dns_server',
help=_('Use another DNS server before any in '
'/etc/resolv.conf.')),
+ cfg.StrOpt('interface_driver',
+ help=_("The driver used to manage the virtual interface.")),
]
IPV4 = 4
DNS_PORT = 53
DHCPV4_PORT = 67
DHCPV6_PORT = 547
+METADATA_DEFAULT_PREFIX = 16
METADATA_DEFAULT_IP = '169.254.169.254'
+METADATA_DEFAULT_CIDR = '%s/%d' % (METADATA_DEFAULT_IP,
+ METADATA_DEFAULT_PREFIX)
+METADATA_PORT = 80
WIN2k3_STATIC_DNS = 249
+NS_PREFIX = 'qdhcp-'
+
+
+class DictModel(object):
+ """Convert dict into an object that provides attribute access to values."""
+ def __init__(self, d):
+ for key, value in d.iteritems():
+ if isinstance(value, list):
+ value = [DictModel(item) if isinstance(item, dict) else item
+ for item in value]
+ elif isinstance(value, dict):
+ value = DictModel(value)
+
+ setattr(self, key, value)
+
+
+class NetModel(DictModel):
+
+ def __init__(self, use_namespaces, d):
+ super(NetModel, self).__init__(d)
+
+ self._ns_name = (use_namespaces and
+ "%s%s" % (NS_PREFIX, self.id) or None)
+
+ @property
+ def namespace(self):
+ return self._ns_name
class DhcpBase(object):
__metaclass__ = abc.ABCMeta
def __init__(self, conf, network, root_helper='sudo',
- device_delegate=None, namespace=None, version=None):
+ version=None, plugin=None):
self.conf = conf
self.network = network
self.root_helper = root_helper
- self.device_delegate = device_delegate
- self.namespace = namespace
+ self.device_manager = DeviceManager(self.conf,
+ self.root_helper, plugin)
self.version = version
@abc.abstractmethod
"""Restart the dhcp service for the network."""
self.disable(retain_port=True)
self.enable()
+ self.device_manager.update(self.network)
@abc.abstractproperty
def active(self):
def enable(self):
"""Enables DHCP for this network by spawning a local process."""
- interface_name = self.device_delegate.setup(self.network,
- reuse_existing=True)
+ interface_name = self.device_manager.setup(self.network,
+ reuse_existing=True)
if self.active:
self.restart()
elif self._enable_dhcp():
cmd = ['kill', '-9', pid]
utils.execute(cmd, self.root_helper)
if not retain_port:
- self.device_delegate.destroy(self.network, self.interface_name)
+ self.device_manager.destroy(self.network, self.interface_name)
elif pid:
LOG.debug(_('DHCP for %(net_id)s pid %(pid)d is stale, ignoring '
if self.conf.dhcp_domain:
cmd.append('--domain=%s' % self.conf.dhcp_domain)
- if self.namespace:
- ip_wrapper = ip_lib.IPWrapper(self.root_helper, self.namespace)
+ if self.network.namespace:
+ ip_wrapper = ip_lib.IPWrapper(self.root_helper,
+ self.network.namespace)
ip_wrapper.netns.execute(cmd, addl_env=env)
else:
# For normal sudo prepend the env vars before command
"""Release a DHCP lease."""
for ip in removed_ips or []:
cmd = ['dhcp_release', self.interface_name, ip, mac_address]
- if self.namespace:
- ip_wrapper = ip_lib.IPWrapper(self.root_helper, self.namespace)
+ if self.network.namespace:
+ ip_wrapper = ip_lib.IPWrapper(self.root_helper,
+ self.network.namespace)
ip_wrapper.netns.execute(cmd)
else:
utils.execute(cmd, self.root_helper)
else:
LOG.debug(_('Pid %d is stale, relaunching dnsmasq'), self.pid)
LOG.debug(_('Reloading allocations for network: %s'), self.network.id)
+ self.device_manager.update(self.network)
def _output_hosts_file(self):
"""Writes a dnsmasq compatible hosts file."""
ip_dev = ip_lib.IPDevice(
self.interface_name,
self.root_helper,
- self.namespace
+ self.network.namespace
)
subnet_lookup = dict(
sock.connect(dhcp_relay_socket)
sock.send(jsonutils.dumps(data))
sock.close()
+
+
+class DeviceManager(object):
+
+ def __init__(self, conf, root_helper, plugin):
+ self.conf = conf
+ self.root_helper = root_helper
+ self.plugin = plugin
+ if not conf.interface_driver:
+ raise SystemExit(_('You must specify an interface driver'))
+ try:
+ self.driver = importutils.import_object(
+ conf.interface_driver, conf)
+ except Exception as e:
+ msg = (_("Error importing interface driver '%(driver)s': "
+ "%(inner)s") % {'driver': conf.interface_driver,
+ 'inner': e})
+ raise SystemExit(msg)
+
+ def get_interface_name(self, network, port):
+ """Return interface(device) name for use by the DHCP process."""
+ return self.driver.get_device_name(port)
+
+ def get_device_id(self, network):
+ """Return a unique DHCP device ID for this host on the network."""
+ # There could be more than one dhcp server per network, so create
+ # a device id that combines host and network ids
+
+ host_uuid = uuid.uuid5(uuid.NAMESPACE_DNS, socket.gethostname())
+ return 'dhcp%s-%s' % (host_uuid, network.id)
+
+ def _get_device(self, network):
+ """Return DHCP ip_lib device for this host on the network."""
+ device_id = self.get_device_id(network)
+ port = self.plugin.get_dhcp_port(network.id, device_id)
+ interface_name = self.get_interface_name(network, port)
+ return ip_lib.IPDevice(interface_name,
+ self.root_helper,
+ network.namespace)
+
+ def _set_default_route(self, network):
+ """Sets the default gateway for this dhcp namespace.
+
+ This method is idempotent and will only adjust the route if adjusting
+ it would change it from what it already is. This makes it safe to call
+ and avoids unnecessary perturbation of the system.
+ """
+ device = self._get_device(network)
+ gateway = device.route.get_gateway()
+ if gateway:
+ gateway = gateway['gateway']
+
+ for subnet in network.subnets:
+ skip_subnet = (
+ subnet.ip_version != 4
+ or not subnet.enable_dhcp
+ or subnet.gateway_ip is None)
+
+ if skip_subnet:
+ continue
+
+ if gateway != subnet.gateway_ip:
+ m = _('Setting gateway for dhcp netns on net %(n)s to %(ip)s')
+ LOG.debug(m, {'n': network.id, 'ip': subnet.gateway_ip})
+
+ device.route.add_gateway(subnet.gateway_ip)
+
+ return
+
+ # No subnets on the network have a valid gateway. Clean it up to avoid
+ # confusion from seeing an invalid gateway here.
+ if gateway is not None:
+ msg = _('Removing gateway for dhcp netns on net %s')
+ LOG.debug(msg, network.id)
+
+ device.route.delete_gateway(gateway)
+
+ def setup_dhcp_port(self, network):
+ """Create/update DHCP port for the host if needed and return port."""
+
+ device_id = self.get_device_id(network)
+ subnets = {}
+ dhcp_enabled_subnet_ids = []
+ for subnet in network.subnets:
+ if subnet.enable_dhcp:
+ dhcp_enabled_subnet_ids.append(subnet.id)
+ subnets[subnet.id] = subnet
+
+ dhcp_port = None
+ for port in network.ports:
+ port_device_id = getattr(port, 'device_id', None)
+ if port_device_id == device_id:
+ port_fixed_ips = []
+ for fixed_ip in port.fixed_ips:
+ port_fixed_ips.append({'subnet_id': fixed_ip.subnet_id,
+ 'ip_address': fixed_ip.ip_address})
+ if fixed_ip.subnet_id in dhcp_enabled_subnet_ids:
+ dhcp_enabled_subnet_ids.remove(fixed_ip.subnet_id)
+
+ # If there are dhcp_enabled_subnet_ids here that means that
+ # we need to add those to the port and call update.
+ if dhcp_enabled_subnet_ids:
+ port_fixed_ips.extend(
+ [dict(subnet_id=s) for s in dhcp_enabled_subnet_ids])
+ dhcp_port = self.plugin.update_dhcp_port(
+ port.id, {'port': {'fixed_ips': port_fixed_ips}})
+ else:
+ dhcp_port = port
+ # break since we found port that matches device_id
+ break
+
+ # DHCP port has not yet been created.
+ if dhcp_port is None:
+ LOG.debug(_('DHCP port %(device_id)s on network %(network_id)s'
+ ' does not yet exist.'), {'device_id': device_id,
+ 'network_id': network.id})
+ port_dict = dict(
+ name='',
+ admin_state_up=True,
+ device_id=device_id,
+ network_id=network.id,
+ tenant_id=network.tenant_id,
+ fixed_ips=[dict(subnet_id=s) for s in dhcp_enabled_subnet_ids])
+ dhcp_port = self.plugin.create_dhcp_port({'port': port_dict})
+
+ # Convert subnet_id to subnet dict
+ fixed_ips = [dict(subnet_id=fixed_ip.subnet_id,
+ ip_address=fixed_ip.ip_address,
+ subnet=subnets[fixed_ip.subnet_id])
+ for fixed_ip in dhcp_port.fixed_ips]
+
+ ips = [DictModel(item) if isinstance(item, dict) else item
+ for item in fixed_ips]
+ dhcp_port.fixed_ips = ips
+
+ return dhcp_port
+
+ def setup(self, network, reuse_existing=False):
+ """Create and initialize a device for network's DHCP on this host."""
+ port = self.setup_dhcp_port(network)
+ interface_name = self.get_interface_name(network, port)
+
+ if ip_lib.device_exists(interface_name,
+ self.root_helper,
+ network.namespace):
+ if not reuse_existing:
+ raise exceptions.PreexistingDeviceFailure(
+ dev_name=interface_name)
+
+ LOG.debug(_('Reusing existing device: %s.'), interface_name)
+ else:
+ self.driver.plug(network.id,
+ port.id,
+ interface_name,
+ port.mac_address,
+ namespace=network.namespace)
+ ip_cidrs = []
+ for fixed_ip in port.fixed_ips:
+ subnet = fixed_ip.subnet
+ net = netaddr.IPNetwork(subnet.cidr)
+ ip_cidr = '%s/%s' % (fixed_ip.ip_address, net.prefixlen)
+ ip_cidrs.append(ip_cidr)
+
+ if (self.conf.enable_isolated_metadata and
+ self.conf.use_namespaces):
+ ip_cidrs.append(METADATA_DEFAULT_CIDR)
+
+ self.driver.init_l3(interface_name, ip_cidrs,
+ namespace=network.namespace)
+
+ # ensure that the dhcp interface is first in the list
+ if network.namespace is None:
+ device = ip_lib.IPDevice(interface_name,
+ self.root_helper)
+ device.route.pullup_route(interface_name)
+
+ if self.conf.use_namespaces:
+ self._set_default_route(network)
+
+ return interface_name
+
+ def update(self, network):
+ """Update device settings for the network's DHCP on this host."""
+ if self.conf.use_namespaces:
+ self._set_default_route(network)
+
+ def destroy(self, network, device_name):
+ """Destroy the device used for the network's DHCP on this host."""
+ self.driver.unplug(device_name, namespace=network.namespace)
+
+ self.plugin.release_dhcp_port(network.id,
+ self.get_device_id(network))
from oslo.config import cfg
from neutron.agent.common import config as agent_config
-from neutron.agent import dhcp_agent
from neutron.agent import l3_agent
from neutron.agent.linux import dhcp
from neutron.agent.linux import ip_lib
LOG = logging.getLogger(__name__)
-NS_MANGLING_PATTERN = ('(%s|%s)' % (dhcp_agent.NS_PREFIX, l3_agent.NS_PREFIX) +
+NS_MANGLING_PATTERN = ('(%s|%s)' % (dhcp.NS_PREFIX, l3_agent.NS_PREFIX) +
attributes.UUID_PATTERN)
-class NullDelegate(object):
- def __getattribute__(self, name):
- def noop(*args, **kwargs):
- pass
- return noop
-
-
class FakeNetwork(object):
def __init__(self, id):
self.id = id
def kill_dhcp(conf, namespace):
"""Disable DHCP for a network if DHCP is still active."""
root_helper = agent_config.get_root_helper(conf)
- network_id = namespace.replace(dhcp_agent.NS_PREFIX, '')
+ network_id = namespace.replace(dhcp.NS_PREFIX, '')
- null_delegate = NullDelegate()
dhcp_driver = importutils.import_object(
conf.dhcp_driver,
conf,
FakeNetwork(network_id),
- root_helper,
- null_delegate)
+ root_helper)
if dhcp_driver.active:
dhcp_driver.disable()
from oslo.config import cfg
from neutron.agent.common import config
-from neutron.agent.dhcp_agent import DictModel
+from neutron.agent.linux.dhcp import DictModel
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.openstack.common import log as logging
from neutron.tests import base
-class TestNullDelegate(base.BaseTestCase):
- def test_getattribute(self):
- null_delegate = util.NullDelegate()
- self.assertIsNone(null_delegate.test())
-
-
class TestNetnsCleanup(base.BaseTestCase):
def setUp(self):
super(TestNetnsCleanup, self).setUp()
from neutron.agent.dhcp_agent import DhcpAgentWithStateReport
from neutron.agent.linux import dhcp
from neutron.agent.linux import interface
-from neutron.common import constants
+from neutron.common import constants as const
from neutron.common import exceptions
from neutron.tests import base
ROOTDIR = os.path.dirname(os.path.dirname(__file__))
ETCDIR = os.path.join(ROOTDIR, 'etc')
HOSTNAME = 'hostname'
+dev_man = dhcp.DeviceManager
+rpc_api = dhcp_agent.DhcpPluginApi
+DEVICE_MANAGER = '%s.%s' % (dev_man.__module__, dev_man.__name__)
+DHCP_PLUGIN = '%s.%s' % (rpc_api.__module__, rpc_api.__name__)
def etcdir(*p):
return os.path.join(ETCDIR, *p)
-class FakeModel:
- def __init__(self, id_, **kwargs):
- self.id = id_
- self.__dict__.update(kwargs)
-
- def __str__(self):
- return str(self.__dict__)
-
fake_tenant_id = 'aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'
-fake_subnet1_allocation_pools = FakeModel('', start='172.9.9.2',
- end='172.9.9.254')
-fake_subnet1 = FakeModel('bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
- network_id='12345678-1234-5678-1234567890ab',
- cidr='172.9.9.0/24', enable_dhcp=True, name='',
- tenant_id=fake_tenant_id, gateway_ip='172.9.9.1',
- host_routes=[], dns_nameservers=[], ip_version=4,
- allocation_pools=fake_subnet1_allocation_pools)
-
-fake_subnet2_allocation_pools = FakeModel('', start='172.9.8.2',
- end='172.9.8.254')
-fake_subnet2 = FakeModel('dddddddd-dddd-dddd-dddddddddddd',
- network_id='12345678-1234-5678-1234567890ab',
- cidr='172.9.8.0/24', enable_dhcp=False, name='',
- tenant_id=fake_tenant_id, gateway_ip='172.9.8.1',
- host_routes=[], dns_nameservers=[], ip_version=4,
- allocation_pools=fake_subnet2_allocation_pools)
-
-fake_subnet3 = FakeModel('bbbbbbbb-1111-2222-bbbbbbbbbbbb',
- network_id='12345678-1234-5678-1234567890ab',
- cidr='192.168.1.1/24', enable_dhcp=True)
-
-fake_meta_subnet = FakeModel('bbbbbbbb-1111-2222-bbbbbbbbbbbb',
- network_id='12345678-1234-5678-1234567890ab',
- cidr='169.254.169.252/30',
- gateway_ip='169.254.169.253', enable_dhcp=True)
-
-fake_fixed_ip1 = FakeModel('', subnet_id=fake_subnet1.id,
- ip_address='172.9.9.9')
-fake_meta_fixed_ip = FakeModel('', subnet=fake_meta_subnet,
- ip_address='169.254.169.254')
-fake_allocation_pool_subnet1 = FakeModel('', start='172.9.9.2',
- end='172.9.9.254')
-
-
-fake_port1 = FakeModel('12345678-1234-aaaa-1234567890ab',
- device_id='dhcp-12345678-1234-aaaa-1234567890ab',
- allocation_pools=fake_subnet1_allocation_pools,
- mac_address='aa:bb:cc:dd:ee:ff',
- network_id='12345678-1234-5678-1234567890ab',
- fixed_ips=[fake_fixed_ip1])
-
-fake_port2 = FakeModel('12345678-1234-aaaa-123456789000',
- mac_address='aa:bb:cc:dd:ee:99',
- network_id='12345678-1234-5678-1234567890ab',
- fixed_ips=[])
-
-fake_meta_port = FakeModel('12345678-1234-aaaa-1234567890ab',
- mac_address='aa:bb:cc:dd:ee:ff',
- network_id='12345678-1234-5678-1234567890ab',
- device_owner=constants.DEVICE_OWNER_ROUTER_INTF,
- device_id='forzanapoli',
- fixed_ips=[fake_meta_fixed_ip])
-
-fake_network = FakeModel('12345678-1234-5678-1234567890ab',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
- admin_state_up=True,
- subnets=[fake_subnet1, fake_subnet2],
- ports=[fake_port1])
-
-fake_meta_network = FakeModel('12345678-1234-5678-1234567890ab',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
- admin_state_up=True,
- subnets=[fake_meta_subnet],
- ports=[fake_meta_port])
-
-fake_down_network = FakeModel('12345678-dddd-dddd-1234567890ab',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
- admin_state_up=False,
- subnets=[],
- ports=[])
+fake_subnet1_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.9.2',
+ end='172.9.9.254'))
+fake_subnet1 = dhcp.DictModel(dict(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
+ network_id='12345678-1234-5678-1234567890ab',
+ cidr='172.9.9.0/24', enable_dhcp=True, name='',
+ tenant_id=fake_tenant_id,
+ gateway_ip='172.9.9.1', host_routes=[],
+ dns_nameservers=[], ip_version=4,
+ allocation_pools=fake_subnet1_allocation_pools))
+
+fake_subnet2_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.8.2',
+ end='172.9.8.254'))
+fake_subnet2 = dhcp.DictModel(dict(id='dddddddd-dddd-dddd-dddddddddddd',
+ network_id='12345678-1234-5678-1234567890ab',
+ cidr='172.9.8.0/24', enable_dhcp=False, name='',
+ tenant_id=fake_tenant_id, gateway_ip='172.9.8.1',
+ host_routes=[], dns_nameservers=[], ip_version=4,
+ allocation_pools=fake_subnet2_allocation_pools))
+
+fake_subnet3 = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
+ network_id='12345678-1234-5678-1234567890ab',
+ cidr='192.168.1.1/24', enable_dhcp=True))
+
+fake_meta_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
+ network_id='12345678-1234-5678-1234567890ab',
+ cidr='169.254.169.252/30',
+ gateway_ip='169.254.169.253',
+ enable_dhcp=True))
+
+fake_fixed_ip1 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet1.id,
+ ip_address='172.9.9.9'))
+fake_meta_fixed_ip = dhcp.DictModel(dict(id='', subnet=fake_meta_subnet,
+ ip_address='169.254.169.254'))
+fake_allocation_pool_subnet1 = dhcp.DictModel(dict(id='', start='172.9.9.2',
+ end='172.9.9.254'))
+
+fake_port1 = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
+ device_id='dhcp-12345678-1234-aaaa-1234567890ab',
+ allocation_pools=fake_subnet1_allocation_pools,
+ mac_address='aa:bb:cc:dd:ee:ff',
+ network_id='12345678-1234-5678-1234567890ab',
+ fixed_ips=[fake_fixed_ip1]))
+
+fake_port2 = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789000',
+ mac_address='aa:bb:cc:dd:ee:99',
+ network_id='12345678-1234-5678-1234567890ab',
+ fixed_ips=[]))
+
+fake_meta_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
+ mac_address='aa:bb:cc:dd:ee:ff',
+ network_id='12345678-1234-5678-1234567890ab',
+ device_owner=const.DEVICE_OWNER_ROUTER_INTF,
+ device_id='forzanapoli',
+ fixed_ips=[fake_meta_fixed_ip]))
+
+fake_network = dhcp.NetModel(True, dict(id='12345678-1234-5678-1234567890ab',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
+ admin_state_up=True,
+ subnets=[fake_subnet1, fake_subnet2],
+ ports=[fake_port1]))
+
+fake_meta_network = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
+ admin_state_up=True,
+ subnets=[fake_meta_subnet],
+ ports=[fake_meta_port]))
+
+fake_down_network = dhcp.NetModel(True,
+ dict(id='12345678-dddd-dddd-1234567890ab',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
+ admin_state_up=False,
+ subnets=[],
+ ports=[]))
class TestDhcpAgent(base.BaseTestCase):
cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS)
config.register_agent_state_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)
- cfg.CONF.register_opts(
- dhcp_agent.DeviceManager.OPTS)
cfg.CONF.register_opts(dhcp.OPTS)
cfg.CONF.register_opts(interface.OPTS)
cfg.CONF(project='neutron')
mock.call().wait()])
def test_run_completes_single_pass(self):
- with mock.patch('neutron.agent.dhcp_agent.DeviceManager'):
+ with mock.patch(DEVICE_MANAGER):
dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
attrs_to_mock = dict(
[(a, mock.DEFAULT) for a in
mocks['sync_state'].assert_called_once_with()
mocks['periodic_resync'].assert_called_once_with()
- def test_ns_name(self):
- with mock.patch('neutron.agent.dhcp_agent.DeviceManager'):
- mock_net = mock.Mock(id='foo')
- dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
- self.assertEqual(dhcp._ns_name(mock_net), 'qdhcp-foo')
-
- def test_ns_name_disabled_namespace(self):
- with mock.patch('neutron.agent.dhcp_agent.DeviceManager'):
- cfg.CONF.set_override('use_namespaces', False)
- mock_net = mock.Mock(id='foo')
- dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
- self.assertIsNone(dhcp._ns_name(mock_net))
-
def test_call_driver(self):
network = mock.Mock()
network.id = '1'
- with mock.patch('neutron.agent.dhcp_agent.DeviceManager') as dev_mgr:
- dhcp = dhcp_agent.DhcpAgent(cfg.CONF)
- self.assertTrue(dhcp.call_driver('foo', network))
- self.assertTrue(dev_mgr.called)
- self.driver.assert_called_once_with(cfg.CONF,
- mock.ANY,
- 'sudo',
- mock.ANY,
- 'qdhcp-1',
- mock.ANY)
+ dhcp = dhcp_agent.DhcpAgent(cfg.CONF)
+ self.assertTrue(dhcp.call_driver('foo', network))
+ self.driver.assert_called_once_with(cfg.CONF,
+ mock.ANY,
+ 'sudo',
+ mock.ANY,
+ mock.ANY)
def test_call_driver_failure(self):
network = mock.Mock()
network.id = '1'
self.driver.return_value.foo.side_effect = Exception
- with mock.patch('neutron.agent.dhcp_agent.DeviceManager') as dev_mgr:
- with mock.patch.object(dhcp_agent.LOG, 'exception') as log:
- dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
- self.assertIsNone(dhcp.call_driver('foo', network))
- self.assertTrue(dev_mgr.called)
- self.driver.assert_called_once_with(cfg.CONF,
- mock.ANY,
- 'sudo',
- mock.ANY,
- 'qdhcp-1',
- mock.ANY)
- self.assertEqual(log.call_count, 1)
- self.assertTrue(dhcp.needs_resync)
+ with mock.patch.object(dhcp_agent.LOG, 'exception') as log:
+ dhcp = dhcp_agent.DhcpAgent(HOSTNAME)
+ self.assertIsNone(dhcp.call_driver('foo', network))
+ self.driver.assert_called_once_with(cfg.CONF,
+ mock.ANY,
+ 'sudo',
+ mock.ANY,
+ mock.ANY)
+ self.assertEqual(log.call_count, 1)
+ self.assertTrue(dhcp.needs_resync)
def _test_sync_state_helper(self, known_networks, active_networks):
- with mock.patch('neutron.agent.dhcp_agent.DhcpPluginApi') as plug:
+ with mock.patch(DHCP_PLUGIN) as plug:
mock_plugin = mock.Mock()
mock_plugin.get_active_networks_info.return_value = active_networks
plug.return_value = mock_plugin
self._test_sync_state_helper(['b'], ['a'])
def test_sync_state_plugin_error(self):
- with mock.patch('neutron.agent.dhcp_agent.DhcpPluginApi') as plug:
+ with mock.patch(DHCP_PLUGIN) as plug:
mock_plugin = mock.Mock()
mock_plugin.get_active_networks_info.side_effect = Exception
plug.return_value = mock_plugin
'log_file': None,
'use_syslog': True,
'syslog_log_facility': 'LOG_USER'}
- conf = dhcp_agent.DictModel(conf_dict)
+ conf = dhcp.DictModel(conf_dict)
expected_args = ['--debug',
'--use-syslog',
'--syslog-log-facility=LOG_USER']
'log_file': None,
'use_syslog': False,
'syslog_log_facility': 'LOG_USER'}
- conf = dhcp_agent.DictModel(conf_dict)
+ conf = dhcp.DictModel(conf_dict)
expected_args = ['--debug',
'--verbose',
'--log-file=log_file_name',
'log_file': 'tests/filelog',
'use_syslog': False,
'syslog_log_facility': 'LOG_USER'}
- conf = dhcp_agent.DictModel(conf_dict)
+ conf = dhcp.DictModel(conf_dict)
expected_args = ['--debug',
'--log-file=log_file_name',
'--log-dir=/etc/tests/tests']
'log_dir': None,
'use_syslog': False,
'syslog_log_facility': 'LOG_USER'}
- conf = dhcp_agent.DictModel(conf_dict)
+ conf = dhcp.DictModel(conf_dict)
expected_args = ['--debug',
'--log-file=log_file_name',
'--log-dir=tests']
'log_dir': '/etc/tests',
'use_syslog': True,
'syslog_log_facility': 'LOG_USER'}
- conf = dhcp_agent.DictModel(conf_dict)
+ conf = dhcp.DictModel(conf_dict)
expected_args = ['--debug',
'--verbose',
'--log-file=log_file_name',
class TestDhcpAgentEventHandler(base.BaseTestCase):
def setUp(self):
super(TestDhcpAgentEventHandler, self).setUp()
- cfg.CONF.register_opts(dhcp_agent.DeviceManager.OPTS)
cfg.CONF.register_opts(dhcp.OPTS)
cfg.CONF.set_override('interface_driver',
'neutron.agent.linux.interface.NullDriver')
config.register_root_helper(cfg.CONF)
cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS)
- self.plugin_p = mock.patch('neutron.agent.dhcp_agent.DhcpPluginApi')
+ self.plugin_p = mock.patch(DHCP_PLUGIN)
plugin_cls = self.plugin_p.start()
self.plugin = mock.Mock()
plugin_cls.return_value = self.plugin
disable.assertCalledOnceWith(fake_network.id)
def test_refresh_dhcp_helper_no_dhcp_enabled_networks(self):
- network = FakeModel('net-id',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
- admin_state_up=True,
- subnets=[],
- ports=[])
+ network = dhcp.NetModel(True, dict(id='net-id',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
+ admin_state_up=True,
+ subnets=[],
+ ports=[]))
self.cache.get_network_by_id.return_value = network
self.plugin.get_network_info.return_value = network
[mock.call.get_network_by_id('net-id')])
def test_refresh_dhcp_helper_exception_during_rpc(self):
- network = FakeModel('net-id',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
- admin_state_up=True,
- subnets=[],
- ports=[])
+ network = dhcp.NetModel(True, dict(id='net-id',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
+ admin_state_up=True,
+ subnets=[],
+ ports=[]))
self.cache.get_network_by_id.return_value = network
self.plugin.get_network_info.side_effect = Exception
payload = dict(subnet=dict(network_id=fake_network.id))
self.cache.get_network_by_id.return_value = fake_network
self.plugin.get_network_info.return_value = fake_network
- self.dhcp.device_manager.update = mock.Mock()
self.dhcp.subnet_update_end(None, payload)
self.cache.assert_has_calls([mock.call.put(fake_network)])
self.call_driver.assert_called_once_with('reload_allocations',
fake_network)
- self.dhcp.device_manager.update.assert_called_once_with(fake_network)
def test_subnet_update_end_restart(self):
- new_state = FakeModel(fake_network.id,
- tenant_id=fake_network.tenant_id,
- admin_state_up=True,
- subnets=[fake_subnet1, fake_subnet3],
- ports=[fake_port1])
+ new_state = dhcp.NetModel(True, dict(id=fake_network.id,
+ tenant_id=fake_network.tenant_id,
+ admin_state_up=True,
+ subnets=[fake_subnet1, fake_subnet3],
+ ports=[fake_port1]))
payload = dict(subnet=dict(network_id=fake_network.id))
self.cache.get_network_by_id.return_value = fake_network
self.plugin.get_network_info.return_value = new_state
- self.dhcp.device_manager.update = mock.Mock()
self.dhcp.subnet_update_end(None, payload)
self.cache.assert_has_calls([mock.call.put(new_state)])
self.call_driver.assert_called_once_with('restart',
new_state)
- self.dhcp.device_manager.update.assert_called_once_with(new_state)
def test_subnet_update_end_delete_payload(self):
- prev_state = FakeModel(fake_network.id,
- tenant_id=fake_network.tenant_id,
- admin_state_up=True,
- subnets=[fake_subnet1, fake_subnet3],
- ports=[fake_port1])
+ prev_state = dhcp.NetModel(True, dict(id=fake_network.id,
+ tenant_id=fake_network.tenant_id,
+ admin_state_up=True,
+ subnets=[fake_subnet1, fake_subnet3],
+ ports=[fake_port1]))
payload = dict(subnet_id=fake_subnet1.id)
self.cache.get_network_by_subnet_id.return_value = prev_state
self.cache.get_network_by_id.return_value = prev_state
self.plugin.get_network_info.return_value = fake_network
- self.dhcp.device_manager.update = mock.Mock()
self.dhcp.subnet_delete_end(None, payload)
mock.call.put(fake_network)])
self.call_driver.assert_called_once_with('restart',
fake_network)
- self.dhcp.device_manager.update.assert_called_once_with(fake_network)
def test_port_update_end(self):
payload = dict(port=vars(fake_port2))
class TestDhcpPluginApiProxy(base.BaseTestCase):
def setUp(self):
super(TestDhcpPluginApiProxy, self).setUp()
- self.proxy = dhcp_agent.DhcpPluginApi('foo', {})
+ self.proxy = dhcp_agent.DhcpPluginApi('foo', {}, None)
self.proxy.host = 'foo'
self.call_p = mock.patch.object(self.proxy, 'call')
fake_network)
def test_put_port(self):
- fake_network = FakeModel('12345678-1234-5678-1234567890ab',
+ fake_net = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
subnets=[fake_subnet1],
- ports=[fake_port1])
+ ports=[fake_port1]))
nc = dhcp_agent.NetworkCache()
- nc.put(fake_network)
+ nc.put(fake_net)
nc.put_port(fake_port2)
self.assertEqual(len(nc.port_lookup), 2)
- self.assertIn(fake_port2, fake_network.ports)
+ self.assertIn(fake_port2, fake_net.ports)
def test_put_port_existing(self):
- fake_network = FakeModel('12345678-1234-5678-1234567890ab',
+ fake_net = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
subnets=[fake_subnet1],
- ports=[fake_port1, fake_port2])
+ ports=[fake_port1, fake_port2]))
nc = dhcp_agent.NetworkCache()
- nc.put(fake_network)
+ nc.put(fake_net)
nc.put_port(fake_port2)
self.assertEqual(len(nc.port_lookup), 2)
- self.assertIn(fake_port2, fake_network.ports)
+ self.assertIn(fake_port2, fake_net.ports)
def test_remove_port_existing(self):
- fake_network = FakeModel('12345678-1234-5678-1234567890ab',
+ fake_net = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa',
subnets=[fake_subnet1],
- ports=[fake_port1, fake_port2])
-
+ ports=[fake_port1, fake_port2]))
nc = dhcp_agent.NetworkCache()
- nc.put(fake_network)
+ nc.put(fake_net)
nc.remove_port(fake_port2)
self.assertEqual(len(nc.port_lookup), 1)
- self.assertNotIn(fake_port2, fake_network.ports)
+ self.assertNotIn(fake_port2, fake_net.ports)
def test_get_port_by_id(self):
nc = dhcp_agent.NetworkCache()
class TestDeviceManager(base.BaseTestCase):
def setUp(self):
super(TestDeviceManager, self).setUp()
- cfg.CONF.register_opts(dhcp_agent.DeviceManager.OPTS)
cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS)
+ cfg.CONF.register_opts(dhcp.OPTS)
cfg.CONF.set_override('interface_driver',
'neutron.agent.linux.interface.NullDriver')
config.register_root_helper(cfg.CONF)
self.device_exists.return_value = device_exists
self.mock_driver.get_device_name.return_value = 'tap12345678-12'
- dh = dhcp_agent.DeviceManager(cfg.CONF, plugin)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin)
dh._set_default_route = mock.Mock()
interface_name = dh.setup(net, reuse_existing)
[{'subnet_id': fake_fixed_ip1.subnet_id}],
'device_id': mock.ANY}})])
- namespace = dhcp_agent.NS_PREFIX + net.id
-
expected_ips = ['172.9.9.9/24', '169.254.169.254/16']
- expected = [mock.call.init_l3('tap12345678-12',
- expected_ips,
- namespace=namespace)]
+ expected = [
+ mock.call.get_device_name(port),
+ mock.call.init_l3(
+ 'tap12345678-12',
+ expected_ips,
+ namespace=net.namespace)]
if not reuse_existing:
- expected.insert(0,
+ expected.insert(1,
mock.call.plug(net.id,
port.id,
'tap12345678-12',
'aa:bb:cc:dd:ee:ff',
- namespace=namespace))
+ namespace=net.namespace))
self.mock_driver.assert_has_calls(expected)
dh._set_default_route.assert_called_once_with(net)
def test_create_dhcp_port_create_new(self):
plugin = mock.Mock()
- dh = dhcp_agent.DeviceManager(cfg.CONF, plugin)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin)
plugin.create_dhcp_port.return_value = fake_network.ports[0]
dh.setup_dhcp_port(fake_network)
plugin.assert_has_calls([
def test_create_dhcp_port_update_add_subnet(self):
plugin = mock.Mock()
- dh = dhcp_agent.DeviceManager(cfg.CONF, plugin)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin)
fake_network_copy = copy.deepcopy(fake_network)
fake_network_copy.ports[0].device_id = dh.get_device_id(fake_network)
fake_network_copy.subnets[1].enable_dhcp = True
def test_create_dhcp_port_no_update_or_create(self):
plugin = mock.Mock()
- dh = dhcp_agent.DeviceManager(cfg.CONF, plugin)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin)
fake_network_copy = copy.deepcopy(fake_network)
fake_network_copy.ports[0].device_id = dh.get_device_id(fake_network)
dh.setup_dhcp_port(fake_network_copy)
self.assertFalse(plugin.update_dhcp_port.called)
def test_destroy(self):
- fake_network = FakeModel('12345678-1234-5678-1234567890ab',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa')
+ fake_net = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'))
- fake_port = FakeModel('12345678-1234-aaaa-1234567890ab',
- mac_address='aa:bb:cc:dd:ee:ff')
+ fake_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
+ mac_address='aa:bb:cc:dd:ee:ff'))
with mock.patch('neutron.agent.linux.interface.NullDriver') as dvr_cls:
mock_driver = mock.MagicMock()
plugin = mock.Mock()
plugin.get_dhcp_port.return_value = fake_port
- dh = dhcp_agent.DeviceManager(cfg.CONF, plugin)
- dh.destroy(fake_network, 'tap12345678-12')
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin)
+ dh.destroy(fake_net, 'tap12345678-12')
dvr_cls.assert_called_once_with(cfg.CONF)
mock_driver.assert_has_calls(
[mock.call.unplug('tap12345678-12',
- namespace='qdhcp-' + fake_network.id)])
+ namespace='qdhcp-' + fake_net.id)])
plugin.assert_has_calls(
- [mock.call.release_dhcp_port(fake_network.id, mock.ANY)])
+ [mock.call.release_dhcp_port(fake_net.id, mock.ANY)])
def test_get_interface_name(self):
- fake_network = FakeModel('12345678-1234-5678-1234567890ab',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa')
+ fake_net = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'))
- fake_port = FakeModel('12345678-1234-aaaa-1234567890ab',
- mac_address='aa:bb:cc:dd:ee:ff')
+ fake_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
+ mac_address='aa:bb:cc:dd:ee:ff'))
with mock.patch('neutron.agent.linux.interface.NullDriver') as dvr_cls:
mock_driver = mock.MagicMock()
plugin = mock.Mock()
plugin.get_dhcp_port.return_value = fake_port
- dh = dhcp_agent.DeviceManager(cfg.CONF, plugin)
- dh.get_interface_name(fake_network, fake_port)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, plugin)
+ dh.get_interface_name(fake_net, fake_port)
dvr_cls.assert_called_once_with(cfg.CONF)
mock_driver.assert_has_calls(
self.assertEqual(len(plugin.mock_calls), 0)
def test_get_device_id(self):
- fake_network = FakeModel('12345678-1234-5678-1234567890ab',
- tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa')
+ fake_net = dhcp.NetModel(True,
+ dict(id='12345678-1234-5678-1234567890ab',
+ tenant_id='aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'))
expected = ('dhcp1ae5f96c-c527-5079-82ea-371a01645457-12345678-1234-'
'5678-1234567890ab')
uuid5.return_value = '1ae5f96c-c527-5079-82ea-371a01645457'
get_host.return_value = 'localhost'
- dh = dhcp_agent.DeviceManager(cfg.CONF, None)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None)
uuid5.called_once_with(uuid.NAMESPACE_DNS, 'localhost')
- self.assertEqual(dh.get_device_id(fake_network), expected)
+ self.assertEqual(dh.get_device_id(fake_net), expected)
def _get_device_manager_with_mock_device(self, conf, device):
- dh = dhcp_agent.DeviceManager(conf, None)
+ dh = dhcp.DeviceManager(conf, cfg.CONF.root_helper, None)
dh._get_device = mock.Mock(return_value=device)
return dh
# Try with namespaces and no metadata network
cfg.CONF.set_override('use_namespaces', True)
cfg.CONF.set_override('enable_metadata_network', False)
- dh = dhcp_agent.DeviceManager(cfg.CONF, None)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None)
dh._set_default_route = mock.Mock()
dh.update(True)
# No namespaces, shouldn't set default route.
cfg.CONF.set_override('use_namespaces', False)
cfg.CONF.set_override('enable_metadata_network', False)
- dh = dhcp_agent.DeviceManager(cfg.CONF, None)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None)
dh._set_default_route = mock.Mock()
dh.update(FakeV4Network())
# Meta data network enabled, don't interfere with its gateway.
cfg.CONF.set_override('use_namespaces', True)
cfg.CONF.set_override('enable_metadata_network', True)
- dh = dhcp_agent.DeviceManager(cfg.CONF, None)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None)
dh._set_default_route = mock.Mock()
dh.update(FakeV4Network())
# For completeness
cfg.CONF.set_override('use_namespaces', False)
cfg.CONF.set_override('enable_metadata_network', True)
- dh = dhcp_agent.DeviceManager(cfg.CONF, None)
+ dh = dhcp.DeviceManager(cfg.CONF, cfg.CONF.root_helper, None)
dh._set_default_route = mock.Mock()
dh.update(FakeV4Network())
def test_basic_dict(self):
d = dict(a=1, b=2)
- m = dhcp_agent.DictModel(d)
+ m = dhcp.DictModel(d)
self.assertEqual(m.a, 1)
self.assertEqual(m.b, 2)
def test_dict_has_sub_dict(self):
d = dict(a=dict(b=2))
- m = dhcp_agent.DictModel(d)
+ m = dhcp.DictModel(d)
self.assertEqual(m.a.b, 2)
def test_dict_contains_list(self):
d = dict(a=[1, 2])
- m = dhcp_agent.DictModel(d)
+ m = dhcp.DictModel(d)
self.assertEqual(m.a, [1, 2])
def test_dict_contains_list_of_dicts(self):
d = dict(a=[dict(b=2), dict(c=3)])
- m = dhcp_agent.DictModel(d)
+ m = dhcp.DictModel(d)
self.assertEqual(m.a[0].b, 2)
self.assertEqual(m.a[1].c, 3)
+
+
+class TestNetModel(base.BaseTestCase):
+ def test_ns_name(self):
+ network = dhcp.NetModel(True, {'id': 'foo'})
+ self.assertEqual(network.namespace, 'qdhcp-foo')
+
+ def test_ns_name_false_namespace(self):
+ network = dhcp.NetModel(False, {'id': 'foo'})
+ self.assertIsNone(network.namespace)
+
+ def test_ns_name_none_namespace(self):
+ network = dhcp.NetModel(None, {'id': 'foo'})
+ self.assertIsNone(network.namespace)
id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
subnets = [FakeV4Subnet(), FakeV6Subnet()]
ports = [FakePort1(), FakePort2(), FakePort3()]
+ namespace = 'qdhcp-ns'
class FakeDualNetworkGatewayRoute:
id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
subnets = [FakeV4SubnetGatewayRoute(), FakeV6Subnet()]
ports = [FakePort1(), FakePort2(), FakePort3()]
+ namespace = 'qdhcp-ns'
class FakeDualNetworkSingleDHCP:
id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
subnets = [FakeV4Subnet(), FakeV4SubnetNoDHCP()]
ports = [FakePort1(), FakePort2(), FakePort3()]
+ namespace = 'qdhcp-ns'
class FakeV4NoGatewayNetwork:
ports = [FakePort1()]
-class TestDhcpBase(base.BaseTestCase):
- def test_existing_dhcp_networks_abstract_error(self):
- self.assertRaises(NotImplementedError,
- dhcp.DhcpBase.existing_dhcp_networks,
- None, None)
-
- def test_check_version_abstract_error(self):
- self.assertRaises(NotImplementedError,
- dhcp.DhcpBase.check_version)
-
- def test_base_abc_error(self):
- self.assertRaises(TypeError, dhcp.DhcpBase, None)
-
- def test_restart(self):
- class SubClass(dhcp.DhcpBase):
- def __init__(self):
- dhcp.DhcpBase.__init__(self, None, None, None)
- self.called = []
-
- def enable(self):
- self.called.append('enable')
-
- def disable(self, retain_port=False):
- self.called.append('disable %s' % retain_port)
-
- def reload_allocations(self):
- pass
-
- def release_lease(self):
- pass
-
- @property
- def active(self):
- return True
-
- c = SubClass()
- c.restart()
- self.assertEqual(c.called, ['disable True', 'enable'])
-
-
class LocalChild(dhcp.DhcpLocalProcess):
PORTS = {4: [4], 6: [6]}
self.conf = config.setup_conf()
self.conf.register_opts(base_config.core_opts)
self.conf.register_opts(dhcp.OPTS)
+ instance = mock.patch("neutron.agent.linux.dhcp.DeviceManager")
+ self.mock_mgr = instance.start()
+ self.addCleanup(self.mock_mgr.stop)
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=True))
self.conf(args=args)
self.execute = self.execute_p.start()
+class TestDhcpBase(TestBase):
+
+ def test_existing_dhcp_networks_abstract_error(self):
+ self.assertRaises(NotImplementedError,
+ dhcp.DhcpBase.existing_dhcp_networks,
+ None, None)
+
+ def test_check_version_abstract_error(self):
+ self.assertRaises(NotImplementedError,
+ dhcp.DhcpBase.check_version)
+
+ def test_base_abc_error(self):
+ self.assertRaises(TypeError, dhcp.DhcpBase, None)
+
+ def test_restart(self):
+ class SubClass(dhcp.DhcpBase):
+ def __init__(self):
+ dhcp.DhcpBase.__init__(self, cfg.CONF, FakeV4Network(), None)
+ self.called = []
+
+ def enable(self):
+ self.called.append('enable')
+
+ def disable(self, retain_port=False):
+ self.called.append('disable %s' % retain_port)
+
+ def reload_allocations(self):
+ pass
+
+ def release_lease(self):
+ pass
+
+ @property
+ def active(self):
+ return True
+
+ c = SubClass()
+ c.restart()
+ self.assertEqual(c.called, ['disable True', 'enable'])
+
+
class TestDhcpLocalProcess(TestBase):
def test_active(self):
dummy_cmd_line = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
self.assertTrue(makedirs.called)
def test_enable_already_active(self):
- delegate = mock.Mock()
- delegate.setup.return_value = 'tap0'
with mock.patch.object(LocalChild, 'active') as patched:
patched.__get__ = mock.Mock(return_value=True)
- lp = LocalChild(self.conf, FakeV4Network(),
- device_delegate=delegate)
+ lp = LocalChild(self.conf, FakeV4Network())
lp.enable()
self.assertEqual(lp.called, ['restart'])
def test_enable(self):
- delegate = mock.Mock(return_value='tap0')
attrs_to_mock = dict(
[(a, mock.DEFAULT) for a in
['active', 'get_conf_file_name', 'interface_name']]
mocks['get_conf_file_name'].return_value = '/dir'
mocks['interface_name'].__set__ = mock.Mock()
lp = LocalChild(self.conf,
- FakeDualNetwork(),
- device_delegate=delegate)
+ FakeDualNetwork())
lp.enable()
- delegate.assert_has_calls(
- [mock.call.setup(mock.ANY, reuse_existing=True)])
+ self.mock_mgr.assert_has_calls(
+ [mock.call(self.conf, 'sudo', None),
+ mock.call().setup(mock.ANY, reuse_existing=True)])
self.assertEqual(lp.called, ['spawn'])
self.assertTrue(mocks['interface_name'].__set__.called)
def test_disable_retain_port(self):
attrs_to_mock = dict([(a, mock.DEFAULT) for a in
['active', 'interface_name', 'pid']])
- delegate = mock.Mock()
network = FakeDualNetwork()
with mock.patch.multiple(LocalChild, **attrs_to_mock) as mocks:
mocks['active'].__get__ = mock.Mock(return_value=True)
mocks['pid'].__get__ = mock.Mock(return_value=5)
mocks['interface_name'].__get__ = mock.Mock(return_value='tap0')
- lp = LocalChild(self.conf, network, device_delegate=delegate,
- namespace='qdhcp-ns')
+ lp = LocalChild(self.conf, network)
lp.disable(retain_port=True)
- self.assertFalse(delegate.called)
- exp_args = ['kill', '-9', 5]
- self.execute.assert_called_once_with(exp_args, 'sudo')
+ exp_args = ['kill', '-9', 5]
+ self.execute.assert_called_once_with(exp_args, 'sudo')
def test_disable(self):
attrs_to_mock = dict([(a, mock.DEFAULT) for a in
['active', 'interface_name', 'pid']])
- delegate = mock.Mock()
network = FakeDualNetwork()
with mock.patch.multiple(LocalChild, **attrs_to_mock) as mocks:
mocks['active'].__get__ = mock.Mock(return_value=True)
mocks['pid'].__get__ = mock.Mock(return_value=5)
mocks['interface_name'].__get__ = mock.Mock(return_value='tap0')
- lp = LocalChild(self.conf, network, device_delegate=delegate,
- namespace='qdhcp-ns')
+ lp = LocalChild(self.conf, network)
lp.disable()
- delegate.assert_has_calls([mock.call.destroy(network, 'tap0')])
+ self.mock_mgr.assert_has_calls([mock.call(self.conf, 'sudo', None),
+ mock.call().destroy(network, 'tap0')])
exp_args = ['kill', '-9', 5]
self.execute.assert_called_once_with(exp_args, 'sudo')
expected.extend(extra_options)
self.execute.return_value = ('', '')
- delegate = mock.Mock()
- delegate.get_interface_name.return_value = 'tap0'
attrs_to_mock = dict(
[(a, mock.DEFAULT) for a in
with mock.patch.object(dhcp.sys, 'argv') as argv:
argv.__getitem__.side_effect = fake_argv
dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(),
- device_delegate=delegate,
- namespace='qdhcp-ns',
version=float(2.59))
dm.spawn_process()
self.assertTrue(mocks['_output_opts_file'].called)
self.safe.assert_called_once_with('/foo/opts', expected)
def test_release_lease(self):
- dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(), namespace='qdhcp-ns',
- version=float(2.59))
+ dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(), version=float(2.59))
dm.release_lease(mac_address=FakePort2.mac_address,
removed_ips=[FakePort2.fixed_ips[0].ip_address])
exp_args = ['ip', 'netns', 'exec', 'qdhcp-ns', 'dhcp_release',
with mock.patch.object(dhcp.Dnsmasq, 'pid') as pid:
pid.__get__ = mock.Mock(return_value=5)
dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(),
- namespace='qdhcp-ns',
version=float(2.59))
method_name = '_make_subnet_interface_ip_map'
with mock.patch.object(dhcp.Dnsmasq, 'pid') as pid:
pid.__get__ = mock.Mock(return_value=5)
dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(),
- namespace='qdhcp-ns', version=float(2.59))
+ version=float(2.59))
method_name = '_make_subnet_interface_ip_map'
with mock.patch.object(dhcp.Dnsmasq, method_name) as ip_map:
]
dm = dhcp.Dnsmasq(self.conf,
- FakeDualNetwork(),
- namespace='qdhcp-ns')
+ FakeDualNetwork())
self.assertEqual(
dm._make_subnet_interface_ip_map(),
import mock
from neutron.agent.common import config
-from neutron.agent.dhcp_agent import DeviceManager
+from neutron.agent.linux import dhcp
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
class TestMetaInterfaceDriver(TestBase):
def setUp(self):
super(TestMetaInterfaceDriver, self).setUp()
- self.conf.register_opts(DeviceManager.OPTS)
+ self.conf.register_opts(dhcp.OPTS)
self.client_cls_p = mock.patch('neutronclient.v2_0.client.Client')
client_cls = self.client_cls_p.start()
self.addCleanup(self.client_cls_p.stop)
Code Review / openstack-build / neutron-build.git / commitdiff