[Filters]
# dhcp-agent
-dnsmasq: EnvFilter, dnsmasq, root, NEUTRON_NETWORK_ID=
+dnsmasq: CommandFilter, dnsmasq, root
# dhcp-agent uses kill as well, that's handled by the generic KillFilter
# it looks like these are the only signals needed, per
# neutron/agent/linux/dhcp.py
import os
import re
import shutil
-import socket
-import sys
import netaddr
-from oslo_serialization import jsonutils
from oslo_utils import importutils
import six
_TAG_PREFIX = 'tag%d'
- NEUTRON_NETWORK_ID_KEY = 'NEUTRON_NETWORK_ID'
- NEUTRON_RELAY_SOCKET_PATH_KEY = 'NEUTRON_RELAY_SOCKET_PATH'
-
@classmethod
def check_version(cls):
pass
cmd_callback=self._build_cmdline_callback,
namespace=self.network.namespace,
service=DNSMASQ_SERVICE_NAME,
- cmd_addl_env={self.NEUTRON_NETWORK_ID_KEY: self.network.id},
reload_cfg=reload_with_HUP,
pid_file=pid_filename)
isolated_subnets = cls.get_isolated_subnets(network)
return any(isolated_subnets[subnet.id] for subnet in network.subnets)
- @classmethod
- def lease_update(cls):
- network_id = os.environ.get(cls.NEUTRON_NETWORK_ID_KEY)
- dhcp_relay_socket = os.environ.get(cls.NEUTRON_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()
-
class DeviceManager(object):
def mock_get_conf_file_name(kind):
return '/dhcp/%s/%s' % (network.id, kind)
- def fake_argv(index):
- if index == 0:
- return '/usr/local/bin/neutron-dhcp-agent'
- else:
- raise IndexError()
-
# if you need to change this path here, think twice,
# that means pid files will move around, breaking upgrades
# or backwards-compatibility
expected_pid_file = '/dhcp/%s/pid' % network.id
- expected_env = {'NEUTRON_NETWORK_ID': network.id}
expected = [
'dnsmasq',
'--no-hosts',
)
mocks['interface_name'].__get__ = mock.Mock(return_value='tap0')
- with mock.patch.object(dhcp.sys, 'argv') as argv:
- argv.__getitem__.side_effect = fake_argv
- dm = self._get_dnsmasq(network, test_pm)
- dm.spawn_process()
- self.assertTrue(mocks['_output_opts_file'].called)
+ dm = self._get_dnsmasq(network, test_pm)
+ dm.spawn_process()
+ self.assertTrue(mocks['_output_opts_file'].called)
- test_pm.enable.assert_called_once_with(
- cmd_addl_env=expected_env,
- uuid=network.id,
- service='dnsmasq',
- namespace='qdhcp-ns',
- cmd_callback=mock.ANY,
- reload_cfg=False,
- pid_file=expected_pid_file)
- call_kwargs = test_pm.method_calls[0][2]
- cmd_callback = call_kwargs['cmd_callback']
+ test_pm.enable.assert_called_once_with(
+ uuid=network.id,
+ service='dnsmasq',
+ namespace='qdhcp-ns',
+ cmd_callback=mock.ANY,
+ reload_cfg=False,
+ pid_file=expected_pid_file)
+ call_kwargs = test_pm.method_calls[0][2]
+ cmd_callback = call_kwargs['cmd_callback']
- result_cmd = cmd_callback(expected_pid_file)
+ result_cmd = cmd_callback(expected_pid_file)
- self.assertEqual(expected, result_cmd)
+ self.assertEqual(expected, result_cmd)
def test_spawn(self):
self._test_spawn(['--conf-file=', '--domain=openstacklocal'])
cmd_callback=mock.ANY,
namespace=mock.ANY,
service=mock.ANY,
- cmd_addl_env=mock.ANY,
reload_cfg=True,
pid_file=mock.ANY)])