Support Neutron Address Group CRUD
Add support for Neutron address group CRUD operations. Subsequent patches will be added to use address groups in security group rules. Change-Id: I3c313fc9329837dde67815901528a34dca98ebcc Implements: blueprint address-groups-in-sg-rules Depends-On: https://review.opendev.org/738274 Depends-On: https://review.opendev.org/745594
This commit is contained in:
Hang Yang
parent
fe98069c20
commit
f57e10b903
12
doc/source/cli/command-objects/address-group.rst
Normal file
12
doc/source/cli/command-objects/address-group.rst
Normal file
@ -0,0 +1,12 @@
|
||||
=============
|
||||
address group
|
||||
=============
|
||||
|
||||
An **address group** is a group of IPv4 or IPv6 address blocks which could be
|
||||
referenced as a remote source or destination when creating a security group
|
||||
rule.
|
||||
|
||||
Network v2
|
||||
|
||||
.. autoprogram-cliff:: openstack.network.v2
|
||||
:command: address group *
|
||||
296
openstackclient/network/v2/address_group.py
Normal file
296
openstackclient/network/v2/address_group.py
Normal file
@ -0,0 +1,296 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
#
|
||||
|
||||
"""Address group action implementations"""
|
||||
|
||||
import logging
|
||||
|
||||
import netaddr
|
||||
from osc_lib.command import command
|
||||
from osc_lib import exceptions
|
||||
from osc_lib import utils
|
||||
|
||||
from openstackclient.i18n import _
|
||||
from openstackclient.identity import common as identity_common
|
||||
from openstackclient.network import sdk_utils
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _get_columns(item):
|
||||
column_map = {
|
||||
'tenant_id': 'project_id',
|
||||
}
|
||||
return sdk_utils.get_osc_show_columns_for_sdk_resource(item, column_map)
|
||||
|
||||
|
||||
def _format_addresses(addresses):
|
||||
ret = []
|
||||
for addr in addresses:
|
||||
ret.append(str(netaddr.IPNetwork(addr)))
|
||||
return ret
|
||||
|
||||
|
||||
def _get_attrs(client_manager, parsed_args):
|
||||
attrs = {}
|
||||
attrs['name'] = parsed_args.name
|
||||
if parsed_args.description:
|
||||
attrs['description'] = parsed_args.description
|
||||
attrs['addresses'] = _format_addresses(parsed_args.address)
|
||||
if 'project' in parsed_args and parsed_args.project is not None:
|
||||
identity_client = client_manager.identity
|
||||
project_id = identity_common.find_project(
|
||||
identity_client,
|
||||
parsed_args.project,
|
||||
parsed_args.project_domain,
|
||||
).id
|
||||
attrs['tenant_id'] = project_id
|
||||
|
||||
return attrs
|
||||
|
||||
|
||||
class CreateAddressGroup(command.ShowOne):
|
||||
_description = _("Create a new Address Group")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(CreateAddressGroup, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'name',
|
||||
metavar="<name>",
|
||||
help=_("New address group name")
|
||||
)
|
||||
parser.add_argument(
|
||||
'--description',
|
||||
metavar="<description>",
|
||||
help=_("New address group description")
|
||||
)
|
||||
parser.add_argument(
|
||||
"--address",
|
||||
metavar="<ip-address>",
|
||||
action='append',
|
||||
default=[],
|
||||
help=_("IP address or CIDR "
|
||||
"(repeat option to set multiple addresses)"),
|
||||
)
|
||||
parser.add_argument(
|
||||
'--project',
|
||||
metavar="<project>",
|
||||
help=_("Owner's project (name or ID)")
|
||||
)
|
||||
identity_common.add_project_domain_option_to_parser(parser)
|
||||
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
client = self.app.client_manager.network
|
||||
attrs = _get_attrs(self.app.client_manager, parsed_args)
|
||||
|
||||
obj = client.create_address_group(**attrs)
|
||||
display_columns, columns = _get_columns(obj)
|
||||
data = utils.get_item_properties(obj, columns, formatters={})
|
||||
|
||||
return (display_columns, data)
|
||||
|
||||
|
||||
class DeleteAddressGroup(command.Command):
|
||||
_description = _("Delete address group(s)")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(DeleteAddressGroup, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'address_group',
|
||||
metavar="<address-group>",
|
||||
nargs='+',
|
||||
help=_("Address group(s) to delete (name or ID)")
|
||||
)
|
||||
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
client = self.app.client_manager.network
|
||||
result = 0
|
||||
|
||||
for group in parsed_args.address_group:
|
||||
try:
|
||||
obj = client.find_address_group(group, ignore_missing=False)
|
||||
client.delete_address_group(obj)
|
||||
except Exception as e:
|
||||
result += 1
|
||||
LOG.error(_("Failed to delete address group with "
|
||||
"name or ID '%(group)s': %(e)s"),
|
||||
{'group': group, 'e': e})
|
||||
|
||||
if result > 0:
|
||||
total = len(parsed_args.address_group)
|
||||
msg = (_("%(result)s of %(total)s address groups failed "
|
||||
"to delete.") % {'result': result, 'total': total})
|
||||
raise exceptions.CommandError(msg)
|
||||
|
||||
|
||||
class ListAddressGroup(command.Lister):
|
||||
_description = _("List address groups")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(ListAddressGroup, self).get_parser(prog_name)
|
||||
|
||||
parser.add_argument(
|
||||
'--name',
|
||||
metavar='<name>',
|
||||
help=_("List only address groups of given name in output")
|
||||
)
|
||||
parser.add_argument(
|
||||
'--project',
|
||||
metavar="<project>",
|
||||
help=_("List address groups according to their project "
|
||||
"(name or ID)")
|
||||
)
|
||||
identity_common.add_project_domain_option_to_parser(parser)
|
||||
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
client = self.app.client_manager.network
|
||||
columns = (
|
||||
'id',
|
||||
'name',
|
||||
'description',
|
||||
'project_id',
|
||||
'addresses',
|
||||
)
|
||||
column_headers = (
|
||||
'ID',
|
||||
'Name',
|
||||
'Description',
|
||||
'Project',
|
||||
'Addresses',
|
||||
)
|
||||
attrs = {}
|
||||
if parsed_args.name:
|
||||
attrs['name'] = parsed_args.name
|
||||
if 'project' in parsed_args and parsed_args.project is not None:
|
||||
identity_client = self.app.client_manager.identity
|
||||
project_id = identity_common.find_project(
|
||||
identity_client,
|
||||
parsed_args.project,
|
||||
parsed_args.project_domain,
|
||||
).id
|
||||
attrs['tenant_id'] = project_id
|
||||
attrs['project_id'] = project_id
|
||||
data = client.address_groups(**attrs)
|
||||
|
||||
return (column_headers,
|
||||
(utils.get_item_properties(
|
||||
s, columns, formatters={},
|
||||
) for s in data))
|
||||
|
||||
|
||||
class SetAddressGroup(command.Command):
|
||||
_description = _("Set address group properties")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(SetAddressGroup, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'address_group',
|
||||
metavar="<address-group>",
|
||||
help=_("Address group to modify (name or ID)")
|
||||
)
|
||||
parser.add_argument(
|
||||
'--name',
|
||||
metavar="<name>",
|
||||
help=_('Set address group name')
|
||||
)
|
||||
parser.add_argument(
|
||||
'--description',
|
||||
metavar="<description>",
|
||||
help=_('Set address group description')
|
||||
)
|
||||
parser.add_argument(
|
||||
"--address",
|
||||
metavar="<ip-address>",
|
||||
action='append',
|
||||
default=[],
|
||||
help=_("IP address or CIDR "
|
||||
"(repeat option to set multiple addresses)"),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
client = self.app.client_manager.network
|
||||
obj = client.find_address_group(
|
||||
parsed_args.address_group,
|
||||
ignore_missing=False)
|
||||
attrs = {}
|
||||
if parsed_args.name is not None:
|
||||
attrs['name'] = parsed_args.name
|
||||
if parsed_args.description is not None:
|
||||
attrs['description'] = parsed_args.description
|
||||
if attrs:
|
||||
client.update_address_group(obj, **attrs)
|
||||
if parsed_args.address:
|
||||
client.add_addresses_to_address_group(
|
||||
obj, _format_addresses(parsed_args.address))
|
||||
|
||||
|
||||
class ShowAddressGroup(command.ShowOne):
|
||||
_description = _("Display address group details")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(ShowAddressGroup, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'address_group',
|
||||
metavar="<address-group>",
|
||||
help=_("Address group to display (name or ID)")
|
||||
)
|
||||
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
client = self.app.client_manager.network
|
||||
obj = client.find_address_group(
|
||||
parsed_args.address_group,
|
||||
ignore_missing=False)
|
||||
display_columns, columns = _get_columns(obj)
|
||||
data = utils.get_item_properties(obj, columns, formatters={})
|
||||
|
||||
return (display_columns, data)
|
||||
|
||||
|
||||
class UnsetAddressGroup(command.Command):
|
||||
_description = _("Unset address group properties")
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(UnsetAddressGroup, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'address_group',
|
||||
metavar="<address-group>",
|
||||
help=_("Address group to modify (name or ID)")
|
||||
)
|
||||
parser.add_argument(
|
||||
"--address",
|
||||
metavar="<ip-address>",
|
||||
action='append',
|
||||
default=[],
|
||||
help=_("IP address or CIDR "
|
||||
"(repeat option to unset multiple addresses)"),
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
client = self.app.client_manager.network
|
||||
obj = client.find_address_group(
|
||||
parsed_args.address_group,
|
||||
ignore_missing=False)
|
||||
if parsed_args.address:
|
||||
client.remove_addresses_from_address_group(
|
||||
obj, _format_addresses(parsed_args.address))
|
||||
@ -0,0 +1,177 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from openstackclient.tests.functional.network.v2 import common
|
||||
|
||||
|
||||
class AddressGroupTests(common.NetworkTests):
|
||||
"""Functional tests for address group"""
|
||||
|
||||
def setUp(self):
|
||||
super(AddressGroupTests, self).setUp()
|
||||
# Nothing in this class works with Nova Network
|
||||
if not self.haz_network:
|
||||
self.skipTest("No Network service present")
|
||||
if not self.is_extension_enabled('address-group'):
|
||||
self.skipTest("No address-group extension present")
|
||||
|
||||
def test_address_group_create_and_delete(self):
|
||||
"""Test create, delete multiple"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group create -f json ' +
|
||||
name1
|
||||
))
|
||||
self.assertEqual(
|
||||
name1,
|
||||
cmd_output['name'],
|
||||
)
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group create -f json ' +
|
||||
name2
|
||||
))
|
||||
self.assertEqual(
|
||||
name2,
|
||||
cmd_output['name'],
|
||||
)
|
||||
|
||||
raw_output = self.openstack(
|
||||
'address group delete ' + name1 + ' ' + name2,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
def test_address_group_list(self):
|
||||
"""Test create, list filters, delete"""
|
||||
# Get project IDs
|
||||
cmd_output = json.loads(self.openstack('token issue -f json '))
|
||||
auth_project_id = cmd_output['project_id']
|
||||
|
||||
cmd_output = json.loads(self.openstack('project list -f json '))
|
||||
admin_project_id = None
|
||||
demo_project_id = None
|
||||
for p in cmd_output:
|
||||
if p['Name'] == 'admin':
|
||||
admin_project_id = p['ID']
|
||||
if p['Name'] == 'demo':
|
||||
demo_project_id = p['ID']
|
||||
|
||||
# Verify assumptions:
|
||||
# * admin and demo projects are present
|
||||
# * demo and admin are distinct projects
|
||||
# * tests run as admin
|
||||
self.assertIsNotNone(admin_project_id)
|
||||
self.assertIsNotNone(demo_project_id)
|
||||
self.assertNotEqual(admin_project_id, demo_project_id)
|
||||
self.assertEqual(admin_project_id, auth_project_id)
|
||||
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group create -f json ' +
|
||||
name1
|
||||
))
|
||||
self.addCleanup(self.openstack, 'address group delete ' + name1)
|
||||
self.assertEqual(
|
||||
admin_project_id,
|
||||
cmd_output["project_id"],
|
||||
)
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group create -f json ' +
|
||||
'--project ' + demo_project_id +
|
||||
' ' + name2
|
||||
))
|
||||
self.addCleanup(self.openstack, 'address group delete ' + name2)
|
||||
self.assertEqual(
|
||||
demo_project_id,
|
||||
cmd_output["project_id"],
|
||||
)
|
||||
|
||||
# Test list
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group list -f json ',
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# Test list --project
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group list -f json ' +
|
||||
'--project ' + demo_project_id
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertNotIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# Test list --name
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group list -f json ' +
|
||||
'--name ' + name1
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertIn(name1, names)
|
||||
self.assertNotIn(name2, names)
|
||||
|
||||
def test_address_group_set_unset_and_show(self):
|
||||
"""Tests create options, set, unset, and show"""
|
||||
name = uuid.uuid4().hex
|
||||
newname = name + "_"
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group create -f json ' +
|
||||
'--description aaaa ' +
|
||||
'--address 10.0.0.1 --address 2001::/16 ' +
|
||||
name
|
||||
))
|
||||
self.addCleanup(self.openstack, 'address group delete ' + newname)
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
self.assertEqual('aaaa', cmd_output['description'])
|
||||
self.assertEqual(2, len(cmd_output['addresses']))
|
||||
|
||||
# Test set name, description and address
|
||||
raw_output = self.openstack(
|
||||
'address group set ' +
|
||||
'--name ' + newname + ' ' +
|
||||
'--description bbbb ' +
|
||||
'--address 10.0.0.2 --address 192.0.0.0/8 ' +
|
||||
name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
# Show the updated address group
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group show -f json ' +
|
||||
newname,
|
||||
))
|
||||
self.assertEqual(newname, cmd_output['name'])
|
||||
self.assertEqual('bbbb', cmd_output['description'])
|
||||
self.assertEqual(4, len(cmd_output['addresses']))
|
||||
|
||||
# Test unset address
|
||||
raw_output = self.openstack(
|
||||
'address group unset ' +
|
||||
'--address 10.0.0.1 --address 2001::/16 ' +
|
||||
'--address 10.0.0.2 --address 192.0.0.0/8 ' +
|
||||
newname,
|
||||
)
|
||||
self.assertEqual('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'address group show -f json ' +
|
||||
newname,
|
||||
))
|
||||
self.assertEqual(0, len(cmd_output['addresses']))
|
||||
@ -83,6 +83,79 @@ class TestNetworkV2(utils.TestCommand):
|
||||
)
|
||||
|
||||
|
||||
class FakeAddressGroup(object):
|
||||
"""Fake one or more address groups."""
|
||||
|
||||
@staticmethod
|
||||
def create_one_address_group(attrs=None):
|
||||
"""Create a fake address group.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:return:
|
||||
A FakeResource object with name, id, etc.
|
||||
"""
|
||||
attrs = attrs or {}
|
||||
|
||||
# Set default attributes.
|
||||
address_group_attrs = {
|
||||
'name': 'address-group-name-' + uuid.uuid4().hex,
|
||||
'description': 'address-group-description-' + uuid.uuid4().hex,
|
||||
'id': 'address-group-id-' + uuid.uuid4().hex,
|
||||
'tenant_id': 'project-id-' + uuid.uuid4().hex,
|
||||
'addresses': ['10.0.0.1/32'],
|
||||
}
|
||||
|
||||
# Overwrite default attributes.
|
||||
address_group_attrs.update(attrs)
|
||||
|
||||
address_group = fakes.FakeResource(
|
||||
info=copy.deepcopy(address_group_attrs),
|
||||
loaded=True)
|
||||
|
||||
# Set attributes with special mapping in OpenStack SDK.
|
||||
address_group.project_id = address_group_attrs['tenant_id']
|
||||
|
||||
return address_group
|
||||
|
||||
@staticmethod
|
||||
def create_address_groups(attrs=None, count=2):
|
||||
"""Create multiple fake address groups.
|
||||
|
||||
:param Dictionary attrs:
|
||||
A dictionary with all attributes
|
||||
:param int count:
|
||||
The number of address groups to fake
|
||||
:return:
|
||||
A list of FakeResource objects faking the address groups
|
||||
"""
|
||||
address_groups = []
|
||||
for i in range(0, count):
|
||||
address_groups.append(
|
||||
FakeAddressGroup.create_one_address_group(attrs))
|
||||
|
||||
return address_groups
|
||||
|
||||
@staticmethod
|
||||
def get_address_groups(address_groups=None, count=2):
|
||||
"""Get an iterable Mock object with a list of faked address groups.
|
||||
|
||||
If address groups list is provided, then initialize the Mock object
|
||||
with the list. Otherwise create one.
|
||||
|
||||
:param List address_groups:
|
||||
A list of FakeResource objects faking address groups
|
||||
:param int count:
|
||||
The number of address groups to fake
|
||||
:return:
|
||||
An iterable Mock object with side_effect set to a list of faked
|
||||
address groups
|
||||
"""
|
||||
if address_groups is None:
|
||||
address_groups = FakeAddressGroup.create_address_groups(count)
|
||||
return mock.Mock(side_effect=address_groups)
|
||||
|
||||
|
||||
class FakeAddressScope(object):
|
||||
"""Fake one or more address scopes."""
|
||||
|
||||
|
||||
503
openstackclient/tests/unit/network/v2/test_address_group.py
Normal file
503
openstackclient/tests/unit/network/v2/test_address_group.py
Normal file
@ -0,0 +1,503 @@
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
#
|
||||
|
||||
from unittest import mock
|
||||
from unittest.mock import call
|
||||
|
||||
from osc_lib import exceptions
|
||||
|
||||
from openstackclient.network.v2 import address_group
|
||||
from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes_v3
|
||||
from openstackclient.tests.unit.network.v2 import fakes as network_fakes
|
||||
from openstackclient.tests.unit import utils as tests_utils
|
||||
|
||||
|
||||
class TestAddressGroup(network_fakes.TestNetworkV2):
|
||||
|
||||
def setUp(self):
|
||||
super(TestAddressGroup, self).setUp()
|
||||
|
||||
# Get a shortcut to the network client
|
||||
self.network = self.app.client_manager.network
|
||||
# Get a shortcut to the ProjectManager Mock
|
||||
self.projects_mock = self.app.client_manager.identity.projects
|
||||
# Get a shortcut to the DomainManager Mock
|
||||
self.domains_mock = self.app.client_manager.identity.domains
|
||||
|
||||
|
||||
class TestCreateAddressGroup(TestAddressGroup):
|
||||
|
||||
project = identity_fakes_v3.FakeProject.create_one_project()
|
||||
domain = identity_fakes_v3.FakeDomain.create_one_domain()
|
||||
# The new address group created.
|
||||
new_address_group = (
|
||||
network_fakes.FakeAddressGroup.create_one_address_group(
|
||||
attrs={
|
||||
'tenant_id': project.id,
|
||||
}
|
||||
))
|
||||
columns = (
|
||||
'addresses',
|
||||
'description',
|
||||
'id',
|
||||
'name',
|
||||
'project_id',
|
||||
)
|
||||
data = (
|
||||
new_address_group.addresses,
|
||||
new_address_group.description,
|
||||
new_address_group.id,
|
||||
new_address_group.name,
|
||||
new_address_group.project_id,
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
super(TestCreateAddressGroup, self).setUp()
|
||||
self.network.create_address_group = mock.Mock(
|
||||
return_value=self.new_address_group)
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = address_group.CreateAddressGroup(self.app, self.namespace)
|
||||
|
||||
self.projects_mock.get.return_value = self.project
|
||||
self.domains_mock.get.return_value = self.domain
|
||||
|
||||
def test_create_no_options(self):
|
||||
arglist = []
|
||||
verifylist = []
|
||||
|
||||
# Missing required args should bail here
|
||||
self.assertRaises(tests_utils.ParserException, self.check_parser,
|
||||
self.cmd, arglist, verifylist)
|
||||
|
||||
def test_create_default_options(self):
|
||||
arglist = [
|
||||
self.new_address_group.name,
|
||||
]
|
||||
verifylist = [
|
||||
('project', None),
|
||||
('name', self.new_address_group.name),
|
||||
('description', None),
|
||||
('address', []),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = (self.cmd.take_action(parsed_args))
|
||||
|
||||
self.network.create_address_group.assert_called_once_with(**{
|
||||
'name': self.new_address_group.name,
|
||||
'addresses': [],
|
||||
})
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, data)
|
||||
|
||||
def test_create_all_options(self):
|
||||
arglist = [
|
||||
'--project', self.project.name,
|
||||
'--project-domain', self.domain.name,
|
||||
'--address', '10.0.0.1',
|
||||
'--description', self.new_address_group.description,
|
||||
self.new_address_group.name,
|
||||
]
|
||||
verifylist = [
|
||||
('project', self.project.name),
|
||||
('project_domain', self.domain.name),
|
||||
('address', ['10.0.0.1']),
|
||||
('description', self.new_address_group.description),
|
||||
('name', self.new_address_group.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = (self.cmd.take_action(parsed_args))
|
||||
|
||||
self.network.create_address_group.assert_called_once_with(**{
|
||||
'addresses': ['10.0.0.1/32'],
|
||||
'tenant_id': self.project.id,
|
||||
'name': self.new_address_group.name,
|
||||
'description': self.new_address_group.description,
|
||||
})
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, data)
|
||||
|
||||
|
||||
class TestDeleteAddressGroup(TestAddressGroup):
|
||||
|
||||
# The address group to delete.
|
||||
_address_groups = (
|
||||
network_fakes.FakeAddressGroup.create_address_groups(count=2))
|
||||
|
||||
def setUp(self):
|
||||
super(TestDeleteAddressGroup, self).setUp()
|
||||
self.network.delete_address_group = mock.Mock(return_value=None)
|
||||
self.network.find_address_group = (
|
||||
network_fakes.FakeAddressGroup.get_address_groups(
|
||||
address_groups=self._address_groups)
|
||||
)
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = address_group.DeleteAddressGroup(self.app, self.namespace)
|
||||
|
||||
def test_address_group_delete(self):
|
||||
arglist = [
|
||||
self._address_groups[0].name,
|
||||
]
|
||||
verifylist = [
|
||||
('address_group', [self._address_groups[0].name]),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.network.find_address_group.assert_called_once_with(
|
||||
self._address_groups[0].name, ignore_missing=False)
|
||||
self.network.delete_address_group.assert_called_once_with(
|
||||
self._address_groups[0])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_multi_address_groups_delete(self):
|
||||
arglist = []
|
||||
|
||||
for a in self._address_groups:
|
||||
arglist.append(a.name)
|
||||
verifylist = [
|
||||
('address_group', arglist),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
|
||||
calls = []
|
||||
for a in self._address_groups:
|
||||
calls.append(call(a))
|
||||
self.network.delete_address_group.assert_has_calls(calls)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_multi_address_groups_delete_with_exception(self):
|
||||
arglist = [
|
||||
self._address_groups[0].name,
|
||||
'unexist_address_group',
|
||||
]
|
||||
verifylist = [
|
||||
('address_group',
|
||||
[self._address_groups[0].name, 'unexist_address_group']),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
find_mock_result = [self._address_groups[0], exceptions.CommandError]
|
||||
self.network.find_address_group = (
|
||||
mock.Mock(side_effect=find_mock_result)
|
||||
)
|
||||
|
||||
try:
|
||||
self.cmd.take_action(parsed_args)
|
||||
self.fail('CommandError should be raised.')
|
||||
except exceptions.CommandError as e:
|
||||
self.assertEqual('1 of 2 address groups failed to delete.', str(e))
|
||||
|
||||
self.network.find_address_group.assert_any_call(
|
||||
self._address_groups[0].name, ignore_missing=False)
|
||||
self.network.find_address_group.assert_any_call(
|
||||
'unexist_address_group', ignore_missing=False)
|
||||
self.network.delete_address_group.assert_called_once_with(
|
||||
self._address_groups[0]
|
||||
)
|
||||
|
||||
|
||||
class TestListAddressGroup(TestAddressGroup):
|
||||
|
||||
# The address groups to list up.
|
||||
address_groups = (
|
||||
network_fakes.FakeAddressGroup.create_address_groups(count=3))
|
||||
columns = (
|
||||
'ID',
|
||||
'Name',
|
||||
'Description',
|
||||
'Project',
|
||||
'Addresses',
|
||||
)
|
||||
data = []
|
||||
for group in address_groups:
|
||||
data.append((
|
||||
group.id,
|
||||
group.name,
|
||||
group.description,
|
||||
group.project_id,
|
||||
group.addresses,
|
||||
))
|
||||
|
||||
def setUp(self):
|
||||
super(TestListAddressGroup, self).setUp()
|
||||
self.network.address_groups = mock.Mock(
|
||||
return_value=self.address_groups)
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = address_group.ListAddressGroup(self.app, self.namespace)
|
||||
|
||||
def test_address_group_list(self):
|
||||
arglist = []
|
||||
verifylist = []
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.network.address_groups.assert_called_once_with(**{})
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, list(data))
|
||||
|
||||
def test_address_group_list_name(self):
|
||||
arglist = [
|
||||
'--name', self.address_groups[0].name,
|
||||
]
|
||||
verifylist = [
|
||||
('name', self.address_groups[0].name),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.network.address_groups.assert_called_once_with(
|
||||
**{'name': self.address_groups[0].name})
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, list(data))
|
||||
|
||||
def test_address_group_list_project(self):
|
||||
project = identity_fakes_v3.FakeProject.create_one_project()
|
||||
self.projects_mock.get.return_value = project
|
||||
arglist = [
|
||||
'--project', project.id,
|
||||
]
|
||||
verifylist = [
|
||||
('project', project.id),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.network.address_groups.assert_called_once_with(
|
||||
**{'tenant_id': project.id, 'project_id': project.id})
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, list(data))
|
||||
|
||||
def test_address_group_project_domain(self):
|
||||
project = identity_fakes_v3.FakeProject.create_one_project()
|
||||
self.projects_mock.get.return_value = project
|
||||
arglist = [
|
||||
'--project', project.id,
|
||||
'--project-domain', project.domain_id,
|
||||
]
|
||||
verifylist = [
|
||||
('project', project.id),
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
filters = {'tenant_id': project.id, 'project_id': project.id}
|
||||
|
||||
self.network.address_groups.assert_called_once_with(**filters)
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, list(data))
|
||||
|
||||
|
||||
class TestSetAddressGroup(TestAddressGroup):
|
||||
|
||||
# The address group to set.
|
||||
_address_group = network_fakes.FakeAddressGroup.create_one_address_group()
|
||||
|
||||
def setUp(self):
|
||||
super(TestSetAddressGroup, self).setUp()
|
||||
self.network.update_address_group = mock.Mock(return_value=None)
|
||||
self.network.find_address_group = mock.Mock(
|
||||
return_value=self._address_group)
|
||||
self.network.add_addresses_to_address_group = mock.Mock(
|
||||
return_value=self._address_group)
|
||||
# Get the command object to test
|
||||
self.cmd = address_group.SetAddressGroup(self.app, self.namespace)
|
||||
|
||||
def test_set_nothing(self):
|
||||
arglist = [self._address_group.name, ]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.network.update_address_group.assert_not_called()
|
||||
self.network.add_addresses_to_address_group.assert_not_called()
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_set_name_and_description(self):
|
||||
arglist = [
|
||||
'--name', 'new_address_group_name',
|
||||
'--description', 'new_address_group_description',
|
||||
self._address_group.name,
|
||||
]
|
||||
verifylist = [
|
||||
('name', 'new_address_group_name'),
|
||||
('description', 'new_address_group_description'),
|
||||
('address_group', self._address_group.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
attrs = {
|
||||
'name': "new_address_group_name",
|
||||
'description': 'new_address_group_description',
|
||||
}
|
||||
self.network.update_address_group.assert_called_with(
|
||||
self._address_group, **attrs)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_set_one_address(self):
|
||||
arglist = [
|
||||
self._address_group.name,
|
||||
'--address', '10.0.0.2',
|
||||
]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
('address', ['10.0.0.2']),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.network.add_addresses_to_address_group.assert_called_once_with(
|
||||
self._address_group, ['10.0.0.2/32'])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_set_multiple_addresses(self):
|
||||
arglist = [
|
||||
self._address_group.name,
|
||||
'--address', '10.0.0.2',
|
||||
'--address', '2001::/16',
|
||||
]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
('address', ['10.0.0.2', '2001::/16']),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.network.add_addresses_to_address_group.assert_called_once_with(
|
||||
self._address_group, ['10.0.0.2/32', '2001::/16'])
|
||||
self.assertIsNone(result)
|
||||
|
||||
|
||||
class TestShowAddressGroup(TestAddressGroup):
|
||||
|
||||
# The address group to show.
|
||||
_address_group = network_fakes.FakeAddressGroup.create_one_address_group()
|
||||
columns = (
|
||||
'addresses',
|
||||
'description',
|
||||
'id',
|
||||
'name',
|
||||
'project_id',
|
||||
)
|
||||
data = (
|
||||
_address_group.addresses,
|
||||
_address_group.description,
|
||||
_address_group.id,
|
||||
_address_group.name,
|
||||
_address_group.project_id,
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
super(TestShowAddressGroup, self).setUp()
|
||||
self.network.find_address_group = mock.Mock(
|
||||
return_value=self._address_group)
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = address_group.ShowAddressGroup(self.app, self.namespace)
|
||||
|
||||
def test_show_no_options(self):
|
||||
arglist = []
|
||||
verifylist = []
|
||||
|
||||
# Missing required args should bail here
|
||||
self.assertRaises(tests_utils.ParserException, self.check_parser,
|
||||
self.cmd, arglist, verifylist)
|
||||
|
||||
def test_show_all_options(self):
|
||||
arglist = [
|
||||
self._address_group.name,
|
||||
]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.network.find_address_group.assert_called_once_with(
|
||||
self._address_group.name, ignore_missing=False)
|
||||
self.assertEqual(self.columns, columns)
|
||||
self.assertItemsEqual(self.data, list(data))
|
||||
|
||||
|
||||
class TestUnsetAddressGroup(TestAddressGroup):
|
||||
|
||||
# The address group to unset.
|
||||
_address_group = network_fakes.FakeAddressGroup.create_one_address_group()
|
||||
|
||||
def setUp(self):
|
||||
super(TestUnsetAddressGroup, self).setUp()
|
||||
self.network.find_address_group = mock.Mock(
|
||||
return_value=self._address_group)
|
||||
self.network.remove_addresses_from_address_group = mock.Mock(
|
||||
return_value=self._address_group)
|
||||
# Get the command object to test
|
||||
self.cmd = address_group.UnsetAddressGroup(self.app, self.namespace)
|
||||
|
||||
def test_unset_nothing(self):
|
||||
arglist = [self._address_group.name, ]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.network.remove_addresses_from_address_group.assert_not_called()
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_unset_one_address(self):
|
||||
arglist = [
|
||||
self._address_group.name,
|
||||
'--address', '10.0.0.2',
|
||||
]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
('address', ['10.0.0.2']),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.network.remove_addresses_from_address_group.\
|
||||
assert_called_once_with(self._address_group, ['10.0.0.2/32'])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_unset_multiple_addresses(self):
|
||||
arglist = [
|
||||
self._address_group.name,
|
||||
'--address', '10.0.0.2',
|
||||
'--address', '2001::/16',
|
||||
]
|
||||
verifylist = [
|
||||
('address_group', self._address_group.name),
|
||||
('address', ['10.0.0.2', '2001::/16']),
|
||||
]
|
||||
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.take_action(parsed_args)
|
||||
self.network.remove_addresses_from_address_group.\
|
||||
assert_called_once_with(self._address_group,
|
||||
['10.0.0.2/32', '2001::/16'])
|
||||
self.assertIsNone(result)
|
||||
@ -0,0 +1,7 @@
|
||||
---
|
||||
features:
|
||||
- Add ``address group create``, ``address group delete``,
|
||||
``address group list``, ``address group set``, ``address group show``
|
||||
and ``address group unset`` commands to support Neutron address group
|
||||
CRUD operations.
|
||||
[Blueprint `address-groups-in-sg-rules <https://blueprints.launchpad.net/neutron/+spec/address-groups-in-sg-rules>`_]
|
||||
@ -377,6 +377,13 @@ openstack.image.v2 =
|
||||
image_unset = openstackclient.image.v2.image:UnsetImage
|
||||
|
||||
openstack.network.v2 =
|
||||
address_group_create = openstackclient.network.v2.address_group:CreateAddressGroup
|
||||
address_group_delete = openstackclient.network.v2.address_group:DeleteAddressGroup
|
||||
address_group_list = openstackclient.network.v2.address_group:ListAddressGroup
|
||||
address_group_set = openstackclient.network.v2.address_group:SetAddressGroup
|
||||
address_group_show = openstackclient.network.v2.address_group:ShowAddressGroup
|
||||
address_group_unset = openstackclient.network.v2.address_group:UnsetAddressGroup
|
||||
|
||||
address_scope_create = openstackclient.network.v2.address_scope:CreateAddressScope
|
||||
address_scope_delete = openstackclient.network.v2.address_scope:DeleteAddressScope
|
||||
address_scope_list = openstackclient.network.v2.address_scope:ListAddressScope
|
||||
|
||||
Loading…
Reference in New Issue
Block a user