openstack/python-neutronclient
Code Issues Proposed changes

Merge "BGPVPN: make resource_association method calls logic simpler"

Browse Source
This commit is contained in:
Zuul 2023-06-22 16:03:42 +00:00 committed by Gerrit Code Review
commit d497615240
3 changed files with 69 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
neutronclient/osc/v2/networking_bgpvpn/resource_association.py
View File

@ -24,6 +24,7 @@ from osc_lib.utils import columns as column_util
from neutronclient._i18n import _
from neutronclient.osc import utils as nc_osc_utils
from neutronclient.osc.v2.networking_bgpvpn import constants
LOG = logging.getLogger(__name__)
@ -56,8 +57,6 @@ class CreateBgpvpnResAssoc(command.ShowOne):
def take_action(self, parsed_args):
client = self.app.client_manager.network
create_method = getattr(
client, 'create_bgpvpn_%s_association' % self._assoc_res_name)
bgpvpn = client.find_bgpvpn(parsed_args.bgpvpn)
find_res_method = getattr(
client, 'find_%s' % self._assoc_res_name)
@ -76,7 +75,14 @@ class CreateBgpvpnResAssoc(command.ShowOne):
body.update(
arg2body(bgpvpn['id'], parsed_args))
obj = create_method(bgpvpn['id'], **body)
if self._assoc_res_name == constants.NETWORK_ASSOC:
obj = client.create_bgpvpn_network_association(
bgpvpn['id'], **body)
elif self._assoc_res_name == constants.PORT_ASSOCS:
obj = client.create_bgpvpn_port_association(bgpvpn['id'], **body)
else:
obj = client.create_bgpvpn_router_association(
bgpvpn['id'], **body)
transform = getattr(self, '_transform_resource', None)
if callable(transform):
transform(obj)
@ -113,14 +119,19 @@ class SetBgpvpnResAssoc(command.Command):
def take_action(self, parsed_args):
client = self.app.client_manager.network
update_method = getattr(
client, 'update_bgpvpn_%s_association' % self._assoc_res_name)
bgpvpn = client.find_bgpvpn(parsed_args.bgpvpn)
arg2body = getattr(self, '_args2body', None)
if callable(arg2body):
body = arg2body(bgpvpn['id'], parsed_args)
update_method(bgpvpn['id'], parsed_args.resource_association_id,
**body)
if self._assoc_res_name == constants.NETWORK_ASSOC:
client.update_bgpvpn_network_association(
bgpvpn['id'], parsed_args.resource_association_id, **body)
elif self._assoc_res_name == constants.PORT_ASSOCS:
client.update_bgpvpn_port_association(
bgpvpn['id'], parsed_args.resource_association_id, **body)
else:
client.update_bgpvpn_router_association(
bgpvpn['id'], parsed_args.resource_association_id, **body)
class UnsetBgpvpnResAssoc(SetBgpvpnResAssoc):
@ -150,13 +161,16 @@ class DeleteBgpvpnResAssoc(command.Command):
def take_action(self, parsed_args):
client = self.app.client_manager.network
delete_method = getattr(
client, 'delete_bgpvpn_%s_association' % self._assoc_res_name)
bgpvpn = client.find_bgpvpn(parsed_args.bgpvpn)
fails = 0
for id in parsed_args.resource_association_ids:
try:
delete_method(bgpvpn['id'], id)
if self._assoc_res_name == constants.NETWORK_ASSOC:
client.delete_bgpvpn_network_association(bgpvpn['id'], id)
elif self._assoc_res_name == constants.PORT_ASSOCS:
client.delete_bgpvpn_port_association(bgpvpn['id'], id)
else:
client.delete_bgpvpn_router_association(bgpvpn['id'], id)
LOG.warning(
"%(assoc_res_name)s association %(id)s deleted",
{'assoc_res_name': self._assoc_res_name.capitalize(),
@ -203,14 +217,19 @@ class ListBgpvpnResAssoc(command.Lister):
def take_action(self, parsed_args):
client = self.app.client_manager.network
list_method = getattr(client,
'bgpvpn_%s_associations' % self._assoc_res_name)
bgpvpn = client.find_bgpvpn(parsed_args.bgpvpn)
params = {}
if parsed_args.property:
params.update(parsed_args.property)
objs = list_method(bgpvpn['id'],
retrieve_all=True, **params)
if self._assoc_res_name == constants.NETWORK_ASSOC:
objs = client.bgpvpn_network_associations(
bgpvpn['id'], retrieve_all=True, **params)
elif self._assoc_res_name == constants.PORT_ASSOCS:
objs = client.bgpvpn_port_associations(
bgpvpn['id'], retrieve_all=True, **params)
else:
objs = client.bgpvpn_router_associations(
bgpvpn['id'], retrieve_all=True, **params)
transform = getattr(self, '_transform_resource', None)
transformed_objs = []
if callable(transform):
@ -245,11 +264,16 @@ class ShowBgpvpnResAssoc(command.ShowOne):
def take_action(self, parsed_args):
client = self.app.client_manager.network
show_method = getattr(
client, 'get_bgpvpn_%s_association' % self._assoc_res_name)
bgpvpn = client.find_bgpvpn(parsed_args.bgpvpn)
obj = show_method(bgpvpn['id'],
parsed_args.resource_association_id)
if self._assoc_res_name == constants.NETWORK_ASSOC:
obj = client.get_bgpvpn_network_association(
bgpvpn['id'], parsed_args.resource_association_id)
elif self._assoc_res_name == constants.PORT_ASSOCS:
obj = client.get_bgpvpn_port_association(
bgpvpn['id'], parsed_args.resource_association_id)
else:
obj = client.get_bgpvpn_router_association(
bgpvpn['id'], parsed_args.resource_association_id)
transform = getattr(self, '_transform_resource', None)
if callable(transform):
transform(obj)

32
neutronclient/tests/unit/osc/v2/networking_bgpvpn/test_resource_association.py
View File

@ -65,7 +65,7 @@ class TestCreateResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res = fakes.create_one_resource()
fake_res_assoc = fakes.create_one_resource_association(
fake_res)
self.networkclient.create_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.create_bgpvpn_router_association = mock.Mock(
return_value=fake_res_assoc)
self.networkclient.find_bgpvpn_fake_resource_association = mock.Mock(
side_effect=lambda name_or_id: {'id': name_or_id})
@ -89,7 +89,7 @@ class TestCreateResAssoc(fakes.TestNeutronClientBgpvpn):
'tenant_id': 'fake_project_id'
}
self.networkclient.create_bgpvpn_fake_resource_association.\
self.networkclient.create_bgpvpn_router_association.\
assert_called_once_with(
fake_bgpvpn['id'],
**fake_res_assoc_call)
@ -107,7 +107,7 @@ class TestSetResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res = fakes.create_one_resource()
fake_res_assoc = fakes.create_one_resource_association(
fake_res)
self.networkclient.update_bgpvpn_fake_resource_assoc = mock.Mock(
self.networkclient.update_bgpvpn_router_association = mock.Mock(
return_value={fakes.BgpvpnFakeAssoc._resource: fake_res_assoc})
arglist = [
fake_res_assoc['id'],
@ -122,7 +122,7 @@ class TestSetResAssoc(fakes.TestNeutronClientBgpvpn):
result = self.cmd.take_action(parsed_args)
self.networkclient.update_bgpvpn_fake_resource_assoc.\
self.networkclient.update_bgpvpn_router_association.\
assert_not_called()
self.assertIsNone(result)
@ -139,7 +139,7 @@ class TestDeleteResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res = fakes.create_one_resource()
fake_res_assoc = fakes.create_one_resource_association(
fake_res)
self.networkclient.delete_bgpvpn_fake_resource_association = \
self.networkclient.delete_bgpvpn_router_association = \
mock.Mock()
arglist = [
fake_res_assoc['id'],
@ -154,7 +154,7 @@ class TestDeleteResAssoc(fakes.TestNeutronClientBgpvpn):
result = self.cmd.take_action(parsed_args)
self.networkclient.delete_bgpvpn_fake_resource_association.\
self.networkclient.delete_bgpvpn_router_association.\
assert_called_once_with(fake_bgpvpn['id'], fake_res_assoc['id'])
self.assertIsNone(result)
@ -167,7 +167,7 @@ class TestDeleteResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res_assoc_ids = [
fake_res_assoc['id'] for fake_res_assoc in fake_res_assocs
]
self.networkclient.delete_bgpvpn_fake_resource_association = \
self.networkclient.delete_bgpvpn_router_association = \
mock.Mock()
arglist = \
fake_res_assoc_ids + [
@ -182,7 +182,7 @@ class TestDeleteResAssoc(fakes.TestNeutronClientBgpvpn):
result = self.cmd.take_action(parsed_args)
self.networkclient.delete_bgpvpn_fake_resource_association.\
self.networkclient.delete_bgpvpn_router_association.\
assert_has_calls([
mock.call(
fake_bgpvpn['id'], id) for id in fake_res_assoc_ids])
@ -201,7 +201,7 @@ class TestDeleteResAssoc(fakes.TestNeutronClientBgpvpn):
def raise_unknonw_resource(resource_path, name_or_id):
if str(count - 2) in name_or_id:
raise Exception()
self.networkclient.delete_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.delete_bgpvpn_router_association = mock.Mock(
side_effect=raise_unknonw_resource)
arglist = \
fake_res_assoc_ids + [
@ -217,7 +217,7 @@ class TestDeleteResAssoc(fakes.TestNeutronClientBgpvpn):
self.assertRaises(exceptions.CommandError, self.cmd.take_action,
parsed_args)
self.networkclient.delete_bgpvpn_fake_resource_association.\
self.networkclient.delete_bgpvpn_router_association.\
assert_has_calls([
mock.call(fake_bgpvpn['id'], id) for id in fake_res_assoc_ids]
)
@ -236,7 +236,7 @@ class TestListResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res = fakes.create_resources(count=count)
fake_res_assocs = fakes.create_resource_associations(
fake_res)
self.networkclient.bgpvpn_fake_resource_associations = mock.Mock(
self.networkclient.bgpvpn_router_associations = mock.Mock(
return_value=fake_res_assocs)
arglist = [
fake_bgpvpn['id'],
@ -249,7 +249,7 @@ class TestListResAssoc(fakes.TestNeutronClientBgpvpn):
headers, data = self.cmd.take_action(parsed_args)
self.networkclient.bgpvpn_fake_resource_associations.\
self.networkclient.bgpvpn_router_associations.\
assert_called_once_with(fake_bgpvpn['id'], retrieve_all=True)
self.assertEqual(headers, list(headers_short))
self.assertEqual(
@ -263,7 +263,7 @@ class TestListResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res = fakes.create_resources(count=count)
fake_res_assocs = fakes.create_resource_associations(
fake_res)
self.networkclient.bgpvpn_fake_resource_associations = mock.Mock(
self.networkclient.bgpvpn_router_associations = mock.Mock(
return_value=fake_res_assocs)
arglist = [
'--long',
@ -278,7 +278,7 @@ class TestListResAssoc(fakes.TestNeutronClientBgpvpn):
headers, data = self.cmd.take_action(parsed_args)
self.networkclient.bgpvpn_fake_resource_associations.\
self.networkclient.bgpvpn_router_associations.\
assert_called_once_with(fake_bgpvpn['id'], retrieve_all=True)
self.assertEqual(headers, list(headers_long))
self.assertEqual(
@ -299,7 +299,7 @@ class TestShowResAssoc(fakes.TestNeutronClientBgpvpn):
fake_res = fakes.create_one_resource()
fake_res_assoc = fakes.create_one_resource_association(
fake_res)
self.networkclient.get_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.get_bgpvpn_router_association = mock.Mock(
return_value=fake_res_assoc)
arglist = [
fake_res_assoc['id'],
@ -314,7 +314,7 @@ class TestShowResAssoc(fakes.TestNeutronClientBgpvpn):
columns, data = self.cmd.take_action(parsed_args)
self.networkclient.get_bgpvpn_fake_resource_association.\
self.networkclient.get_bgpvpn_router_association.\
assert_called_once_with(fake_bgpvpn['id'], fake_res_assoc['id'])
self.assertEqual(sorted_columns, columns)
self.assertEqual(data, _get_data(fake_res_assoc))

22
neutronclient/tests/unit/osc/v2/networking_bgpvpn/test_router_association.py
View File

@ -101,17 +101,17 @@ class TestCreateRouterAssoc(fakes.TestNeutronClientBgpvpn):
fake_res_assoc_call[key] = value
fake_res_assoc_call.pop('bgpvpn')
self.networkclient.create_bgpvpn_fake_resource_association.\
self.networkclient.create_bgpvpn_router_association.\
assert_called_once_with(
self.fake_bgpvpn['id'],
**fake_res_assoc_call)
return cols, data
def test_create_router_associationx(self):
def test_create_router_association(self):
fake_res_assoc = fakes.create_one_resource_association(
self.fake_router)
self.networkclient.create_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.create_bgpvpn_router_association = mock.Mock(
return_value={
fakes.BgpvpnFakeRouterAssoc._resource: fake_res_assoc,
'advertise_extra_routes': True})
@ -129,7 +129,7 @@ class TestCreateRouterAssoc(fakes.TestNeutronClientBgpvpn):
self.fake_router,
{'advertise_extra_routes': True})
self.networkclient.create_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.create_bgpvpn_router_association = mock.Mock(
return_value=fake_res_assoc)
arglist = self._build_args('--advertise_extra_routes')
@ -145,7 +145,7 @@ class TestCreateRouterAssoc(fakes.TestNeutronClientBgpvpn):
self.fake_router,
{'advertise_extra_routes': False})
self.networkclient.create_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.create_bgpvpn_router_association = mock.Mock(
return_value=fake_res_assoc)
arglist = self._build_args('--no-advertise_extra_routes')
@ -209,7 +209,7 @@ class TestSetRouterAssoc(fakes.TestNeutronClientBgpvpn):
fake_res_assoc = fakes.create_one_resource_association(
self.fake_router,
{'advertise_extra_routes': True})
self.networkclient.update_bgpvpn_fake_resource_association = \
self.networkclient.update_bgpvpn_router_association = \
mock.Mock()
arglist = self._build_args(
@ -223,7 +223,7 @@ class TestSetRouterAssoc(fakes.TestNeutronClientBgpvpn):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.networkclient.update_bgpvpn_fake_resource_association.\
self.networkclient.update_bgpvpn_router_association.\
assert_called_once_with(
self.fake_bgpvpn['id'],
fake_res_assoc['id'],
@ -235,7 +235,7 @@ class TestSetRouterAssoc(fakes.TestNeutronClientBgpvpn):
fake_res_assoc = fakes.create_one_resource_association(
self.fake_router,
{'advertise_extra_routes': False})
self.networkclient.update_bgpvpn_fake_resource_association = \
self.networkclient.update_bgpvpn_router_association = \
mock.Mock()
arglist = self._build_args(
@ -249,7 +249,7 @@ class TestSetRouterAssoc(fakes.TestNeutronClientBgpvpn):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.networkclient.update_bgpvpn_fake_resource_association.\
self.networkclient.update_bgpvpn_router_association.\
assert_called_once_with(
self.fake_bgpvpn['id'],
fake_res_assoc['id'],
@ -271,7 +271,7 @@ class TestShowRouterAssoc(fakes.TestNeutronClientBgpvpn):
fake_res_assoc = fakes.create_one_resource_association(
fake_res,
{'advertise_extra_routes': True})
self.networkclient.get_bgpvpn_fake_resource_association = mock.Mock(
self.networkclient.get_bgpvpn_router_association = mock.Mock(
return_value=fake_res_assoc)
arglist = [
fake_res_assoc['id'],
@ -286,7 +286,7 @@ class TestShowRouterAssoc(fakes.TestNeutronClientBgpvpn):
cols, data = self.cmd.take_action(parsed_args)
self.networkclient.get_bgpvpn_fake_resource_association.\
self.networkclient.get_bgpvpn_router_association.\
assert_called_once_with(fake_bgpvpn['id'], fake_res_assoc['id'])
self.assertEqual(sorted_columns, cols)
self.assertEqual(data, _get_data(fake_res_assoc))