openstack/python-openstackclient
Code Issues Proposed changes

Low-level Compute v2 API: network

Browse Source 
api.compute.APIv2 network functions.

novaclient 8.0 is now released without support for the previously
deprecated nova-net functions, so include a new low-level REST
implementation of the removed APIs.

Change-Id: If230f128e91cda44461fe93c976cac2aecec2252
This commit is contained in:
Dean Troyer 2017-04-17 17:04:43 -05:00
parent 6f473be588
commit b2783dc3c4
7 changed files with 429 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
openstackclient/api/compute_v2.py
View File

@ -175,6 +175,100 @@ class APIv2(api.BaseAPI):
return self.list(url)["floating_ips"]
# Networks
def network_create(
self,
name=None,
subnet=None,
share_subnet=None,
):
"""Create a new network
https://developer.openstack.org/api-ref/compute/#create-project-network
:param string name:
Network label
:param integer subnet:
Subnet for IPv4 fixed addresses in CIDR notation
:param integer share_subnet:
Shared subnet between projects, True or False
:returns: A dict of the network attributes
"""
url = "/os-tenant-networks"
params = {
'label': name,
'cidr': subnet,
'share_address': share_subnet,
}
return self.create(
url,
json={'network': params},
)['network']
def network_delete(
self,
network=None,
):
"""Delete a network
https://developer.openstack.org/api-ref/compute/#delete-project-network
:param string network:
Network name or ID
"""
url = "/os-tenant-networks"
network = self.find(
url,
attr='label',
value=network,
)['id']
if network is not None:
return self.delete('/%s/%s' % (url, network))
return None
def network_find(
self,
network=None,
):
"""Return a network given name or ID
https://developer.openstack.org/api-ref/compute/#show-project-network-details
:param string network:
Network name or ID
:returns: A dict of the network attributes
"""
url = "/os-tenant-networks"
return self.find(
url,
attr='label',
value=network,
)
def network_list(
self,
):
"""Get networks
https://developer.openstack.org/api-ref/compute/#list-project-networks
:returns:
list of networks
"""
url = "/os-tenant-networks"
return self.list(url)["networks"]
# Security Groups
def security_group_create(

10
openstackclient/compute/v2/server.py
View File

@ -215,11 +215,13 @@ class AddFixedIP(command.Command):
server = utils.find_resource(
compute_client.servers, parsed_args.server)
network = utils.find_resource(
compute_client.networks, parsed_args.network)
network = compute_client.api.network_find(parsed_args.network)
server.interface_attach(port_id=None, net_id=network.id,
fixed_ip=parsed_args.fixed_ip_address)
server.interface_attach(
port_id=None,
net_id=network['id'],
fixed_ip=parsed_args.fixed_ip_address,
)
class AddFloatingIP(command.Command):

30
openstackclient/network/v2/network.py
View File

@ -171,13 +171,13 @@ def _add_additional_network_options(parser):
def _get_attrs_compute(client_manager, parsed_args):
attrs = {}
if parsed_args.name is not None:
attrs['label'] = str(parsed_args.name)
attrs['name'] = str(parsed_args.name)
if parsed_args.share:
attrs['share_address'] = True
attrs['share_subnet'] = True
if parsed_args.no_share:
attrs['share_address'] = False
attrs['share_subnet'] = False
if parsed_args.subnet is not None:
attrs['cidr'] = parsed_args.subnet
attrs['subnet'] = parsed_args.subnet
return attrs
@ -302,9 +302,9 @@ class CreateNetwork(common.NetworkAndComputeShowOne):
def take_action_compute(self, client, parsed_args):
attrs = _get_attrs_compute(self.app.client_manager, parsed_args)
obj = client.networks.create(**attrs)
display_columns, columns = _get_columns(obj._info)
data = utils.get_dict_properties(obj._info, columns)
obj = client.api.network_create(**attrs)
display_columns, columns = _get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return (display_columns, data)
@ -330,8 +330,7 @@ class DeleteNetwork(common.NetworkAndComputeDelete):
client.delete_network(obj)
def take_action_compute(self, client, parsed_args):
network = utils.find_resource(client.networks, self.r)
client.networks.delete(network.id)
client.api.network_delete(self.r)
# TODO(sindhu): Use the SDK resource mapped attribute names once the
@ -552,10 +551,10 @@ class ListNetwork(common.NetworkAndComputeLister):
'Subnet',
)
data = client.networks.list()
data = client.api.network_list()
return (column_headers,
(utils.get_item_properties(
(utils.get_dict_properties(
s, columns,
formatters=_formatters,
) for s in data))
@ -683,10 +682,7 @@ class ShowNetwork(common.NetworkAndComputeShowOne):
return (display_columns, data)
def take_action_compute(self, client, parsed_args):
obj = utils.find_resource(
client.networks,
parsed_args.network,
)
display_columns, columns = _get_columns(obj._info)
data = utils.get_dict_properties(obj._info, columns)
obj = client.api.network_find(parsed_args.network)
display_columns, columns = _get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return (display_columns, data)

151
openstackclient/tests/unit/api/test_compute_v2.py
View File

@ -145,6 +145,157 @@ class TestFloatingIP(TestComputeAPIv2):
self.assertEqual(self.LIST_FLOATING_IP_RESP, ret)
class TestNetwork(TestComputeAPIv2):
FAKE_NETWORK_RESP = {
'id': '1',
'label': 'label1',
'cidr': '1.2.3.0/24',
}
FAKE_NETWORK_RESP_2 = {
'id': '2',
'label': 'label2',
'cidr': '4.5.6.0/24',
}
LIST_NETWORK_RESP = [
FAKE_NETWORK_RESP,
FAKE_NETWORK_RESP_2,
]
def test_network_create_default(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/os-tenant-networks',
json={'network': self.FAKE_NETWORK_RESP},
status_code=200,
)
ret = self.api.network_create('label1')
self.assertEqual(self.FAKE_NETWORK_RESP, ret)
def test_network_create_options(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/os-tenant-networks',
json={'network': self.FAKE_NETWORK_RESP},
status_code=200,
)
ret = self.api.network_create(
name='label1',
subnet='1.2.3.0/24',
)
self.assertEqual(self.FAKE_NETWORK_RESP, ret)
def test_network_delete_id(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks/1',
json={'network': self.FAKE_NETWORK_RESP},
status_code=200,
)
self.requests_mock.register_uri(
'DELETE',
FAKE_URL + '/os-tenant-networks/1',
status_code=202,
)
ret = self.api.network_delete('1')
self.assertEqual(202, ret.status_code)
self.assertEqual("", ret.text)
def test_network_delete_name(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks/label1',
status_code=404,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks',
json={'networks': self.LIST_NETWORK_RESP},
status_code=200,
)
self.requests_mock.register_uri(
'DELETE',
FAKE_URL + '/os-tenant-networks/1',
status_code=202,
)
ret = self.api.network_delete('label1')
self.assertEqual(202, ret.status_code)
self.assertEqual("", ret.text)
def test_network_delete_not_found(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks/label3',
status_code=404,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks',
json={'networks': self.LIST_NETWORK_RESP},
status_code=200,
)
self.assertRaises(
osc_lib_exceptions.NotFound,
self.api.network_delete,
'label3',
)
def test_network_find_id(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks/1',
json={'network': self.FAKE_NETWORK_RESP},
status_code=200,
)
ret = self.api.network_find('1')
self.assertEqual(self.FAKE_NETWORK_RESP, ret)
def test_network_find_name(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks/label2',
status_code=404,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks',
json={'networks': self.LIST_NETWORK_RESP},
status_code=200,
)
ret = self.api.network_find('label2')
self.assertEqual(self.FAKE_NETWORK_RESP_2, ret)
def test_network_find_not_found(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks/label3',
status_code=404,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks',
json={'networks': self.LIST_NETWORK_RESP},
status_code=200,
)
self.assertRaises(
osc_lib_exceptions.NotFound,
self.api.network_find,
'label3',
)
def test_network_list_no_options(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-tenant-networks',
json={'networks': self.LIST_NETWORK_RESP},
status_code=200,
)
ret = self.api.network_list()
self.assertEqual(self.LIST_NETWORK_RESP, ret)
class TestSecurityGroup(TestComputeAPIv2):
FAKE_SECURITY_GROUP_RESP = {

8
openstackclient/tests/unit/compute/v2/fakes.py
View File

@ -190,9 +190,6 @@ class FakeComputev2Client(object):
self.floating_ip_pools = mock.Mock()
self.floating_ip_pools.resource_class = fakes.FakeResource(None, {})
self.networks = mock.Mock()
self.networks.resource_class = fakes.FakeResource(None, {})
self.keypairs = mock.Mock()
self.keypairs.resource_class = fakes.FakeResource(None, {})
@ -1155,10 +1152,7 @@ class FakeNetwork(object):
# Overwrite default attributes.
network_attrs.update(attrs)
network = fakes.FakeResource(info=copy.deepcopy(network_attrs),
loaded=True)
return network
return network_attrs
@staticmethod
def create_networks(attrs=None, count=2):

43
openstackclient/tests/unit/compute/v2/test_server.py
View File

@ -91,9 +91,6 @@ class TestServerAddFixedIP(TestServer):
def setUp(self):
super(TestServerAddFixedIP, self).setUp()
# Get a shortcut to the compute client ServerManager Mock
self.networks_mock = self.app.client_manager.compute.networks
# Get the command object to test
self.cmd = server.AddFixedIP(self.app, None)
@ -105,25 +102,30 @@ class TestServerAddFixedIP(TestServer):
def _test_server_add_fixed_ip(self, extralist, fixed_ip_address):
servers = self.setup_servers_mock(count=1)
network = compute_fakes.FakeNetwork.create_one_network()
self.networks_mock.get.return_value = network
with mock.patch(
'openstackclient.api.compute_v2.APIv2.network_find'
) as net_mock:
net_mock.return_value = network
arglist = [
servers[0].id,
network.id,
] + extralist
verifylist = [
('server', servers[0].id),
('network', network.id),
('fixed_ip_address', fixed_ip_address)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
arglist = [
servers[0].id,
network['id'],
] + extralist
verifylist = [
('server', servers[0].id),
('network', network['id']),
('fixed_ip_address', fixed_ip_address),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
result = self.cmd.take_action(parsed_args)
servers[0].interface_attach.assert_called_once_with(
port_id=None, net_id=network.id, fixed_ip=fixed_ip_address
)
self.assertIsNone(result)
servers[0].interface_attach.assert_called_once_with(
port_id=None,
net_id=network['id'],
fixed_ip=fixed_ip_address,
)
self.assertIsNone(result)
def test_server_add_fixed_ip(self):
self._test_server_add_fixed_ip([], None)
@ -138,9 +140,6 @@ class TestServerAddFloatingIP(TestServer):
def setUp(self):
super(TestServerAddFloatingIP, self).setUp()
# Get a shortcut to the compute client ServerManager Mock
self.networks_mock = self.app.client_manager.compute.networks
# Get the command object to test
self.cmd = server.AddFloatingIP(self.app, None)

280
openstackclient/tests/unit/network/v2/test_network_compute.py
View File

@ -32,6 +32,9 @@ class TestNetworkCompute(compute_fakes.TestComputev2):
self.compute = self.app.client_manager.compute
@mock.patch(
'openstackclient.api.compute_v2.APIv2.network_create'
)
class TestCreateNetworkCompute(TestNetworkCompute):
# The network to create.
@ -73,38 +76,38 @@ class TestCreateNetworkCompute(TestNetworkCompute):
)
data = (
_network.bridge,
_network.bridge_interface,
_network.broadcast,
_network.cidr,
_network.cidr_v6,
_network.created_at,
_network.deleted,
_network.deleted_at,
_network.dhcp_server,
_network.dhcp_start,
_network.dns1,
_network.dns2,
_network.enable_dhcp,
_network.gateway,
_network.gateway_v6,
_network.host,
_network.id,
_network.injected,
_network.label,
_network.mtu,
_network.multi_host,
_network.netmask,
_network.netmask_v6,
_network.priority,
_network.project_id,
_network.rxtx_base,
_network.share_address,
_network.updated_at,
_network.vlan,
_network.vpn_private_address,
_network.vpn_public_address,
_network.vpn_public_port,
_network['bridge'],
_network['bridge_interface'],
_network['broadcast'],
_network['cidr'],
_network['cidr_v6'],
_network['created_at'],
_network['deleted'],
_network['deleted_at'],
_network['dhcp_server'],
_network['dhcp_start'],
_network['dns1'],
_network['dns2'],
_network['enable_dhcp'],
_network['gateway'],
_network['gateway_v6'],
_network['host'],
_network['id'],
_network['injected'],
_network['label'],
_network['mtu'],
_network['multi_host'],
_network['netmask'],
_network['netmask_v6'],
_network['priority'],
_network['project_id'],
_network['rxtx_base'],
_network['share_address'],
_network['updated_at'],
_network['vlan'],
_network['vpn_private_address'],
_network['vpn_public_address'],
_network['vpn_public_port'],
)
def setUp(self):
@ -112,40 +115,48 @@ class TestCreateNetworkCompute(TestNetworkCompute):
self.app.client_manager.network_endpoint_enabled = False
self.compute.networks.create.return_value = self._network
# Get the command object to test
self.cmd = network.CreateNetwork(self.app, None)
def test_create_no_options(self):
def test_network_create_no_options(self, net_mock):
net_mock.return_value = self._network
arglist = []
verifylist = []
# Missing required args should raise exception here
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
self.assertRaises(
tests_utils.ParserException,
self.check_parser,
self.cmd,
arglist,
verifylist,
)
def test_create_default_options(self):
def test_network_create_default_options(self, net_mock):
net_mock.return_value = self._network
arglist = [
"--subnet", self._network.cidr,
self._network.label,
"--subnet", self._network['cidr'],
self._network['label'],
]
verifylist = [
('subnet', self._network.cidr),
('name', self._network.label),
('subnet', self._network['cidr']),
('name', self._network['label']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.compute.networks.create.assert_called_once_with(**{
'cidr': self._network.cidr,
'label': self._network.label,
net_mock.assert_called_once_with(**{
'subnet': self._network['cidr'],
'name': self._network['label'],
})
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@mock.patch(
'openstackclient.api.compute_v2.APIv2.network_delete'
)
class TestDeleteNetworkCompute(TestNetworkCompute):
def setUp(self):
@ -156,34 +167,35 @@ class TestDeleteNetworkCompute(TestNetworkCompute):
# The networks to delete
self._networks = compute_fakes.FakeNetwork.create_networks(count=3)
self.compute.networks.delete.return_value = None
# Return value of utils.find_resource()
self.compute.networks.get = \
self.compute.api.network_find = \
compute_fakes.FakeNetwork.get_networks(networks=self._networks)
# Get the command object to test
self.cmd = network.DeleteNetwork(self.app, None)
def test_delete_one_network(self):
def test_network_delete_one(self, net_mock):
net_mock.return_value = mock.Mock(return_value=None)
arglist = [
self._networks[0].label,
self._networks[0]['label'],
]
verifylist = [
('network', [self._networks[0].label]),
('network', [self._networks[0]['label']]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.compute.networks.delete.assert_called_once_with(
self._networks[0].id)
net_mock.assert_called_once_with(
self._networks[0]['label'],
)
self.assertIsNone(result)
def test_delete_multiple_networks(self):
def test_network_delete_multi(self, net_mock):
net_mock.return_value = mock.Mock(return_value=None)
arglist = []
for n in self._networks:
arglist.append(n.label)
arglist.append(n['id'])
verifylist = [
('network', arglist),
]
@ -193,53 +205,40 @@ class TestDeleteNetworkCompute(TestNetworkCompute):
calls = []
for n in self._networks:
calls.append(call(n.id))
self.compute.networks.delete.assert_has_calls(calls)
calls.append(call(n['id']))
net_mock.assert_has_calls(calls)
self.assertIsNone(result)
def test_delete_multiple_networks_exception(self):
def test_network_delete_multi_with_exception(self, net_mock):
net_mock.return_value = mock.Mock(return_value=None)
net_mock.side_effect = ([
mock.Mock(return_value=None),
exceptions.CommandError,
])
arglist = [
self._networks[0].id,
self._networks[0]['id'],
'xxxx-yyyy-zzzz',
self._networks[1].id,
self._networks[1]['id'],
]
verifylist = [
('network', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# Fake exception in utils.find_resource()
# In compute v2, we use utils.find_resource() to find a network.
# It calls get() several times, but find() only one time. So we
# choose to fake get() always raise exception, then pass through.
# And fake find() to find the real network or not.
self.compute.networks.get.side_effect = Exception()
ret_find = [
self._networks[0],
Exception(),
self._networks[1],
]
self.compute.networks.find.side_effect = ret_find
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('2 of 3 networks failed to delete.', str(e))
# Fake exception in delete()
ret_delete = [
None,
Exception(),
]
self.compute.networks.delete = mock.Mock(side_effect=ret_delete)
self.assertRaises(exceptions.CommandError, self.cmd.take_action,
parsed_args)
# The second call of utils.find_resource() should fail. So delete()
# was only called twice.
calls = [
call(self._networks[0].id),
call(self._networks[1].id),
]
self.compute.networks.delete.assert_has_calls(calls)
net_mock.assert_any_call(self._networks[0]['id'])
net_mock.assert_any_call(self._networks[1]['id'])
net_mock.assert_any_call('xxxx-yyyy-zzzz')
@mock.patch(
'openstackclient.api.compute_v2.APIv2.network_list'
)
class TestListNetworkCompute(TestNetworkCompute):
# The networks going to be listed up.
@ -254,9 +253,9 @@ class TestListNetworkCompute(TestNetworkCompute):
data = []
for net in _networks:
data.append((
net.id,
net.label,
net.cidr,
net['id'],
net['label'],
net['cidr'],
))
def setUp(self):
@ -264,12 +263,11 @@ class TestListNetworkCompute(TestNetworkCompute):
self.app.client_manager.network_endpoint_enabled = False
self.compute.networks.list.return_value = self._networks
# Get the command object to test
self.cmd = network.ListNetwork(self.app, None)
def test_network_list_no_options(self):
def test_network_list_no_options(self, net_mock):
net_mock.return_value = self._networks
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -279,11 +277,14 @@ class TestListNetworkCompute(TestNetworkCompute):
# containing the data to be listed.
columns, data = self.cmd.take_action(parsed_args)
self.compute.networks.list.assert_called_once_with()
net_mock.assert_called_once_with()
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
@mock.patch(
'openstackclient.api.compute_v2.APIv2.network_find'
)
class TestShowNetworkCompute(TestNetworkCompute):
# The network to show.
@ -325,38 +326,38 @@ class TestShowNetworkCompute(TestNetworkCompute):
)
data = (
_network.bridge,
_network.bridge_interface,
_network.broadcast,
_network.cidr,
_network.cidr_v6,
_network.created_at,
_network.deleted,
_network.deleted_at,
_network.dhcp_server,
_network.dhcp_start,
_network.dns1,
_network.dns2,
_network.enable_dhcp,
_network.gateway,
_network.gateway_v6,
_network.host,
_network.id,
_network.injected,
_network.label,
_network.mtu,
_network.multi_host,
_network.netmask,
_network.netmask_v6,
_network.priority,
_network.project_id,
_network.rxtx_base,
_network.share_address,
_network.updated_at,
_network.vlan,
_network.vpn_private_address,
_network.vpn_public_address,
_network.vpn_public_port,
_network['bridge'],
_network['bridge_interface'],
_network['broadcast'],
_network['cidr'],
_network['cidr_v6'],
_network['created_at'],
_network['deleted'],
_network['deleted_at'],
_network['dhcp_server'],
_network['dhcp_start'],
_network['dns1'],
_network['dns2'],
_network['enable_dhcp'],
_network['gateway'],
_network['gateway_v6'],
_network['host'],
_network['id'],
_network['injected'],
_network['label'],
_network['mtu'],
_network['multi_host'],
_network['netmask'],
_network['netmask_v6'],
_network['priority'],
_network['project_id'],
_network['rxtx_base'],
_network['share_address'],
_network['updated_at'],
_network['vlan'],
_network['vpn_private_address'],
_network['vpn_public_address'],
_network['vpn_public_port'],
)
def setUp(self):
@ -364,29 +365,34 @@ class TestShowNetworkCompute(TestNetworkCompute):
self.app.client_manager.network_endpoint_enabled = False
# Return value of utils.find_resource()
self.compute.networks.get.return_value = self._network
# Get the command object to test
self.cmd = network.ShowNetwork(self.app, None)
def test_show_no_options(self):
def test_show_no_options(self, net_mock):
net_mock.return_value = self._network
arglist = []
verifylist = []
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
self.assertRaises(
tests_utils.ParserException,
self.check_parser,
self.cmd,
arglist,
verifylist,
)
def test_show_all_options(self):
def test_show_all_options(self, net_mock):
net_mock.return_value = self._network
arglist = [
self._network.label,
self._network['label'],
]
verifylist = [
('network', self._network.label),
('network', self._network['label']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
net_mock.assert_called_once_with(self._network['label'])
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)