openstack
/
neutron-dynamic-routing
Code Issues Proposed changes

Merge "Fix TypeError when formatting BGP IP address." into stable/wallaby

This commit is contained in:
Zuul 2021-07-16 09:52:27 +00:00 committed by Gerrit Code Review
parent cf256fa24d c10de7defe
commit 2f9dbc1df5
2 changed files with 33 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
neutron_dynamic_routing/services/bgp/bgp_plugin.py
View File

@ -232,7 +232,7 @@ class BgpPlugin(service_base.ServicePluginBase,
new_router_id = kwargs['router_id']
last_router_id = kwargs.get('last_known_router_id')
floating_ip_address = kwargs['floating_ip_address']
dest = floating_ip_address + '/32'
dest = str(floating_ip_address) + '/32'
bgp_speakers = self._bgp_speakers_for_gw_network_by_family(
ctx,
kwargs['floating_network_id'],

32
neutron_dynamic_routing/tests/unit/services/bgp/test_bgp_plugin.py
View File

@ -16,10 +16,12 @@
from unittest import mock
import netaddr
from neutron.tests import base
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants as n_const
from neutron_dynamic_routing.api.rpc.callbacks import resources as dr_resources
from neutron_dynamic_routing.db import bgp_db
@ -88,3 +90,33 @@ class TestBgpPlugin(base.BaseTestCase):
events.AFTER_CREATE,
self.plugin,
payload=payload)
def test_floatingip_update_callback(self):
kwargs = {'floating_ip_address': netaddr.IPAddress('10.10.10.10'),
'last_known_router_id': 'old-router-id',
'router_id': '', 'floating_network_id': 'a-b-c-d-e'}
test_context = 'test_context'
get_bpg_speakers_name = '_bgp_speakers_for_gw_network_by_family'
with mock.patch.object(self.plugin, get_bpg_speakers_name) as get_bgp:
with mock.patch.object(self.plugin,
'stop_route_advertisements') as stop_ad:
with mock.patch.object(self.plugin, '_bgp_rpc') as bgp_rpc:
bgp_speaker = mock.Mock()
bgp_speaker.id = '11111111-2222-3333-4444-555555555555'
get_bgp.return_value = [bgp_speaker]
self.plugin.floatingip_update_callback(
test_context, events.AFTER_UPDATE, None, **kwargs)
get_bgp.assert_called_once_with(self.fake_admin_ctx,
'a-b-c-d-e',
n_const.IP_VERSION_4)
old_host_route = [{'destination': '10.10.10.10/32',
'next_hop': None}]
stop_ad.assert_called_once_with(self.fake_admin_ctx,
bgp_rpc,
bgp_speaker.id,
old_host_route)