else:
self.disable_dhcp_helper(network.id)
- def release_lease_for_removed_ips(self, prev_port, updated_port, network):
- """Releases the dhcp lease for ips removed from a port."""
- if prev_port:
- previous_ips = set(fixed_ip.ip_address
- for fixed_ip in prev_port.fixed_ips)
- current_ips = set(fixed_ip.ip_address
- for fixed_ip in updated_port.fixed_ips)
- # pass in port with removed ips on it
- removed_ips = previous_ips - current_ips
- if removed_ips:
- self.call_driver('release_lease',
- network,
- mac_address=updated_port.mac_address,
- removed_ips=removed_ips)
-
@utils.synchronized('dhcp-agent')
def network_create_end(self, context, payload):
"""Handle the network.create.end notification event."""
updated_port = dhcp.DictModel(payload['port'])
network = self.cache.get_network_by_id(updated_port.network_id)
if network:
- prev_port = self.cache.get_port_by_id(updated_port.id)
self.cache.put_port(updated_port)
self.call_driver('reload_allocations', network)
- self.release_lease_for_removed_ips(prev_port, updated_port,
- network)
# Use the update handler for the port create event.
port_create_end = port_update_end
network = self.cache.get_network_by_id(port.network_id)
self.cache.remove_port(port)
self.call_driver('reload_allocations', network)
- removed_ips = [fixed_ip.ip_address
- for fixed_ip in port.fixed_ips]
- self.call_driver('release_lease',
- network,
- mac_address=port.mac_address,
- removed_ips=removed_ips)
def enable_isolated_metadata_proxy(self, network):
def active(self):
"""Boolean representing the running state of the DHCP server."""
- @abc.abstractmethod
- def release_lease(self, mac_address, removed_ips):
- """Release a DHCP lease."""
-
@abc.abstractmethod
def reload_allocations(self):
"""Force the DHCP server to reload the assignment database."""
cmd = ['%s=%s' % pair for pair in env.items()] + cmd
utils.execute(cmd, self.root_helper)
- def release_lease(self, mac_address, removed_ips):
- pass
-
def _release_lease(self, mac_address, ip):
"""Release a DHCP lease."""
cmd = ['dhcp_release', self.interface_name, ip, mac_address]
self.device_manager.destroy(self.network, self.interface_name)
self._remove_config_files()
- def release_lease(self, mac_address, removed_ips):
- pass
-
def reload_allocations(self):
"""Force the DHCP server to reload the assignment database."""
pass
self.dhcp.port_update_end(None, payload)
self.cache.assert_has_calls(
[mock.call.get_network_by_id(fake_port2.network_id),
- mock.call.get_port_by_id(fake_port2.id),
mock.call.put_port(mock.ANY)])
self.call_driver.assert_called_once_with('reload_allocations',
fake_network)
self.dhcp.port_update_end(None, payload)
self.cache.assert_has_calls(
[mock.call.get_network_by_id(fake_port1.network_id),
- mock.call.get_port_by_id(fake_port1.id),
mock.call.put_port(mock.ANY)])
self.call_driver.assert_has_calls(
- [mock.call.call_driver('reload_allocations', fake_network),
- mock.call.call_driver(
- 'release_lease',
- fake_network,
- mac_address=fake_port1.mac_address,
- removed_ips=set([updated_fake_port1.fixed_ips[0].ip_address]))
- ])
+ [mock.call.call_driver('reload_allocations', fake_network)])
def test_port_delete_end(self):
payload = dict(port_id=fake_port2.id)
self.cache.get_port_by_id.return_value = fake_port2
self.dhcp.port_delete_end(None, payload)
- removed_ips = [fixed_ip.ip_address
- for fixed_ip in fake_port2.fixed_ips]
self.cache.assert_has_calls(
[mock.call.get_port_by_id(fake_port2.id),
mock.call.get_network_by_id(fake_network.id),
mock.call.remove_port(fake_port2)])
self.call_driver.assert_has_calls(
- [mock.call.call_driver('reload_allocations', fake_network),
- mock.call.call_driver('release_lease',
- fake_network,
- mac_address=fake_port2.mac_address,
- removed_ips=removed_ips)])
+ [mock.call.call_driver('reload_allocations', fake_network)])
def test_port_delete_end_unknown_port(self):
payload = dict(port_id='unknown')
def spawn_process(self):
self.called.append('spawn')
- def release_lease(self):
- self.called.append('release_lease')
-
class TestBase(base.BaseTestCase):
def setUp(self):
def reload_allocations(self):
pass
- def release_lease(self):
- pass
-
@property
def active(self):
return True