openstack/neutron-dynamic-routing
Code Issues Proposed changes

pyupgrade changes for Python3.9+

Browse Source 
As discussed at the Epoxy PTG meeting, run an automated
upgrade tool to make code python 3.9+ compliant.

Result of running:

$ pyupgrade --py39-plus $(git ls-files | grep ".py$")

Fixed PEP8 errors introduced by pyupgrade by running:

$ autopep8 --select=E127,E128,E501 --max-line-length 79 -r \
  --in-place neutron-dynamic-routing

Also did manual updates as necessary to fix other errors
and warnings after above commands.

Inspired by Octavia and Nova [0].

[0] https://review.opendev.org/c/openstack/nova/+/896986

Change-Id: I60f3ab4419c107a69aa07f90c66197b124117f65
This commit is contained in:
elajkat
2024-11-05 16:59:12 +01:00
committed by Lajos Katona
parent c7951ae90c
commit 67fbc2ef60
22 changed files with 267 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.pylintrc
View File

@@ -43,7 +43,6 @@ disable=
global-variable-not-assigned,
logging-not-lazy,
no-init,
non-parent-init-called,
pointless-string-statement,
protected-access,
redefined-builtin,

45
neutron_dynamic_routing/api/rpc/agentnotifiers/bgp_dr_rpc_agent_api.py
View File

@@ -19,7 +19,7 @@ import oslo_messaging
from neutron_dynamic_routing.services.bgp.common import constants as bgp_consts
class BgpDrAgentNotifyApi(object):
class BgpDrAgentNotifyApi:
"""API for plugin to notify BGP DrAgent.
This class implements the client side of an rpc interface. The server side
@@ -38,8 +38,9 @@ class BgpDrAgentNotifyApi(object):
This effectively tells the bgp_dragent to resync.
"""
self._notification_host_cast(context, 'agent_updated',
{'admin_state_up': admin_state_up}, host)
self._notification_host_cast(
context, 'agent_updated',
{'admin_state_up': admin_state_up}, host)
def bgp_routes_advertisement(self, context, bgp_speaker_id,
routes, host):
@@ -48,9 +49,10 @@ class BgpDrAgentNotifyApi(object):
Invoked on FIP association, adding router port to a tenant network,
and new DVR port-host bindings, and subnet creation(?).
"""
self._notification_host_cast(context, 'bgp_routes_advertisement_end',
{'advertise_routes': {'speaker_id': bgp_speaker_id,
'routes': routes}}, host)
self._notification_host_cast(
context, 'bgp_routes_advertisement_end',
{'advertise_routes': {'speaker_id': bgp_speaker_id,
'routes': routes}}, host)
def bgp_routes_withdrawal(self, context, bgp_speaker_id,
routes, host):
@@ -59,9 +61,10 @@ class BgpDrAgentNotifyApi(object):
Invoked on FIP disassociation, removal of a router port on a
network, and removal of DVR port-host binding, and subnet delete(?).
"""
self._notification_host_cast(context, 'bgp_routes_withdrawal_end',
{'withdraw_routes': {'speaker_id': bgp_speaker_id,
'routes': routes}}, host)
self._notification_host_cast(
context, 'bgp_routes_withdrawal_end',
{'withdraw_routes': {'speaker_id': bgp_speaker_id,
'routes': routes}}, host)
def bgp_peer_disassociated(self, context, bgp_speaker_id,
bgp_peer_ip, host):
@@ -69,9 +72,10 @@ class BgpDrAgentNotifyApi(object):
This effectively tells the BgpDrAgent to stop a peering session.
"""
self._notification_host_cast(context, 'bgp_peer_disassociation_end',
{'bgp_peer': {'speaker_id': bgp_speaker_id,
'peer_ip': bgp_peer_ip}}, host)
self._notification_host_cast(
context, 'bgp_peer_disassociation_end',
{'bgp_peer': {'speaker_id': bgp_speaker_id,
'peer_ip': bgp_peer_ip}}, host)
def bgp_peer_associated(self, context, bgp_speaker_id,
bgp_peer_id, host):
@@ -79,9 +83,10 @@ class BgpDrAgentNotifyApi(object):
This effectively tells the bgp_dragent to open a peering session.
"""
self._notification_host_cast(context, 'bgp_peer_association_end',
{'bgp_peer': {'speaker_id': bgp_speaker_id,
'peer_id': bgp_peer_id}}, host)
self._notification_host_cast(
context, 'bgp_peer_association_end',
{'bgp_peer': {'speaker_id': bgp_speaker_id,
'peer_id': bgp_peer_id}}, host)
def bgp_speaker_created(self, context, bgp_speaker_id, host):
"""Tell BgpDrAgent about the creation of a BGP Speaker.
@@ -90,8 +95,9 @@ class BgpDrAgentNotifyApi(object):
we need to inform the BgpDrAgent of a new BGP Speaker in case a
peering session needs to opened immediately.
"""
self._notification_host_cast(context, 'bgp_speaker_create_end',
{'bgp_speaker': {'id': bgp_speaker_id}}, host)
self._notification_host_cast(
context, 'bgp_speaker_create_end',
{'bgp_speaker': {'id': bgp_speaker_id}}, host)
def bgp_speaker_removed(self, context, bgp_speaker_id, host):
"""Tell BgpDrAgent about the removal of a BGP Speaker.
@@ -100,8 +106,9 @@ class BgpDrAgentNotifyApi(object):
place, we need to inform the BgpDrAgent of the removal of a
BGP Speaker in case peering sessions need to be stopped.
"""
self._notification_host_cast(context, 'bgp_speaker_remove_end',
{'bgp_speaker': {'id': bgp_speaker_id}}, host)
self._notification_host_cast(
context, 'bgp_speaker_remove_end',
{'bgp_speaker': {'id': bgp_speaker_id}}, host)
def _notification_host_cast(self, context, method, payload, host):
"""Send payload to BgpDrAgent in the cast mode"""

2
neutron_dynamic_routing/api/rpc/handlers/bgp_speaker_rpc.py
View File

@@ -18,7 +18,7 @@ from neutron_lib.plugins import directory
import oslo_messaging
class BgpSpeakerRpcCallback(object):
class BgpSpeakerRpcCallback:
"""BgpDrAgent RPC callback in plugin implementations.
This class implements the server side of an RPC interface.

72
neutron_dynamic_routing/db/bgp_db.py
View File

@@ -127,7 +127,7 @@ class BgpPeer(model_base.BASEV2,
password = sa.Column(sa.String(255), nullable=True)
class BgpDbMixin(object):
class BgpDbMixin:
def create_bgp_speaker(self, context, bgp_speaker):
uuid = uuidutils.generate_uuid()
@@ -155,10 +155,11 @@ class BgpDbMixin(object):
with db_api.CONTEXT_READER.using(context):
bgp_speaker = self.get_bgp_speaker(context, bgp_speaker_id,
fields=bgp_speaker_attrs)
res = dict((k, bgp_speaker[k]) for k in bgp_speaker_attrs)
res['peers'] = self.get_bgp_peers_by_bgp_speaker(context,
bgp_speaker['id'],
fields=bgp_peer_attrs)
res = {k: bgp_speaker[k] for k in bgp_speaker_attrs}
res['peers'] = self.get_bgp_peers_by_bgp_speaker(
context,
bgp_speaker['id'],
fields=bgp_peer_attrs)
res['advertised_routes'] = self.get_routes_by_bgp_speaker_id(
context,
bgp_speaker_id)
@@ -180,7 +181,7 @@ class BgpDbMixin(object):
res_keys = ['local_as', 'tenant_id', 'name', 'ip_version',
'advertise_floating_ip_host_routes',
'advertise_tenant_networks']
res = dict((k, ri[k]) for k in res_keys)
res = {k: ri[k] for k in res_keys}
res['id'] = uuid
bgp_speaker_db = BgpSpeaker(**res)
context.session.add(bgp_speaker_db)
@@ -235,7 +236,7 @@ class BgpDbMixin(object):
with db_api.CONTEXT_WRITER.using(context):
res_keys = ['tenant_id', 'name', 'remote_as', 'peer_ip',
'auth_type', 'password']
res = dict((k, ri[k]) for k in res_keys)
res = {k: ri[k] for k in res_keys}
res['id'] = uuidutils.generate_uuid()
bgp_peer_db = BgpPeer(**res)
context.session.add(bgp_peer_db)
@@ -422,7 +423,7 @@ class BgpDbMixin(object):
'advertise_tenant_networks'}
peer_bindings = bgp_speaker['peers']
network_bindings = bgp_speaker['networks']
res = dict((k, bgp_speaker[k]) for k in attrs)
res = {k: bgp_speaker[k] for k in attrs}
res['peers'] = [x.bgp_peer_id for x in peer_bindings]
res['networks'] = [x.network_id for x in network_bindings]
return db_utils.resource_fields(res, fields)
@@ -453,7 +454,7 @@ class BgpDbMixin(object):
def _make_bgp_peer_dict(self, bgp_peer, fields=None):
attrs = ['tenant_id', 'id', 'name', 'peer_ip', 'remote_as',
'auth_type', 'password']
res = dict((k, bgp_peer[k]) for k in attrs)
res = {k: bgp_peer[k] for k in attrs}
return db_utils.resource_fields(res, fields)
def _get_address_scope_ids_for_bgp_speaker(self, context, bgp_speaker_id):
@@ -752,7 +753,7 @@ class BgpDbMixin(object):
return self._host_route_list_from_tuples(join_query.all())
def _join_fixed_by_host_binding_to_agent_gateway(self, context,
fixed_subq, gw_subq):
fixed_subq, gw_subq):
join_query = context.session.query(fixed_subq.c.ip_address,
gw_subq.c.ip_address)
and_cond = and_(
@@ -883,7 +884,7 @@ class BgpDbMixin(object):
join_query = context.session.query(left_subq.c.cidr,
right_subq.c.ip_address)
and_cond = and_(left_subq.c.router_id == right_subq.c.router_id,
left_subq.c.ip_version == right_subq.c.ip_version)
left_subq.c.ip_version == right_subq.c.ip_version)
join_query = join_query.join(right_subq, and_cond)
return join_query
@@ -929,19 +930,20 @@ class BgpDbMixin(object):
"""Return the filters for querying tenant networks by BGP speaker"""
router_attrs = aliased(l3_attrs_db.RouterExtraAttributes,
name='router_attrs')
return [models_v2.IPAllocation.port_id == l3_db.RouterPort.port_id,
l3_db.RouterPort.router_id == router_attrs.router_id,
l3_db.Router.id == router_attrs.router_id,
l3_db.Router.admin_state_up == sa.sql.true(),
l3_db.RouterPort.port_type != lib_consts.DEVICE_OWNER_ROUTER_GW,
l3_db.RouterPort.port_type != lib_consts.DEVICE_OWNER_ROUTER_SNAT,
models_v2.IPAllocation.subnet_id == models_v2.Subnet.id,
models_v2.Subnet.network_id != BgpSpeakerNetworkBinding.network_id,
models_v2.Subnet.subnetpool_id == models_v2.SubnetPool.id,
models_v2.SubnetPool.address_scope_id.in_(address_scope_ids),
models_v2.Subnet.ip_version == BgpSpeakerNetworkBinding.ip_version,
BgpSpeakerNetworkBinding.bgp_speaker_id == BgpSpeaker.id,
BgpSpeaker.advertise_tenant_networks == sa.sql.true()]
return [
models_v2.IPAllocation.port_id == l3_db.RouterPort.port_id,
l3_db.RouterPort.router_id == router_attrs.router_id,
l3_db.Router.id == router_attrs.router_id,
l3_db.Router.admin_state_up == sa.sql.true(),
l3_db.RouterPort.port_type != lib_consts.DEVICE_OWNER_ROUTER_GW,
l3_db.RouterPort.port_type != lib_consts.DEVICE_OWNER_ROUTER_SNAT,
models_v2.IPAllocation.subnet_id == models_v2.Subnet.id,
models_v2.Subnet.network_id != BgpSpeakerNetworkBinding.network_id,
models_v2.Subnet.subnetpool_id == models_v2.SubnetPool.id,
models_v2.SubnetPool.address_scope_id.in_(address_scope_ids),
models_v2.Subnet.ip_version == BgpSpeakerNetworkBinding.ip_version,
BgpSpeakerNetworkBinding.bgp_speaker_id == BgpSpeaker.id,
BgpSpeaker.advertise_tenant_networks == sa.sql.true()]
def _nexthop_ip_addresses_by_binding_query(self, context,
network_id, bgp_speaker_id):
@@ -962,7 +964,8 @@ class BgpDbMixin(object):
"""Return the filters for querying nexthops by binding network"""
address_scope = aliased(address_scope_db.AddressScope,
name='address_scope')
return [models_v2.IPAllocation.port_id == l3_db.RouterPort.port_id,
return [
models_v2.IPAllocation.port_id == l3_db.RouterPort.port_id,
models_v2.IPAllocation.subnet_id == models_v2.Subnet.id,
BgpSpeaker.id == bgp_speaker_id,
BgpSpeakerNetworkBinding.bgp_speaker_id == BgpSpeaker.id,
@@ -990,13 +993,14 @@ class BgpDbMixin(object):
router_attrs = aliased(l3_attrs_db.RouterExtraAttributes,
name='router_attrs')
return [l3_db.RouterPort.port_type == DEVICE_OWNER_ROUTER_GW,
l3_db.RouterPort.router_id == router_attrs.router_id,
BgpSpeakerNetworkBinding.network_id == models_v2.Subnet.network_id,
BgpSpeakerNetworkBinding.ip_version == models_v2.Subnet.ip_version,
BgpSpeakerNetworkBinding.bgp_speaker_id == bgp_speaker_id,
models_v2.IPAllocation.port_id == l3_db.RouterPort.port_id,
models_v2.IPAllocation.subnet_id == models_v2.Subnet.id]
return [
l3_db.RouterPort.port_type == DEVICE_OWNER_ROUTER_GW,
l3_db.RouterPort.router_id == router_attrs.router_id,
BgpSpeakerNetworkBinding.network_id == models_v2.Subnet.network_id,
BgpSpeakerNetworkBinding.ip_version == models_v2.Subnet.ip_version,
BgpSpeakerNetworkBinding.bgp_speaker_id == bgp_speaker_id,
models_v2.IPAllocation.port_id == l3_db.RouterPort.port_id,
models_v2.IPAllocation.subnet_id == models_v2.Subnet.id]
def _tenant_prefixes_by_router(self, context, router_id, bgp_speaker_id):
with db_api.CONTEXT_READER.using(context):
@@ -1159,7 +1163,7 @@ class BgpDbMixin(object):
port_subnets = subnet_obj.Subnet.get_objects(ctx, **subnets_filter)
port_subnetpools = subnetpool_obj.SubnetPool.get_objects(
ctx, id=[x.subnetpool_id for x in port_subnets])
port_scopes = set([x.address_scope_id for x in port_subnetpools])
port_scopes = {x.address_scope_id for x in port_subnetpools}
if match_address_scopes and len(port_scopes) == 0:
return []
@@ -1178,7 +1182,7 @@ class BgpDbMixin(object):
# If we don't need to match address scopes, return here
if not match_address_scopes:
return list(set([x.network_id for x in gw_ports]))
return list({x.network_id for x in gw_ports})
# Retrieve address scope info for associated gateway networks
gw_fixed_ips = []

2
neutron_dynamic_routing/extensions/bgp_dragentscheduler.py
View File

@@ -123,7 +123,7 @@ class Bgp_dragentscheduler(api_extensions.APIExtensionDescriptor):
return exts
class BgpDrSchedulerPluginBase(object, metaclass=abc.ABCMeta):
class BgpDrSchedulerPluginBase(metaclass=abc.ABCMeta):
"""REST API to operate BGP dynamic routing agent scheduler.
All the methods must be executed in admin context.

21
neutron_dynamic_routing/services/bgp/agent/bgp_dragent.py
View File

@@ -55,7 +55,7 @@ class BgpDrAgent(manager.Manager):
target = oslo_messaging.Target(version='1.0')
def __init__(self, host, conf=None):
super(BgpDrAgent, self).__init__()
super().__init__()
self.initialize_driver(conf)
self.needs_resync_reasons = collections.defaultdict(list)
self.needs_full_sync_reason = None
@@ -125,8 +125,8 @@ class BgpDrAgent(manager.Manager):
def sync_bgp_speaker(self, bgp_speaker):
# sync BGP Speakers
bgp_peer_ips = set(
[bgp_peer['peer_ip'] for bgp_peer in bgp_speaker['peers']])
bgp_peer_ips = {
bgp_peer['peer_ip'] for bgp_peer in bgp_speaker['peers']}
cached_bgp_peer_ips = set(
self.cache.get_bgp_peer_ips(bgp_speaker['id']))
removed_bgp_peer_ips = cached_bgp_peer_ips - bgp_peer_ips
@@ -336,7 +336,7 @@ class BgpDrAgent(manager.Manager):
except (bgp_ext.BgpSpeakerNotFound, RuntimeError):
LOG.warning(_LW('BGP speaker %s may have been deleted and its '
'resources may have already been disposed.'),
bgp_speaker['id'])
bgp_speaker['id'])
def add_bgp_speaker_on_dragent(self, bgp_speaker):
# Caching BGP speaker details in BGPSpeakerCache. Will be used
@@ -481,9 +481,9 @@ class BgpDrAgent(manager.Manager):
if self.cache.is_route_advertised(bgp_speaker_id, route):
self.cache.remove_adv_route(bgp_speaker_id, route)
LOG.debug('Calling driver for withdrawing prefix: %(cidr)s, '
'next_hop: %(nexthop)s',
{'cidr': route['destination'],
'nexthop': route['next_hop']})
'next_hop: %(nexthop)s',
{'cidr': route['destination'],
'nexthop': route['next_hop']})
try:
self.dr_driver_cls.withdraw_route(bgp_speaker_as,
route['destination'],
@@ -526,7 +526,7 @@ class BgpDrAgent(manager.Manager):
return True
class BgpDrPluginApi(object):
class BgpDrPluginApi:
"""Agent side of BgpDrAgent RPC API.
This class implements the client side of an rpc interface.
@@ -562,7 +562,7 @@ class BgpDrPluginApi(object):
bgp_peer_id=bgp_peer_id)
class BgpSpeakerCache(object):
class BgpSpeakerCache:
"""Agent cache of the current BGP speaker state.
This class is designed to support the advertisement for
@@ -660,8 +660,7 @@ class BgpSpeakerCache(object):
class BgpDrAgentWithStateReport(BgpDrAgent):
def __init__(self, host, conf=None):
super(BgpDrAgentWithStateReport,
self).__init__(host, conf)
super().__init__(host, conf)
self.state_rpc = agent_rpc.PluginReportStateAPI(topics.REPORTS)
self.agent_state = {
'agent_type': bgp_consts.AGENT_TYPE_BGP_ROUTING,

2
neutron_dynamic_routing/services/bgp/agent/driver/base.py
View File

@@ -16,7 +16,7 @@
import abc
class BgpDriverBase(object, metaclass=abc.ABCMeta):
class BgpDriverBase(metaclass=abc.ABCMeta):
"""Base class for BGP Speaking drivers.
Any class which provides BGP functionality should extend this

11
neutron_dynamic_routing/services/bgp/agent/driver/os_ken/driver.py
View File

@@ -86,11 +86,12 @@ class OsKenBgpDriver(base.BgpDriverBase):
# Please note: Since, only the route-advertisement support is
# implemented we are explicitly setting the bgp_server_port
# attribute to 0 which disables listening on port 179.
curr_speaker = bgpspeaker.BGPSpeaker(as_number=speaker_as,
router_id=self.routerid, bgp_server_port=0,
best_path_change_handler=best_path_change_cb,
peer_down_handler=bgp_peer_down_cb,
peer_up_handler=bgp_peer_up_cb)
curr_speaker = bgpspeaker.BGPSpeaker(
as_number=speaker_as,
router_id=self.routerid, bgp_server_port=0,
best_path_change_handler=best_path_change_cb,
peer_down_handler=bgp_peer_down_cb,
peer_up_handler=bgp_peer_up_cb)
LOG.info(_LI('Added BGP Speaker for local_as=%(as)d with '
'router_id= %(rtid)s.'),
{'as': speaker_as, 'rtid': self.routerid})

2
neutron_dynamic_routing/services/bgp/agent/driver/utils.py
View File

@@ -67,7 +67,7 @@ def validate_string(param):
param_type='string')
class BgpMultiSpeakerCache(object):
class BgpMultiSpeakerCache:
"""Class for saving multiple BGP speakers information.
Version history:

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

@@ -49,7 +49,7 @@ class BgpPlugin(service_base.ServicePluginBase,
bgp_4byte_asn.ALIAS]
def __init__(self):
super(BgpPlugin, self).__init__()
super().__init__()
self.bgp_drscheduler = importutils.import_object(
cfg.CONF.bgp_drscheduler_driver)
self.agent_notifiers[bgp_consts.AGENT_TYPE_BGP_ROUTING] = (
@@ -107,7 +107,7 @@ class BgpPlugin(service_base.ServicePluginBase,
def get_bgp_speakers(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
return super(BgpPlugin, self).get_bgp_speakers(
return super().get_bgp_speakers(
context,
filters=filters,
fields=fields,
@@ -117,13 +117,13 @@ class BgpPlugin(service_base.ServicePluginBase,
page_reverse=page_reverse)
def get_bgp_speaker(self, context, bgp_speaker_id, fields=None):
return super(BgpPlugin, self).get_bgp_speaker(context,
bgp_speaker_id,
fields=fields)
return super().get_bgp_speaker(context,
bgp_speaker_id,
fields=fields)
def create_bgp_speaker(self, context, bgp_speaker):
bgp_speaker = super(BgpPlugin, self).create_bgp_speaker(context,
bgp_speaker)
bgp_speaker = super().create_bgp_speaker(context,
bgp_speaker)
registry.publish(dr_resources.BGP_SPEAKER, events.AFTER_CREATE,
self, payload=events.DBEventPayload(
context,
@@ -132,15 +132,15 @@ class BgpPlugin(service_base.ServicePluginBase,
return bgp_speaker
def update_bgp_speaker(self, context, bgp_speaker_id, bgp_speaker):
return super(BgpPlugin, self).update_bgp_speaker(context,
bgp_speaker_id,
bgp_speaker)
return super().update_bgp_speaker(context,
bgp_speaker_id,
bgp_speaker)
def delete_bgp_speaker(self, context, bgp_speaker_id):
hosted_bgp_dragents = self.get_dragents_hosting_bgp_speakers(
context,
[bgp_speaker_id])
super(BgpPlugin, self).delete_bgp_speaker(context, bgp_speaker_id)
super().delete_bgp_speaker(context, bgp_speaker_id)
for agent in hosted_bgp_dragents:
self._bgp_rpc.bgp_speaker_removed(context,
bgp_speaker_id,
@@ -148,32 +148,32 @@ class BgpPlugin(service_base.ServicePluginBase,
def get_bgp_peers(self, context, fields=None, filters=None, sorts=None,
limit=None, marker=None, page_reverse=False):
return super(BgpPlugin, self).get_bgp_peers(
return super().get_bgp_peers(
context, fields=fields,
filters=filters, sorts=sorts,
limit=limit, marker=marker,
page_reverse=page_reverse)
def get_bgp_peer(self, context, bgp_peer_id, fields=None):
return super(BgpPlugin, self).get_bgp_peer(context,
bgp_peer_id,
fields=fields)
return super().get_bgp_peer(context,
bgp_peer_id,
fields=fields)
def create_bgp_peer(self, context, bgp_peer):
return super(BgpPlugin, self).create_bgp_peer(context, bgp_peer)
return super().create_bgp_peer(context, bgp_peer)
def update_bgp_peer(self, context, bgp_peer_id, bgp_peer):
return super(BgpPlugin, self).update_bgp_peer(context,
bgp_peer_id,
bgp_peer)
return super().update_bgp_peer(context,
bgp_peer_id,
bgp_peer)
def delete_bgp_peer(self, context, bgp_peer_id):
super(BgpPlugin, self).delete_bgp_peer(context, bgp_peer_id)
super().delete_bgp_peer(context, bgp_peer_id)
def add_bgp_peer(self, context, bgp_speaker_id, bgp_peer_info):
ret_value = super(BgpPlugin, self).add_bgp_peer(context,
bgp_speaker_id,
bgp_peer_info)
ret_value = super().add_bgp_peer(context,
bgp_speaker_id,
bgp_peer_info)
hosted_bgp_dragents = self.get_dragents_hosting_bgp_speakers(
context,
[bgp_speaker_id])
@@ -187,9 +187,9 @@ class BgpPlugin(service_base.ServicePluginBase,
hosted_bgp_dragents = self.get_dragents_hosting_bgp_speakers(
context, [bgp_speaker_id])
ret_value = super(BgpPlugin, self).remove_bgp_peer(context,
bgp_speaker_id,
bgp_peer_info)
ret_value = super().remove_bgp_peer(context,
bgp_speaker_id,
bgp_peer_info)
for agent in hosted_bgp_dragents:
self._bgp_rpc.bgp_peer_disassociated(context,
@@ -198,37 +198,37 @@ class BgpPlugin(service_base.ServicePluginBase,
agent.host)
def add_bgp_speaker_to_dragent(self, context, agent_id, speaker_id):
super(BgpPlugin, self).add_bgp_speaker_to_dragent(context,
agent_id,
speaker_id)
super().add_bgp_speaker_to_dragent(context,
agent_id,
speaker_id)
def remove_bgp_speaker_from_dragent(self, context, agent_id, speaker_id):
super(BgpPlugin, self).remove_bgp_speaker_from_dragent(context,
agent_id,
speaker_id)
super().remove_bgp_speaker_from_dragent(context,
agent_id,
speaker_id)
def list_bgp_speaker_on_dragent(self, context, agent_id):
return super(BgpPlugin, self).list_bgp_speaker_on_dragent(context,
agent_id)
return super().list_bgp_speaker_on_dragent(context,
agent_id)
def list_dragent_hosting_bgp_speaker(self, context, speaker_id):
return super(BgpPlugin, self).list_dragent_hosting_bgp_speaker(
return super().list_dragent_hosting_bgp_speaker(
context,
speaker_id)
def add_gateway_network(self, context, bgp_speaker_id, network_info):
return super(BgpPlugin, self).add_gateway_network(context,
bgp_speaker_id,
network_info)
return super().add_gateway_network(context,
bgp_speaker_id,
network_info)
def remove_gateway_network(self, context, bgp_speaker_id, network_info):
return super(BgpPlugin, self).remove_gateway_network(context,
bgp_speaker_id,
network_info)
return super().remove_gateway_network(context,
bgp_speaker_id,
network_info)
def get_advertised_routes(self, context, bgp_speaker_id):
return super(BgpPlugin, self).get_advertised_routes(context,
bgp_speaker_id)
return super().get_advertised_routes(context,
bgp_speaker_id)
def floatingip_update_callback(self, resource, event, trigger, payload):
if event not in [events.AFTER_CREATE, events.AFTER_UPDATE]:

8
neutron_dynamic_routing/services/bgp/scheduler/bgp_dragent_scheduler.py
View File

@@ -60,8 +60,8 @@ class BgpDrAgentFilter(base_resource_filter.BaseResourceFilter):
'hosted by BgpDrAgent %(agent_id)s',
{'bgp_speaker_id': bgp_speaker_id,
'agent_id': agent_id})
super(BgpDrAgentFilter, self).bind(context, bound_agents,
bgp_speaker_id, force_scheduling)
super().bind(context, bound_agents,
bgp_speaker_id, force_scheduling)
def filter_agents(self, plugin, context, bgp_speaker):
"""Return the agents that can host the BgpSpeaker."""
@@ -210,7 +210,7 @@ class ChanceScheduler(base_scheduler.BaseChanceScheduler,
BgpDrAgentSchedulerBase):
def __init__(self):
super(ChanceScheduler, self).__init__(self)
super().__init__(self)
self._register_callbacks()
@@ -218,7 +218,7 @@ class WeightScheduler(base_scheduler.BaseWeightScheduler,
BgpDrAgentSchedulerBase):
def __init__(self):
super(WeightScheduler, self).__init__(self)
super().__init__(self)
self._register_callbacks()

2
neutron_dynamic_routing/tests/common/helpers.py
View File

@@ -30,7 +30,7 @@ def _get_bgp_dragent_dict(host):
def register_bgp_dragent(host=helpers.HOST, admin_state_up=True,
alive=True):
alive=True):
agent = helpers._register_agent(
_get_bgp_dragent_dict(host))

4
neutron_dynamic_routing/tests/functional/services/bgp/scheduler/test_bgp_dragent_scheduler.py
View File

@@ -166,14 +166,14 @@ class TestAutoSchedule(testlib_api.SqlTestCase,
msg = 'host_index = %s' % host_index
# create hosts
hosts = ['%s-agent-%s' % (host_index, i)
hosts = ['{}-agent-{}'.format(host_index, i)
for i in range(self.host_count)]
bgp_dragents = self._create_and_set_agents_down(hosts,
self.agent_count,
self.down_agent_count)
# create bgp_speakers
self._bgp_speakers = ['%s-bgp-speaker-%s' % (host_index, i)
self._bgp_speakers = ['{}-bgp-speaker-{}'.format(host_index, i)
for i in range(self.bgp_speaker_count)]
self._save_bgp_speakers(self._bgp_speakers)

2
neutron_dynamic_routing/tests/unit/api/rpc/agentnotifiers/test_bgp_dr_rpc_agent_api.py
View File

@@ -24,7 +24,7 @@ from neutron_dynamic_routing.api.rpc.agentnotifiers import bgp_dr_rpc_agent_api
class TestBgpDrAgentNotifyApi(base.BaseTestCase):
def setUp(self):
super(TestBgpDrAgentNotifyApi, self).setUp()
super().setUp()
self.notifier = (
bgp_dr_rpc_agent_api.BgpDrAgentNotifyApi())

2
neutron_dynamic_routing/tests/unit/api/rpc/handlers/test_bgp_speaker_rpc.py
View File

@@ -25,7 +25,7 @@ from neutron_dynamic_routing.api.rpc.handlers import bgp_speaker_rpc
class TestBgpSpeakerRpcCallback(base.BaseTestCase):
def setUp(self):
super(TestBgpSpeakerRpcCallback, self).setUp()
super().setUp()
self.plugin = mock.Mock()
directory.add_plugin(bgp_ext.ALIAS, self.plugin)
self.callback = bgp_speaker_rpc.BgpSpeakerRpcCallback()

131
neutron_dynamic_routing/tests/unit/db/test_bgp_db.py
View File

@@ -44,7 +44,7 @@ class TestL3Plugin(test_l3.TestL3NatAgentSchedulingServicePlugin,
pass
class BgpEntityCreationMixin(object):
class BgpEntityCreationMixin:
@contextlib.contextmanager
def bgp_speaker(self, ip_version, local_as, name='my-speaker',
@@ -56,7 +56,7 @@ class BgpEntityCreationMixin(object):
'advertise_tenant_networks': advertise_tenant_networks,
'local_as': local_as, 'name': name}
bgp_speaker = self.bgp_plugin.create_bgp_speaker(self.context,
{'bgp_speaker': data})
{'bgp_speaker': data})
bgp_speaker_id = bgp_speaker['id']
if networks:
@@ -181,10 +181,10 @@ class BgpEntityCreationMixin(object):
self.subnetpool([tenant_prefix], **tenant_pool_args) as int_pool:
int_subnetpool_id = int_pool['subnetpool']['id']
with self.subnet(int_net,
cidr=tenant_prefix,
subnetpool_id=int_subnetpool_id,
ip_version=tenant_ip_net.version,
as_admin=True) as int_subnet:
cidr=tenant_prefix,
subnetpool_id=int_subnetpool_id,
ip_version=tenant_ip_net.version,
as_admin=True) as int_subnet:
ext_gw_info = {'network_id': external_network_id}
with self.router(external_gateway_info=ext_gw_info,
distributed=distributed,
@@ -255,8 +255,8 @@ class BgpTests(BgpEntityCreationMixin):
self.assertEqual(v, bgp_speaker[k])
def test_bgp_speaker_list(self):
with self.subnetpool_with_address_scope(4,
prefixes=['8.0.0.0/8']) as sp1, \
with self.subnetpool_with_address_scope(
4, prefixes=['8.0.0.0/8']) as sp1, \
self.subnetpool_with_address_scope(4,
prefixes=['9.0.0.0/8']) as sp2:
with self.bgp_speaker(sp1['ip_version'], 1234,
@@ -270,7 +270,7 @@ class BgpTests(BgpEntityCreationMixin):
local_as_1 = 1234
local_as_2 = 4321
with self.subnetpool_with_address_scope(4,
prefixes=['8.0.0.0/8']) as sp:
prefixes=['8.0.0.0/8']) as sp:
with self.bgp_speaker(sp['ip_version'], local_as_1) as speaker:
self.assertEqual(local_as_1, speaker['local_as'])
new_speaker = self.bgp_plugin.update_bgp_speaker(
@@ -438,8 +438,8 @@ class BgpTests(BgpEntityCreationMixin):
self.gw_network() as network:
network_id = network['network']['id']
self.bgp_plugin.add_gateway_network(self.context,
speaker['id'],
{'network_id': network_id})
speaker['id'],
{'network_id': network_id})
new_speaker = self.bgp_plugin.get_bgp_speaker(self.context,
speaker['id'])
self.assertEqual(1, len(new_speaker['networks']))
@@ -463,8 +463,9 @@ class BgpTests(BgpEntityCreationMixin):
prefixes=['8.0.0.0/8']) as sp:
network1_id = network1['network']['id']
network2_id = network2['network']['id']
with self.bgp_speaker(sp['ip_version'], 1234,
networks=[network1_id, network2_id]) as speaker:
with self.bgp_speaker(
sp['ip_version'], 1234,
networks=[network1_id, network2_id]) as speaker:
self.bgp_plugin.remove_gateway_network(
self.context,
speaker['id'],
@@ -501,11 +502,11 @@ class BgpTests(BgpEntityCreationMixin):
self.gw_network() as network:
network_id = network['network']['id']
self.bgp_plugin.add_gateway_network(self.context,
speaker1['id'],
{'network_id': network_id})
speaker1['id'],
{'network_id': network_id})
self.bgp_plugin.add_gateway_network(self.context,
speaker2['id'],
{'network_id': network_id})
speaker2['id'],
{'network_id': network_id})
speaker1 = self.bgp_plugin.get_bgp_speaker(self.context,
speaker1['id'])
speaker2 = self.bgp_plugin.get_bgp_speaker(self.context,
@@ -570,9 +571,9 @@ class BgpTests(BgpEntityCreationMixin):
self.plugin.create_subnet(self.context, {'subnet': subnet2_data})
self.plugin.create_subnet(self.context, {'subnet': subnet3_data})
self.bgp_plugin.add_gateway_network(self.context, speaker['id'],
{'network_id': network1_id})
{'network_id': network1_id})
self.bgp_plugin.add_gateway_network(self.context, speaker['id'],
{'network_id': network2_id})
{'network_id': network2_id})
scopes = self.bgp_plugin._get_address_scope_ids_for_bgp_speaker(
self.context,
speaker['id'])
@@ -640,12 +641,12 @@ class BgpTests(BgpEntityCreationMixin):
self.assertEqual(next_hop, routes[0]['next_hop'])
def _advertised_routes_by_bgp_speaker(self,
bgp_speaker_ip_version,
local_as,
tenant_cidr,
gateway_cidr,
fip_routes=True,
router_distributed=False):
bgp_speaker_ip_version,
local_as,
tenant_cidr,
gateway_cidr,
fip_routes=True,
router_distributed=False):
tenant_id = _uuid()
scope_data = {'tenant_id': tenant_id,
'ip_version': bgp_speaker_ip_version,
@@ -673,8 +674,8 @@ class BgpTests(BgpEntityCreationMixin):
def test__tenant_prefixes_by_router_no_gateway_port(self):
with self.network() as net1, self.network() as net2, \
self.subnetpool_with_address_scope(6, tenant_id=_uuid(),
prefixes=['2001:db8::/63']) as pool:
self.subnetpool_with_address_scope(
6, tenant_id=_uuid(), prefixes=['2001:db8::/63']) as pool:
subnetpool_id = pool['id']
with self.subnet(network=net1,
cidr=None,
@@ -745,10 +746,10 @@ class BgpTests(BgpEntityCreationMixin):
def test_all_routes_by_bgp_speaker_different_tenant_address_scope(self):
binding_cidr = '2001:db8::/64'
tenant_cidr = '2002:ab8::/64'
with self.subnetpool_with_address_scope(6, tenant_id=_uuid(),
prefixes=[binding_cidr]) as ext_pool, \
self.subnetpool_with_address_scope(6, tenant_id=_uuid(),
prefixes=[tenant_cidr]) as int_pool, \
with self.subnetpool_with_address_scope(
6, tenant_id=_uuid(), prefixes=[binding_cidr]) as ext_pool, \
self.subnetpool_with_address_scope(
6, tenant_id=_uuid(), prefixes=[tenant_cidr]) as int_pool, \
self.gw_network(external=True) as ext_net, \
self.network() as int_net:
gw_net_id = ext_net['network']['id']
@@ -961,36 +962,36 @@ class BgpTests(BgpEntityCreationMixin):
# fip setup
tenant_net_id1 = int_net1['network']['id']
fixed_port_data1 = {'port':
{'name': 'test',
'network_id': tenant_net_id1,
'tenant_id': tenant_id,
'admin_state_up': True,
'device_id': _uuid(),
'device_owner': 'compute:nova',
'mac_address': n_const.ATTR_NOT_SPECIFIED,
'fixed_ips': n_const.ATTR_NOT_SPECIFIED}}
{'name': 'test',
'network_id': tenant_net_id1,
'tenant_id': tenant_id,
'admin_state_up': True,
'device_id': _uuid(),
'device_owner': 'compute:nova',
'mac_address': n_const.ATTR_NOT_SPECIFIED,
'fixed_ips': n_const.ATTR_NOT_SPECIFIED}}
fixed_port1 = self.plugin.create_port(self.context,
fixed_port_data1)
fixed_port_data1)
fip_data1 = {'floatingip': {'floating_network_id': gw_net_id,
'tenant_id': tenant_id,
'port_id': fixed_port1['id']}}
'tenant_id': tenant_id,
'port_id': fixed_port1['id']}}
fip1 = self.l3plugin.create_floatingip(self.context, fip_data1)
tenant_net_id2 = int_net2['network']['id']
fixed_port_data2 = {'port':
{'name': 'test',
'network_id': tenant_net_id2,
'tenant_id': tenant_id,
'admin_state_up': True,
'device_id': _uuid(),
'device_owner': 'compute:nova',
'mac_address': n_const.ATTR_NOT_SPECIFIED,
'fixed_ips': n_const.ATTR_NOT_SPECIFIED}}
{'name': 'test',
'network_id': tenant_net_id2,
'tenant_id': tenant_id,
'admin_state_up': True,
'device_id': _uuid(),
'device_owner': 'compute:nova',
'mac_address': n_const.ATTR_NOT_SPECIFIED,
'fixed_ips': n_const.ATTR_NOT_SPECIFIED}}
fixed_port2 = self.plugin.create_port(self.context,
fixed_port_data2)
fixed_port_data2)
fip_data2 = {'floatingip': {'floating_network_id': gw_net_id,
'tenant_id': tenant_id,
'port_id': fixed_port2['id']}}
'tenant_id': tenant_id,
'port_id': fixed_port2['id']}}
self.l3plugin.create_floatingip(self.context, fip_data2)
# disable router2
@@ -1514,13 +1515,13 @@ class BgpTests(BgpEntityCreationMixin):
'fixed_ips': n_const.ATTR_NOT_SPECIFIED,
portbindings.HOST_ID: 'test-host'}}
fixed_port = self.plugin.create_port(self.context,
fixed_port_data)
fixed_port_data)
fixed_ip_prefix = fixed_port['fixed_ips'][0]['ip_address'] + '/32'
self.plugin.create_or_update_agent(self.context,
{'agent_type': 'L3 agent',
'host': 'test-host',
'binary': 'neutron-l3-agent',
'topic': 'test'})
{'agent_type': 'L3 agent',
'host': 'test-host',
'binary': 'neutron-l3-agent',
'topic': 'test'})
fip_gw = self.l3plugin.create_fip_agent_gw_port_if_not_exists(
self.context,
gw_net_id,
@@ -1650,12 +1651,12 @@ class BgpTests(BgpEntityCreationMixin):
'fixed_ips': n_const.ATTR_NOT_SPECIFIED,
portbindings.HOST_ID: 'test-host'}}
fixed_port = self.plugin.create_port(self.context,
fixed_port_data)
fixed_port_data)
self.plugin.create_or_update_agent(self.context,
{'agent_type': 'L3 agent',
'host': 'test-host',
'binary': 'neutron-l3-agent',
'topic': 'test'})
{'agent_type': 'L3 agent',
'host': 'test-host',
'binary': 'neutron-l3-agent',
'topic': 'test'})
ext_nets = self.bgp_plugin.get_external_networks_for_port(
self.context,
fixed_port,
@@ -1678,10 +1679,10 @@ class Ml2BgpTests(test_plugin.Ml2PluginV2TestCase,
def setup_parent(self):
self.l3_plugin = ('neutron_dynamic_routing.tests.unit.db.test_bgp_db.'
'TestL3Plugin')
super(Ml2BgpTests, self).setup_parent()
super().setup_parent()
def setUp(self):
super(Ml2BgpTests, self).setUp()
super().setUp()
self.l3plugin = directory.get_plugin(plugin_constants.L3)
self.bgp_plugin = bgp_plugin.BgpPlugin()
self.plugin = directory.get_plugin()

4
neutron_dynamic_routing/tests/unit/db/test_bgp_dragentscheduler_db.py
View File

@@ -31,7 +31,7 @@ from neutron_dynamic_routing.tests.unit.db import test_bgp_db
from webob import exc
class BgpDrSchedulerTestExtensionManager(object):
class BgpDrSchedulerTestExtensionManager:
def get_resources(self):
return (agent.Agent.get_resources() +
@@ -195,7 +195,7 @@ class BgpDrPluginSchedulerTests(test_db_base_plugin.NeutronDbPluginV2TestCase,
'bgp_plugin.BgpPlugin'}
ext_mgr = ext_mgr or BgpDrSchedulerTestExtensionManager()
super(BgpDrPluginSchedulerTests, self).setUp(
super().setUp(
plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
self.bgp_plugin = directory.get_plugin(bgp.ALIAS)
self.context = context.get_admin_context()

36
neutron_dynamic_routing/tests/unit/services/bgp/agent/test_bgp_dragent.py
View File

@@ -32,7 +32,7 @@ from neutron_dynamic_routing.services.bgp.agent import config as bgp_config
HOSTNAME = 'hostname'
rpc_api = bgp_dragent.BgpDrPluginApi
BGP_PLUGIN = '%s.%s' % (rpc_api.__module__, rpc_api.__name__)
BGP_PLUGIN = '{}.{}'.format(rpc_api.__module__, rpc_api.__name__)
FAKE_BGPSPEAKER_UUID = uuidutils.generate_uuid()
FAKE_BGPPEER_UUID = uuidutils.generate_uuid()
@@ -63,7 +63,7 @@ FAKE_ROUTES = {'routes': {'id': FAKE_BGPSPEAKER_UUID,
class TestBgpDrAgent(base.BaseTestCase):
def setUp(self):
super(TestBgpDrAgent, self).setUp()
super().setUp()
cfg.CONF.register_opts(bgp_config.BGP_DRIVER_OPTS, 'BGP')
cfg.CONF.register_opts(bgp_config.BGP_PROTO_CONFIG_OPTS, 'BGP')
cfg.CONF.register_opts(config.AGENT_STATE_OPTS, 'AGENT')
@@ -131,11 +131,11 @@ class TestBgpDrAgent(base.BaseTestCase):
synced_bgp_speakers=None):
bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
attrs_to_mock = dict(
[(a, mock.Mock())
for a in ['plugin_rpc', 'sync_bgp_speaker',
'safe_configure_dragent_for_bgp_speaker',
'remove_bgp_speaker_from_dragent']])
attrs_to_mock = {
a: mock.Mock()
for a in ['plugin_rpc', 'sync_bgp_speaker',
'safe_configure_dragent_for_bgp_speaker',
'remove_bgp_speaker_from_dragent']}
with mock.patch.multiple(bgp_dr, **attrs_to_mock):
if not cached_info:
@@ -290,12 +290,12 @@ class TestBgpDrAgent(base.BaseTestCase):
bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
attrs_to_mock = dict(
[(a, mock.Mock())
for a in ['remove_bgp_peer_from_bgp_speaker',
'add_bgp_peers_to_bgp_speaker',
'advertise_routes_via_bgp_speaker',
'withdraw_route_via_bgp_speaker']])
attrs_to_mock = {
a: mock.Mock()
for a in ['remove_bgp_peer_from_bgp_speaker',
'add_bgp_peers_to_bgp_speaker',
'advertise_routes_via_bgp_speaker',
'withdraw_route_via_bgp_speaker']}
with mock.patch.multiple(bgp_dr, **attrs_to_mock):
bgp_dr.cache.cache = cached_info
@@ -412,11 +412,11 @@ class TestBgpDrAgent(base.BaseTestCase):
with mock.patch.object(bgp_dragent.LOG, 'error') as log:
bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
with mock.patch.object(bgp_dr,
'schedule_full_resync') as schedule_full_resync:
'schedule_full_resync') as mock_resync:
bgp_dr.sync_state(mock.ANY)
self.assertTrue(log.called)
self.assertTrue(schedule_full_resync.called)
self.assertTrue(mock_resync.called)
def test_periodic_resync(self):
bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
@@ -511,7 +511,7 @@ class TestBgpDrAgentEventHandler(base.BaseTestCase):
'agent.bgp_dragent.BgpSpeakerCache'
def setUp(self):
super(TestBgpDrAgentEventHandler, self).setUp()
super().setUp()
cfg.CONF.register_opts(bgp_config.BGP_DRIVER_OPTS, 'BGP')
cfg.CONF.register_opts(bgp_config.BGP_PROTO_CONFIG_OPTS, 'BGP')
@@ -573,7 +573,7 @@ class TestBgpDrAgentEventHandler(base.BaseTestCase):
def test_add_bgp_speaker_helper(self):
self.plugin.get_bgp_speaker_info.return_value = FAKE_BGP_SPEAKER
add_bs_p = mock.patch.object(self.bgp_dr,
'add_bgp_speaker_on_dragent')
'add_bgp_speaker_on_dragent')
add_bs = add_bs_p.start()
self.bgp_dr.add_bgp_speaker_helper(FAKE_BGP_SPEAKER['id'])
self.plugin.assert_has_calls([
@@ -635,7 +635,7 @@ class TestBgpDrAgentEventHandler(base.BaseTestCase):
class TestBGPSpeakerCache(base.BaseTestCase):
def setUp(self):
super(TestBGPSpeakerCache, self).setUp()
super().setUp()
self.expected_cache = {FAKE_BGP_SPEAKER['id']:
{'bgp_speaker': FAKE_BGP_SPEAKER,
'peers': {},

74
neutron_dynamic_routing/tests/unit/services/bgp/driver/os_ken/test_driver.py
View File

@@ -45,7 +45,7 @@ FAKE_NEXTHOP = '5.5.5.5'
class TestOsKenBgpDriver(base.BaseTestCase):
def setUp(self):
super(TestOsKenBgpDriver, self).setUp()
super().setUp()
cfg.CONF.register_opts(bgp_config.BGP_PROTO_CONFIG_OPTS, 'BGP')
cfg.CONF.set_override('bgp_router_id', FAKE_ROUTER_ID, 'BGP')
self.os_ken_bgp_driver = os_ken_driver.OsKenBgpDriver(cfg.CONF.BGP)
@@ -54,8 +54,8 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_add_new_bgp_speaker(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.mock_os_ken_speaker.assert_called_once_with(
as_number=FAKE_LOCAL_AS1, router_id=FAKE_ROUTER_ID,
bgp_server_port=0,
@@ -65,21 +65,21 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_remove_bgp_speaker(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
self.os_ken_bgp_driver.delete_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(0,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
0, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(1, speaker.shutdown.call_count)
def test_add_bgp_peer_without_password(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.add_bgp_peer(FAKE_LOCAL_AS1,
FAKE_PEER_IP,
FAKE_PEER_AS)
FAKE_PEER_IP,
FAKE_PEER_AS)
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.neighbor_add.assert_called_once_with(
address=FAKE_PEER_IP,
@@ -91,13 +91,13 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_add_bgp_peer_with_password(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.add_bgp_peer(FAKE_LOCAL_AS1,
FAKE_PEER_IP,
FAKE_PEER_AS,
FAKE_AUTH_TYPE,
FAKE_PEER_PASSWORD)
FAKE_PEER_IP,
FAKE_PEER_AS,
FAKE_AUTH_TYPE,
FAKE_PEER_PASSWORD)
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.neighbor_add.assert_called_once_with(
address=FAKE_PEER_IP,
@@ -109,8 +109,8 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_add_bgp_peer_with_unicode_password(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
NEW_FAKE_PEER_PASSWORD = str(FAKE_PEER_PASSWORD)
self.os_ken_bgp_driver.add_bgp_peer(
FAKE_LOCAL_AS1,
@@ -129,11 +129,11 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_add_bgp_peer_with_ipv6(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.add_bgp_peer(FAKE_LOCAL_AS1,
FAKE_PEER_IPV6,
FAKE_PEER_AS)
FAKE_PEER_IPV6,
FAKE_PEER_AS)
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.neighbor_add.assert_called_once_with(
address=FAKE_PEER_IPV6,
@@ -145,27 +145,27 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_remove_bgp_peer(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.delete_bgp_peer(FAKE_LOCAL_AS1, FAKE_PEER_IP)
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.neighbor_del.assert_called_once_with(address=FAKE_PEER_IP)
def test_advertise_route(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.advertise_route(FAKE_LOCAL_AS1,
FAKE_ROUTE,
FAKE_NEXTHOP)
FAKE_ROUTE,
FAKE_NEXTHOP)
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.prefix_add.assert_called_once_with(prefix=FAKE_ROUTE,
next_hop=FAKE_NEXTHOP)
def test_withdraw_route(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.withdraw_route(FAKE_LOCAL_AS1, FAKE_ROUTE)
speaker = self.os_ken_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.prefix_del.assert_called_once_with(prefix=FAKE_ROUTE)
@@ -289,16 +289,16 @@ class TestOsKenBgpDriver(base.BaseTestCase):
def test_add_multiple_bgp_speakers(self):
self.os_ken_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertRaises(bgp_driver_exc.BgpSpeakerMaxScheduled,
self.os_ken_bgp_driver.add_bgp_speaker,
FAKE_LOCAL_AS2)
self.assertRaises(bgp_driver_exc.BgpSpeakerNotAdded,
self.os_ken_bgp_driver.delete_bgp_speaker,
FAKE_LOCAL_AS2)
self.assertEqual(1,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
1, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.os_ken_bgp_driver.delete_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(0,
self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())
self.assertEqual(
0, self.os_ken_bgp_driver.cache.get_hosted_bgp_speakers_count())

4
neutron_dynamic_routing/tests/unit/services/bgp/driver/test_utils.py
View File

@@ -36,7 +36,7 @@ EXC_INV_AUTHTYPE = "Authentication type not supported. Requested " + \
class TestValidateMethod(base.BaseTestCase):
def setUp(self):
super(TestValidateMethod, self).setUp()
super().setUp()
def test_validate_as_num_with_valid_as_num(self):
self.assertIsNone(bgp_driver_utils.validate_as_num('local_as',
@@ -151,7 +151,7 @@ class TestValidateMethod(base.BaseTestCase):
class TestBgpMultiSpeakerCache(base.BaseTestCase):
def setUp(self):
super(TestBgpMultiSpeakerCache, self).setUp()
super().setUp()
self.expected_cache = {FAKE_LOCAL_AS: FAKE_OS_KEN_SPEAKER}
self.bs_cache = bgp_driver_utils.BgpMultiSpeakerCache()

8
neutron_dynamic_routing/tests/unit/services/bgp/scheduler/test_bgp_dragent_scheduler.py
View File

@@ -36,7 +36,7 @@ load_tests = testscenarios.load_tests_apply_scenarios
class TestBgpDrAgentSchedulerBaseTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(TestBgpDrAgentSchedulerBaseTestCase, self).setUp()
super().setUp()
self.ctx = context.get_admin_context()
self.bgp_speaker = {'id': 'foo_bgp_speaker_id'}
self.bgp_speaker_id = 'foo_bgp_speaker_id'
@@ -82,7 +82,7 @@ class TestBgpDrAgentSchedulerBaseTestCase(testlib_api.SqlTestCase):
class TestSchedulerCallback(TestBgpDrAgentSchedulerBaseTestCase):
def setUp(self):
super(TestSchedulerCallback, self).setUp()
super().setUp()
bgp_notify_p = mock.patch('neutron_dynamic_routing.api.rpc.'
'agentnotifiers.bgp_dr_rpc_agent_api.'
'BgpDrAgentNotifyApi')
@@ -155,7 +155,7 @@ class TestBgpAgentFilter(TestBgpDrAgentSchedulerBaseTestCase,
bgp_dras_db.BgpDrAgentSchedulerDbMixin):
def setUp(self):
super(TestBgpAgentFilter, self).setUp()
super().setUp()
self.bgp_drscheduler = importutils.import_object(
'neutron_dynamic_routing.services.bgp.scheduler.'
'bgp_dragent_scheduler.ChanceScheduler'
@@ -295,7 +295,7 @@ class TestRescheduleBgpSpeaker(TestBgpDrAgentSchedulerBaseTestCase,
bgp_db.BgpDbMixin):
def setUp(self):
super(TestRescheduleBgpSpeaker, self).setUp()
super().setUp()
bgp_notify_p = mock.patch('neutron_dynamic_routing.api.rpc.'
'agentnotifiers.bgp_dr_rpc_agent_api.'
'BgpDrAgentNotifyApi')

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

@@ -31,7 +31,7 @@ from neutron_dynamic_routing.services.bgp import bgp_plugin
class TestBgpPlugin(base.BaseTestCase):
def setUp(self):
super(TestBgpPlugin, self).setUp()
super().setUp()
bgp_notify_p = mock.patch('neutron_dynamic_routing.api.rpc.'
'agentnotifiers.bgp_dr_rpc_agent_api.'
'BgpDrAgentNotifyApi')
@@ -128,7 +128,7 @@ class TestBgpPlugin(base.BaseTestCase):
n_const.IP_VERSION_4)
old_host_route = [{'destination': '10.10.10.10/32',
'next_hop': None}]
'next_hop': None}]
stop_ad.assert_called_once_with(self.fake_admin_ctx,
bgp_rpc,
bgp_speaker.id,