if not subnet.enable_dhcp:
continue
if subnet.dns_nameservers:
- options.append((self._TAG_PREFIX % i,
- 'option',
- 'dns-server',
- ','.join(subnet.dns_nameservers)))
+ options.append(
+ self._format_option(i, 'dns-server',
+ ','.join(subnet.dns_nameservers)))
host_routes = ["%s,%s" % (hr.destination, hr.nexthop)
for hr in subnet.host_routes]
if host_routes:
- options.append((self._TAG_PREFIX % i,
- 'option',
- 'classless-static-route',
- ','.join(host_routes)))
+ options.append(
+ self._format_option(i, 'classless-static-route',
+ ','.join(host_routes)))
- if subnet.ip_version != 6 and subnet.gateway_ip:
- options.append((self._TAG_PREFIX % i,
- 'option',
- 'router',
- subnet.gateway_ip))
+ if subnet.ip_version == 4:
+ if subnet.gateway_ip:
+ options.append(self._format_option(i, 'router',
+ subnet.gateway_ip))
+ else:
+ options.append(self._format_option(i, 'router'))
name = self.get_conf_file_name('opts')
- replace_file(name, '\n'.join(['tag:%s,%s:%s,%s' % o for o in options]))
+ replace_file(name, '\n'.join(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')
+ def _format_option(self, index, option_name, *args):
+ return ','.join(('tag:' + self._TAG_PREFIX % index,
+ 'option:%s' % option_name) + args)
+
@classmethod
def lease_update(cls):
network_id = os.environ.get(cls.QUANTUM_NETWORK_ID_KEY)
dns_nameservers = []
+class FakeV4SubnetNoGateway:
+ id = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee'
+ ip_version = 4
+ cidr = '192.168.1.0/24'
+ gateway_ip = None
+ enable_dhcp = True
+ host_routes = []
+ dns_nameservers = []
+
+
class FakeV4Network:
id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
subnets = [FakeV4Subnet()]
ports = [FakePort1(), FakePort2(), FakePort3()]
+class FakeV4NoGatewayNetwork:
+ id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
+ subnets = [FakeV4SubnetNoGateway()]
+ ports = [FakePort1()]
+
+
class TestDhcpBase(unittest.TestCase):
def test_base_abc_error(self):
self.assertRaises(TypeError, dhcp.DhcpBase, None)
self.safe.assert_called_once_with('/foo/opts', expected)
+ def test_output_opts_file_no_gateway(self):
+ expected = "tag:tag0,option:router"
+
+ with mock.patch.object(dhcp.Dnsmasq, 'get_conf_file_name') as conf_fn:
+ conf_fn.return_value = '/foo/opts'
+ dm = dhcp.Dnsmasq(self.conf, FakeV4NoGatewayNetwork())
+ dm._output_opts_file()
+
+ self.safe.assert_called_once_with('/foo/opts', expected)
+
def test_reload_allocations(self):
exp_host_name = '/dhcp/cccccccc-cccc-cccc-cccc-cccccccccccc/host'
exp_host_data = """