x/group-based-policy
Code Issues Proposed changes

Merge "update_floatingip_status_while_deleting_the_vm"

Browse Source
This commit is contained in:
Zuul 2022-01-31 19:51:29 +00:00 committed by Gerrit Code Review
commit 9febdc6f9c
2 changed files with 27 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
gbpservice/neutron/plugins/ml2plus/drivers/apic_aim/mechanism_driver.py
View File

@ -3196,6 +3196,12 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
self._send_postcommit_notifications(context._plugin_context)
def _update_floatingip_status(self, session, port_id):
query = (sa.update(l3_db.FloatingIP).
where(l3_db.FloatingIP.fixed_port_id == port_id).
values(status=n_constants.FLOATINGIP_STATUS_DOWN))
session.execute(query)
def delete_port_precommit(self, context):
port = context.current
if self._is_port_bound(port):
@ -3211,6 +3217,10 @@ class ApicMechanismDriver(api_plus.MechanismDriver,
self._really_update_sg_rule_with_remote_group_set(
context, port, port['security_groups'], is_delete=True)
# Set status of floating ip DOWN.
self._update_floatingip_status(
context._plugin_context.session, port['id'])
# Handle router gateway port deletion.
if self._is_port_router_gateway(port):
router = self.l3_plugin.get_router(

17
gbpservice/neutron/tests/unit/plugins/ml2plus/test_apic_aim.py
View File

@ -8513,6 +8513,23 @@ class TestExternalConnectivityBase(object):
self._do_test_multiple_router(use_addr_scope=True,
shared_l3out=True)
def test_delete_port_assoc_with_floatingip(self):
net1 = self._make_network(self.fmt, 'pvt-net1', True)['network']
sub1 = self._make_subnet(
self.fmt, {'network': net1}, '10.10.1.1', '10.10.1.0/24')
with self.port(subnet=sub1) as port:
port = self._bind_port_to_host(port['port']['id'], 'host1')
port['port']['dns_name'] = ''
p = port['port']
with self.floatingip_with_assoc(port_id=p['id']) as fip:
fip = fip['floatingip']
self.assertEqual('ACTIVE', fip['status'])
fip = self._show('floatingips', fip['id'])['floatingip']
self.assertEqual('ACTIVE', fip['status'])
self._delete('ports', p['id'])
fip = self._show('floatingips', fip['id'])['floatingip']
self.assertEqual('DOWN', fip['status'])
def test_floatingip(self):
net1 = self._make_network(self.fmt, 'pvt-net1', True)['network']
sub1 = self._make_subnet(