--- /dev/null
+#!/usr/bin/env python
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2012 Openstack, LLC.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from quantum.agent.linux import dhcp
+dhcp.Dnsmasq.lease_update()
# under the License.
import logging
+import os
+import re
import socket
import sys
import uuid
from quantum.agent.linux import dhcp
from quantum.agent.linux import interface
from quantum.agent.linux import ip_lib
+from quantum.api.v2 import attributes
from quantum.common import exceptions
from quantum.common import topics
from quantum.openstack.common import cfg
from quantum.openstack.common import context
from quantum.openstack.common import importutils
+from quantum.openstack.common import jsonutils
from quantum.openstack.common.rpc import proxy
from quantum.version import version_string
self.device_manager = DeviceManager(self.conf, self.plugin_rpc)
self.notifications = agent_rpc.NotificationDispatcher()
+ self.lease_relay = DhcpLeaseRelay(self.update_lease)
def run(self):
"""Activate the DHCP agent."""
for network_id in self.plugin_rpc.get_active_networks():
self.enable_dhcp_helper(network_id)
+ self.lease_relay.start()
self.notifications.run_dispatch(self)
def call_driver(self, action, network):
except Exception, e:
LOG.warn('Unable to %s dhcp. Exception: %s' % (action, e))
+ def update_lease(self, network_id, ip_address, time_remaining):
+ self.plugin_rpc.update_lease_expiration(network_id, ip_address,
+ time_remaining)
+
def enable_dhcp_helper(self, network_id):
"""Enable DHCP for a network that meets enabling criteria."""
network = self.plugin_rpc.get_network_info(network_id)
host=self.host),
topic=self.topic)
+ def update_lease_expiration(self, network_id, ip_address, lease_remaining):
+ """Make a remote process call to update the ip lease expiration."""
+ self.cast(self.context,
+ self.make_msg('update_lease_expiration',
+ network_id=network_id,
+ ip_address=ip_address,
+ lease_remaining=lease_remaining,
+ host=self.host),
+ topic=self.topic)
+
class NetworkCache(object):
"""Agent cache of the current network state."""
setattr(self, key, value)
+class DhcpLeaseRelay(object):
+ """UNIX domain socket server for processing lease updates.
+
+ Network namespace isolation prevents the DHCP process from notifying
+ Quantum directly. This class works around the limitation by using the
+ domain socket to pass the information. This class handles message.
+ receiving and then calls the callback method.
+ """
+
+ OPTS = [
+ cfg.StrOpt('dhcp_lease_relay_socket',
+ default='$state_path/dhcp/lease_relay',
+ help='Location to DHCP lease relay UNIX domain socket')
+ ]
+
+ def __init__(self, lease_update_callback):
+ self.callback = lease_update_callback
+
+ try:
+ os.unlink(cfg.CONF.dhcp_lease_relay_socket)
+ except OSError:
+ if os.path.exists(cfg.CONF.dhcp_lease_relay_socket):
+ raise
+
+ def _validate_field(self, value, regex):
+ """Validate value against a regular expression and return if valid."""
+ match = re.match(regex, value)
+
+ if match:
+ return value
+ raise ValueError(_("Value %s does not match regex: %s") %
+ (value, regex))
+
+ def _handler(self, client_sock, client_addr):
+ """Handle incoming lease relay stream connection.
+
+ This method will only read the first 1024 bytes and then close the
+ connection. The limit exists to limit the impact of misbehaving
+ clients.
+ """
+ try:
+ msg = client_sock.recv(1024)
+ data = jsonutils.loads(msg)
+ client_sock.close()
+
+ network_id = self._validate_field(data['network_id'],
+ attributes.UUID_PATTERN)
+ ip_address = str(netaddr.IPAddress(data['ip_address']))
+ lease_remaining = int(data['lease_remaining'])
+ self.callback(network_id, ip_address, lease_remaining)
+ except ValueError, e:
+ LOG.warn(_('Unable to parse lease relay msg to dict.'))
+ LOG.warn(_('Exception value: %s') % e)
+ LOG.warn(_('Message representation: %s') % repr(msg))
+ except Exception, e:
+ LOG.exception(_('Unable update lease. Exception'))
+
+ def start(self):
+ """Spawn a green thread to run the lease relay unix socket server."""
+ listener = eventlet.listen(cfg.CONF.dhcp_lease_relay_socket,
+ family=socket.AF_UNIX)
+ eventlet.spawn(eventlet.serve, listener, self._handler)
+
+
def main():
eventlet.monkey_patch()
cfg.CONF.register_opts(DhcpAgent.OPTS)
cfg.CONF.register_opts(DeviceManager.OPTS)
+ cfg.CONF.register_opts(DhcpLeaseRelay.OPTS)
cfg.CONF.register_opts(dhcp.OPTS)
cfg.CONF.register_opts(interface.OPTS)
cfg.CONF(args=sys.argv, project='quantum')
import logging
import os
import re
+import socket
import StringIO
+import sys
import tempfile
+import textwrap
import netaddr
from quantum.agent.linux import utils
from quantum.openstack.common import cfg
from quantum.openstack.common import importutils
+from quantum.openstack.common import jsonutils
LOG = logging.getLogger(__name__)
_TAG_PREFIX = 'tag%d'
+ QUANTUM_NETWORK_ID_KEY = 'QUANTUM_NETWORK_ID'
+ QUANTUM_RELAY_SOCKET_PATH_KEY = 'QUANTUM_RELAY_SOCKET_PATH'
+
def spawn_process(self):
"""Spawns a Dnsmasq process for the network."""
interface_name = self.device_delegate.get_interface_name(self.network)
+
+ env = {
+ self.QUANTUM_NETWORK_ID_KEY: self.network.id,
+ self.QUANTUM_RELAY_SOCKET_PATH_KEY:
+ self.conf.dhcp_lease_relay_socket
+ }
+
cmd = [
- # TODO (mark): this is dhcpbridge script we'll need to know
- # when an IP address has been released
'dnsmasq',
'--no-hosts',
'--no-resolv',
#'--dhcp-lease-max=%s' % ?,
'--dhcp-hostsfile=%s' % self._output_hosts_file(),
'--dhcp-optsfile=%s' % self._output_opts_file(),
+ '--dhcp-script=%s' % self._lease_relay_script_path(),
'--leasefile-ro',
]
if self.conf.use_namespaces:
ip_wrapper = ip_lib.IPWrapper(self.root_helper,
namespace=self.network.id)
- ip_wrapper.netns.execute(cmd)
+ ip_wrapper.netns.execute(cmd, addl_env=env)
else:
+ # For normal sudo prepend the env vars before command
+ cmd = ['%s=%s' % pair for pair in env.items()] + cmd
utils.execute(cmd, self.root_helper)
def reload_allocations(self):
replace_file(name, '\n'.join(['tag:%s,%s:%s,%s' % o for o in options]))
return name
+ def _lease_relay_script_path(self):
+ return os.path.join(os.path.dirname(sys.argv[0]),
+ 'quantum-dhcp-agent-dnsmasq-lease-update')
+
+ @classmethod
+ def lease_update(cls):
+ network_id = os.environ.get(cls.QUANTUM_NETWORK_ID_KEY)
+ dhcp_relay_socket = os.environ.get(cls.QUANTUM_RELAY_SOCKET_PATH_KEY)
+
+ action = sys.argv[1]
+ if action not in ('add', 'del', 'old'):
+ sys.exit()
+
+ mac_address = sys.argv[2]
+ ip_address = sys.argv[3]
+
+ if action == 'del':
+ lease_remaining = 0
+ else:
+ lease_remaining = int(os.environ.get('DNSMASQ_TIME_REMAINING', 0))
+
+ data = dict(network_id=network_id, mac_address=mac_address,
+ ip_address=ip_address, lease_remaining=lease_remaining)
+
+ if os.path.exists(dhcp_relay_socket):
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ sock.connect(dhcp_relay_socket)
+ sock.send(jsonutils.dumps(data))
+ sock.close()
+
def replace_file(file_name, data):
"""Replaces the contents of file_name with data in a safe manner.
['ip', 'netns', 'delete', name],
root_helper=self._parent.root_helper)
- def execute(self, cmds):
+ def execute(self, cmds, addl_env={}):
if not self._parent.root_helper:
raise exceptions.SudoRequired()
elif not self._parent.namespace:
raise Exception(_('No namespace defined for parent'))
else:
return utils.execute(
+ ['%s=%s' % pair for pair in addl_env.items()] +
['ip', 'netns', 'exec', self._parent.namespace] + list(cmds),
root_helper=self._parent.root_helper)
return (timeutils.utcnow() +
datetime.timedelta(seconds=cfg.CONF.dhcp_lease_duration))
+ def update_fixed_ip_lease_expiration(self, context, network_id,
+ ip_address, lease_remaining):
+
+ expiration = timeutils.utcnow() + datetime.timedelta(lease_remaining)
+
+ query = context.session.query(models_v2.IPAllocation)
+ query = query.filter_by(network_id=network_id, ip_address=ip_address)
+
+ try:
+ fixed_ip = query.one()
+ fixed_ip.expiration = expiration
+ except exc.NoResultFound:
+ LOG.debug("No fixed IP found that matches the network %s and "
+ "ip address %s.", network_id, ip_address)
+
@staticmethod
def _delete_ip_allocation(context, network_id, subnet_id, port_id,
ip_address):
network_id=port['network_id'],
port_id=port.id,
ip_address=ip['ip_address'],
- subnet_id=ip['subnet_id'])
+ subnet_id=ip['subnet_id'],
+ expiration=self._default_allocation_expiration()
+ )
context.session.add(allocated)
return self._make_port_dict(port)
del fixed_ips[i]
break
plugin.update_port(context, port['id'], dict(port=port))
+
+ def update_lease_expiration(self, context, **kwargs):
+ """Release the fixed_ip associated the subnet on a port."""
+ host = kwargs.get('host')
+ network_id = kwargs.get('network_id')
+ ip_address = kwargs.get('ip_address')
+ lease_remaining = kwargs.get('lease_remaining')
+
+ LOG.debug('Updating lease expiration for %s on network %s from %s.',
+ ip_address, network_id, host)
+
+ context = augment_context(context)
+ plugin = manager.QuantumManager.get_plugin()
+
+ plugin.update_fixed_ip_lease_expiration(context, network_id,
+ ip_address, lease_remaining)
from quantum import context
from quantum.db import api as db
from quantum.db import db_base_plugin_v2
+from quantum.db import models_v2
from quantum.manager import QuantumManager
from quantum.openstack.common import cfg
from quantum.openstack.common import timeutils
res = port_req.get_response(self.api)
self.assertEquals(res.status_int, 422)
+ def test_default_allocation_expiration(self):
+ reference = datetime.datetime(2012, 8, 13, 23, 11, 0)
+ timeutils.utcnow.override_time = reference
+
+ cfg.CONF.set_override('dhcp_lease_duration', 120)
+ expires = QuantumManager.get_plugin()._default_allocation_expiration()
+ timeutils.utcnow
+ cfg.CONF.reset()
+ timeutils.utcnow.override_time = None
+ self.assertEqual(expires, reference + datetime.timedelta(seconds=120))
+
+ def test_update_fixed_ip_lease_expiration(self):
+ cfg.CONF.set_override('dhcp_lease_duration', 10)
+ plugin = QuantumManager.get_plugin()
+ with self.subnet() as subnet:
+ with self.port(subnet=subnet) as port:
+ update_context = context.Context('', port['port']['tenant_id'])
+ plugin.update_fixed_ip_lease_expiration(
+ update_context,
+ subnet['subnet']['network_id'],
+ port['port']['fixed_ips'][0]['ip_address'],
+ 500)
+
+ q = update_context.session.query(models_v2.IPAllocation)
+ q = q.filter_by(
+ port_id=port['port']['id'],
+ ip_address=port['port']['fixed_ips'][0]['ip_address'])
+
+ ip_allocation = q.one()
+
+ self.assertGreater(
+ ip_allocation.expiration - timeutils.utcnow(),
+ datetime.timedelta(seconds=10))
+
+ cfg.CONF.reset()
+
+ def test_update_fixed_ip_lease_expiration_invalid_address(self):
+ cfg.CONF.set_override('dhcp_lease_duration', 10)
+ plugin = QuantumManager.get_plugin()
+ with self.subnet() as subnet:
+ with self.port(subnet=subnet) as port:
+ update_context = context.Context('', port['port']['tenant_id'])
+ with mock.patch.object(db_base_plugin_v2, 'LOG') as log:
+ plugin.update_fixed_ip_lease_expiration(
+ update_context,
+ subnet['subnet']['network_id'],
+ '255.255.255.0',
+ 120)
+ self.assertTrue(log.mock_calls)
+ cfg.CONF.reset()
+
class TestNetworksV2(QuantumDbPluginV2TestCase):
# NOTE(cerberus): successful network update and delete are
req = self.new_delete_request('subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEquals(res.status_int, 204)
-
- def test_default_allocation_expiration(self):
- reference = datetime.datetime(2012, 8, 13, 23, 11, 0)
- timeutils.utcnow.override_time = reference
-
- cfg.CONF.set_override('dhcp_lease_duration', 120)
- expires = QuantumManager.get_plugin()._default_allocation_expiration()
- timeutils.utcnow
- cfg.CONF.reset()
- timeutils.utcnow.override_time = None
- self.assertEqual(expires, reference + datetime.timedelta(seconds=120))
# License for the specific language governing permissions and limitations
# under the License.
+import socket
import uuid
import mock
from quantum.agent.linux import interface
from quantum.common import exceptions
from quantum.openstack.common import cfg
+from quantum.openstack.common import jsonutils
class FakeModel:
class TestDhcpAgent(unittest.TestCase):
def setUp(self):
cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS)
+ cfg.CONF.register_opts(dhcp_agent.DhcpLeaseRelay.OPTS)
self.driver_cls_p = mock.patch(
'quantum.agent.dhcp_agent.importutils.import_class')
self.driver = mock.Mock(name='driver')
dhcp = dhcp_agent.DhcpAgent(cfg.CONF)
with mock.patch.object(dhcp, 'enable_dhcp_helper') as enable:
- dhcp.run()
- enable.assert_called_once_with('a')
- plug.assert_called_once_with('q-plugin', mock.ANY)
- mock_plugin.assert_has_calls(
- [mock.call.get_active_networks()])
+ with mock.patch.object(dhcp, 'lease_relay') as relay:
+ dhcp.run()
+ enable.assert_called_once_with('a')
+ plug.assert_called_once_with('q-plugin', mock.ANY)
+ mock_plugin.assert_has_calls(
+ [mock.call.get_active_networks()])
+ relay.assert_has_mock_calls([mock.call.run()])
self.notification.assert_has_calls([mock.call.run_dispatch()])
device_id='devid',
host='foo')
+ def test_update_lease_expiration(self):
+ with mock.patch.object(self.proxy, 'cast') as mock_cast:
+ self.proxy.update_lease_expiration('netid', 'ipaddr', 1)
+ mock_cast.assert_called()
+ self.make_msg.assert_called_once_with('update_lease_expiration',
+ network_id='netid',
+ ip_address='ipaddr',
+ lease_remaining=1,
+ host='foo')
+
class TestNetworkCache(unittest.TestCase):
def test_put_network(self):
self.assertEqual(dh.get_device_id(fake_network), expected)
+class TestDhcpLeaseRelay(unittest.TestCase):
+ def setUp(self):
+ cfg.CONF.register_opts(dhcp_agent.DhcpLeaseRelay.OPTS)
+ self.unlink_p = mock.patch('os.unlink')
+ self.unlink = self.unlink_p.start()
+
+ def tearDown(self):
+ self.unlink_p.stop()
+
+ def test_init_relay_socket_path_no_prev_socket(self):
+ with mock.patch('os.path.exists') as exists:
+ exists.return_value = False
+ self.unlink.side_effect = OSError
+
+ relay = dhcp_agent.DhcpLeaseRelay(None)
+
+ self.unlink.assert_called_once_with(
+ cfg.CONF.dhcp_lease_relay_socket)
+ exists.assert_called_once_with(cfg.CONF.dhcp_lease_relay_socket)
+
+ def test_init_relay_socket_path_prev_socket_exists(self):
+ with mock.patch('os.path.exists') as exists:
+ exists.return_value = False
+
+ relay = dhcp_agent.DhcpLeaseRelay(None)
+
+ self.unlink.assert_called_once_with(
+ cfg.CONF.dhcp_lease_relay_socket)
+ self.assertFalse(exists.called)
+
+ def test_init_relay_socket_path_prev_socket_unlink_failure(self):
+ self.unlink.side_effect = OSError
+ with mock.patch('os.path.exists') as exists:
+ exists.return_value = True
+ with self.assertRaises(OSError):
+ relay = dhcp_agent.DhcpLeaseRelay(None)
+
+ self.unlink.assert_called_once_with(
+ cfg.CONF.dhcp_lease_relay_socket)
+ exists.assert_called_once_with(
+ cfg.CONF.dhcp_lease_relay_socket)
+
+ def test_validate_field_valid(self):
+ relay = dhcp_agent.DhcpLeaseRelay(None)
+ retval = relay._validate_field('1b', '\d[a-f]')
+ self.assertEqual(retval, '1b')
+
+ def test_validate_field_invalid(self):
+ relay = dhcp_agent.DhcpLeaseRelay(None)
+ with self.assertRaises(ValueError):
+ retval = relay._validate_field('zz', '\d[a-f]')
+
+ def test_handler_valid_data(self):
+ network_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ ip_address = '192.168.1.9'
+ lease_remaining = 120
+
+ json_rep = jsonutils.dumps(dict(network_id=network_id,
+ lease_remaining=lease_remaining,
+ ip_address=ip_address))
+ handler = mock.Mock()
+ mock_sock = mock.Mock()
+ mock_sock.recv.return_value = json_rep
+
+ relay = dhcp_agent.DhcpLeaseRelay(handler)
+
+ relay._handler(mock_sock, mock.Mock())
+ mock_sock.assert_has_calls([mock.call.recv(1024), mock.call.close()])
+ handler.called_once_with(network_id, ip_address, lease_remaining)
+
+ def test_handler_invalid_data(self):
+ network_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ ip_address = '192.168.x.x'
+ lease_remaining = 120
+
+ json_rep = jsonutils.dumps(
+ dict(network_id=network_id,
+ lease_remaining=lease_remaining,
+ ip_address=ip_address))
+
+ handler = mock.Mock()
+ mock_sock = mock.Mock()
+ mock_sock.recv.return_value = json_rep
+
+ relay = dhcp_agent.DhcpLeaseRelay(handler)
+
+ with mock.patch.object(relay, '_validate_field') as validate:
+ validate.side_effect = ValueError
+
+ with mock.patch.object(dhcp_agent.LOG, 'warn') as log:
+
+ relay._handler(mock_sock, mock.Mock())
+ mock_sock.assert_has_calls(
+ [mock.call.recv(1024), mock.call.close()])
+ self.assertFalse(handler.called)
+ self.assertTrue(log.called)
+
+ def test_handler_other_exception(self):
+ network_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ ip_address = '192.168.x.x'
+ lease_remaining = 120
+
+ json_rep = jsonutils.dumps(
+ dict(network_id=network_id,
+ lease_remaining=lease_remaining,
+ ip_address=ip_address))
+ handler = mock.Mock()
+ mock_sock = mock.Mock()
+ mock_sock.recv.side_effect = Exception
+
+ relay = dhcp_agent.DhcpLeaseRelay(handler)
+
+ with mock.patch.object(dhcp_agent.LOG, 'exception') as log:
+ relay._handler(mock_sock, mock.Mock())
+ mock_sock.assert_has_calls([mock.call.recv(1024)])
+ self.assertFalse(handler.called)
+ self.assertTrue(log.called)
+
+ def test_start(self):
+ with mock.patch.object(dhcp_agent, 'eventlet') as mock_eventlet:
+ handler = mock.Mock()
+ relay = dhcp_agent.DhcpLeaseRelay(handler)
+ relay.start()
+
+ mock_eventlet.assert_has_calls(
+ [mock.call.listen(cfg.CONF.dhcp_lease_relay_socket,
+ family=socket.AF_UNIX),
+ mock.call.spawn(mock_eventlet.serve,
+ mock.call.listen.return_value,
+ relay._handler)])
+
+
class TestDictModel(unittest.TestCase):
def test_basic_dict(self):
d = dict(a=1, b=2)
# under the License.
import os
+import socket
import tempfile
import unittest2 as unittest
from quantum.agent.linux import dhcp
from quantum.agent.common import config
from quantum.openstack.common import cfg
+from quantum.openstack.common import jsonutils
class FakeIPAllocation:
os.path.join(root, 'etc', 'quantum.conf.test')]
self.conf = config.setup_conf()
self.conf.register_opts(dhcp.OPTS)
+ self.conf.register_opt(cfg.StrOpt('dhcp_lease_relay_socket',
+ default='$state_path/dhcp/lease_relay'))
self.conf(args=args)
self.conf.set_override('state_path', '')
self.conf.use_namespaces = True
def mock_get_conf_file_name(kind, ensure_conf_dir=False):
return '/dhcp/cccccccc-cccc-cccc-cccc-cccccccccccc/%s' % kind
+ def fake_argv(index):
+ if index == 0:
+ return '/usr/local/bin/quantum-dhcp-agent'
+ else:
+ raise IndexError
+
expected = [
+ 'QUANTUM_RELAY_SOCKET_PATH=/dhcp/lease_relay',
+ 'QUANTUM_NETWORK_ID=cccccccc-cccc-cccc-cccc-cccccccccccc',
'ip',
'netns',
'exec',
'--pid-file=/dhcp/cccccccc-cccc-cccc-cccc-cccccccccccc/pid',
'--dhcp-hostsfile=/dhcp/cccccccc-cccc-cccc-cccc-cccccccccccc/host',
'--dhcp-optsfile=/dhcp/cccccccc-cccc-cccc-cccc-cccccccccccc/opts',
+ ('--dhcp-script=/usr/local/bin/quantum-dhcp-agent-'
+ 'dnsmasq-lease-update'),
'--leasefile-ro',
'--dhcp-range=set:tag0,192.168.0.0,static,120s',
'--dhcp-range=set:tag1,fdca:3ba5:a17a:4ba3::,static,120s'
mocks['_output_opts_file'].return_value = (
'/dhcp/cccccccc-cccc-cccc-cccc-cccccccccccc/opts'
)
- dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(),
- device_delegate=delegate)
- dm.spawn_process()
- self.assertTrue(mocks['_output_opts_file'].called)
- self.execute.assert_called_once_with(expected, root_helper='sudo')
+
+ with mock.patch.object(dhcp.sys, 'argv') as argv:
+ argv.__getitem__.side_effect = fake_argv
+ dm = dhcp.Dnsmasq(self.conf, FakeDualNetwork(),
+ device_delegate=delegate)
+ dm.spawn_process()
+ self.assertTrue(mocks['_output_opts_file'].called)
+ self.execute.assert_called_once_with(expected,
+ root_helper='sudo')
def test_spawn(self):
self._test_spawn([])
self.safe.assert_has_calls([mock.call(exp_host_name, exp_host_data),
mock.call(exp_opt_name, exp_opt_data)])
self.execute.assert_called_once_with(exp_args, root_helper='sudo')
+
+ def _test_lease_relay_script_helper(self, action, lease_remaining,
+ path_exists=True):
+ relay_path = '/dhcp/relay_socket'
+ network_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ mac_address = 'aa:bb:cc:dd:ee:ff'
+ ip_address = '192.168.1.9'
+
+ json_rep = jsonutils.dumps(dict(network_id=network_id,
+ lease_remaining=lease_remaining,
+ mac_address=mac_address,
+ ip_address=ip_address))
+
+ environ = {
+ 'QUANTUM_NETWORK_ID': network_id,
+ 'QUANTUM_RELAY_SOCKET_PATH': relay_path,
+ 'DNSMASQ_TIME_REMAINING': '120',
+ }
+
+ def fake_environ(name, default=None):
+ return environ.get(name, default)
+
+ with mock.patch('os.environ') as mock_environ:
+ mock_environ.get.side_effect = fake_environ
+
+ with mock.patch.object(dhcp, 'sys') as mock_sys:
+ mock_sys.argv = [
+ 'lease-update',
+ action,
+ mac_address,
+ ip_address,
+ ]
+
+ with mock.patch('socket.socket') as mock_socket:
+ mock_conn = mock.Mock()
+ mock_socket.return_value = mock_conn
+
+ with mock.patch('os.path.exists') as mock_exists:
+ mock_exists.return_value = path_exists
+
+ dhcp.Dnsmasq.lease_update()
+
+ mock_exists.assert_called_once_with(relay_path)
+ if path_exists:
+ mock_socket.assert_called_once_with(
+ socket.AF_UNIX, socket.SOCK_STREAM)
+
+ mock_conn.assert_has_calls(
+ [mock.call.connect(relay_path),
+ mock.call.send(json_rep),
+ mock.call.close()])
+
+ def test_lease_relay_script_add(self):
+ self._test_lease_relay_script_helper('add', 120)
+
+ def test_lease_relay_script_old(self):
+ self._test_lease_relay_script_helper('old', 120)
+
+ def test_lease_relay_script_del(self):
+ self._test_lease_relay_script_helper('del', 0)
+
+ def test_lease_relay_script_add_socket_missing(self):
+ self._test_lease_relay_script_helper('add', 120, False)
'link', 'list'],
root_helper='sudo')
+ def test_execute_env_var_prepend(self):
+ self.parent.namespace = 'ns'
+ with mock.patch('quantum.agent.linux.utils.execute') as execute:
+ env = dict(FOO=1, BAR=2)
+ self.netns_cmd.execute(['ip', 'link', 'list'], env)
+ execute.assert_called_once_with(
+ ['FOO=1', 'BAR=2', 'ip', 'netns', 'exec', 'ns', 'ip', 'link',
+ 'list'],
+ root_helper='sudo')
+
class TestDeviceExists(unittest.TestCase):
def test_device_exists(self):
entry_points={
'console_scripts': [
'quantum-dhcp-agent = quantum.agent.dhcp_agent:main',
+ 'quantum-dhcp-agent-dnsmasq-lease-update ='
+ 'quantum.agent.linux.dhcp:Dnsmasq.lease_update',
'quantum-l3-agent = quantum.agent.l3_nat_agent:main',
'quantum-linuxbridge-agent ='
'quantum.plugins.linuxbridge.agent.linuxbridge_quantum_agent:main',
Code Review / openstack-build / neutron-build.git / commitdiff