openstack/python-openstackclient
Code Issues Proposed changes

Begin low-level API for Network v2

Browse Source 
api.network.APIv2 starts with network_list() support to flush out
the skeleton of the Network API.

list_dhcp_agent() supports the --dhcp option of 'network list'

Change-Id: I9a2b90cde84eced1f2ea6a014b769e2bae668211
This commit is contained in:
Dean Troyer 2014-09-18 10:35:15 -05:00
parent b02cce5678
commit 748e0ab6ce
7 changed files with 423 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
doc/source/command-objects/network.rst
View File

@ -64,16 +64,16 @@ List networks
os network list
[--external]
[--dhcp]
[--dhcp <dhcp-id>]
[--long]
.. option:: --external
List external networks
.. option:: --dhcp
.. option:: --dhcp <dhcp-id>
ID of the DHCP agent
DHCP agent ID
.. option:: --long

59
openstackclient/api/network_v2.py Normal file
View File

@ -0,0 +1,59 @@
# 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.
#
"""Network v2 API Library"""
from openstackclient.api import api
class APIv2(api.BaseAPI):
"""Network v2 API"""
def __init__(self, **kwargs):
super(APIv2, self).__init__(**kwargs)
def dhcp_agent_list(
self,
dhcp_id=None,
**filter
):
"""List DHCP agents
:param string dhcp_id:
DHCP Agent ID
:param filter:
used to create the query string filters
http://docs.openstack.org/api/openstack-network/2.0/content/filtering.html
"""
return self.list('dhcp-networks', **filter)['dhcp-networks']
def network_list(
self,
external=False,
**filter
):
"""List external networks
:param string dhcp_id:
DHCP agent ID
:param bool external:
Return external networks if True
:param filter:
used to create the query string filters
http://docs.openstack.org/api/openstack-network/2.0/content/filtering.html
"""
if external:
filter = {'router:external': True}
return self.list('networks', **filter)['networks']

37
openstackclient/network/client.py
View File

@ -24,21 +24,54 @@ API_NAME = "network"
API_VERSIONS = {
"2": "neutronclient.v2_0.client.Client",
}
# Translate our API version to auth plugin version prefix
API_VERSION_MAP = {
'2.0': 'v2.0',
'2': 'v2.0',
}
NETWORK_API_TYPE = 'network'
NETWORK_API_VERSIONS = {
'2': 'openstackclient.api.network_v2.APIv2',
}
def make_client(instance):
"""Returns an network service client."""
"""Returns an network service client"""
network_client = utils.get_client_class(
API_NAME,
instance._api_version[API_NAME],
API_VERSIONS)
LOG.debug('Instantiating network client: %s', network_client)
return network_client(
endpoint = instance.get_endpoint_for_service_type(
API_NAME,
region_name=instance._region_name,
)
client = network_client(
session=instance.session,
region_name=instance._region_name,
)
network_api = utils.get_client_class(
API_NAME,
instance._api_version[API_NAME],
NETWORK_API_VERSIONS)
LOG.debug('Instantiating network api: %s', network_client)
# v2 is hard-coded until discovery is completed, neutron only has one atm
client.api = network_api(
session=instance.session,
service_type=NETWORK_API_TYPE,
endpoint='/'.join([
endpoint,
API_VERSION_MAP[instance._api_version[API_NAME]],
])
)
return client
def build_option_parser(parser):
"""Hook to add global options"""

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

@ -25,10 +25,21 @@ from openstackclient.common import utils
from openstackclient.network import common
def filters(data):
if 'subnets' in data:
data['subnets'] = utils.format_list(data['subnets'])
return data
def _prep_network_detail(net):
"""Prepare network object for output"""
if 'subnets' in net:
net['subnets'] = utils.format_list(net['subnets'])
if 'admin_state_up' in net:
net['state'] = 'UP' if net['admin_state_up'] else 'DOWN'
net.pop('admin_state_up')
if 'router:external' in net:
net['router_type'] = 'External' if net['router:external'] \
else 'Internal'
net.pop('router:external')
if 'tenant_id' in net:
net['project_id'] = net.pop('tenant_id')
return net
class CreateNetwork(show.ShowOne):
@ -80,7 +91,7 @@ class CreateNetwork(show.ShowOne):
create_method = getattr(client, "create_network")
data = create_method(body)['network']
if data:
data = filters(data)
data = _prep_network_detail(data)
else:
data = {'': ''}
return zip(*sorted(six.iteritems(data)))
@ -133,40 +144,63 @@ class ListNetwork(lister.Lister):
)
parser.add_argument(
'--dhcp',
help='ID of the DHCP agent')
metavar='<dhcp-id>',
help='DHCP agent ID')
parser.add_argument(
'--long',
action='store_true',
default=False,
help='Long listing',
help='List additional fields in output',
)
return parser
def take_action(self, parsed_args):
self.log.debug('take_action(%s)' % parsed_args)
client = self.app.client_manager.network
if parsed_args.dhcp:
list_method = getattr(client, 'list_networks_on_dhcp_agent')
resources = 'networks_on_dhcp_agent'
report_filter = {'dhcp_agent': parsed_args.dhcp}
data = list_method(**report_filter)[resources]
data = client.api.dhcp_agent_list(dhcp_id=parsed_args.dhcp)
columns = ('ID',)
column_headers = columns
else:
list_method = getattr(client, "list_networks")
report_filter = {}
if parsed_args.external:
report_filter = {'router:external': True}
data = list_method(**report_filter)['networks']
columns = len(data) > 0 and sorted(data[0].keys()) or []
if parsed_args.columns:
list_columns = parsed_args.columns
else:
list_columns = ['id', 'name', 'subnets']
if not parsed_args.long and not parsed_args.dhcp:
columns = [x for x in list_columns if x in columns]
formatters = {'subnets': utils.format_list}
return (columns,
(utils.get_dict_properties(s, columns, formatters=formatters)
for s in data))
data = client.api.network_list(external=parsed_args.external)
if parsed_args.long:
columns = (
'ID',
'Name',
'Status',
'project_id',
'state',
'Shared',
'Subnets',
'provider:network_type',
'router_type',
)
column_headers = (
'ID',
'Name',
'Status',
'Project',
'State',
'Shared',
'Subnets',
'Network Type',
'Router Type',
)
else:
columns = ('ID', 'Name', 'Subnets')
column_headers = columns
for d in data:
d = _prep_network_detail(d)
return (column_headers,
(utils.get_dict_properties(
s, columns,
formatters={'subnets': utils.format_list},
) for s in data))
class SetNetwork(command.Command):
@ -253,9 +287,9 @@ class ShowNetwork(show.ShowOne):
def take_action(self, parsed_args):
self.log.debug('take_action(%s)' % parsed_args)
client = self.app.client_manager.network
_id = common.find(client, 'network', 'networks',
parsed_args.identifier)
show_method = getattr(client, "show_network")
data = show_method(_id)['network']
data = filters(data)
net = client.api.find_attr(
'networks',
parsed_args.identifier,
)
data = _prep_network_detail(net)
return zip(*sorted(six.iteritems(data)))

52
openstackclient/tests/api/test_network_v2.py Normal file
View File

@ -0,0 +1,52 @@
# 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.
#
"""Network v2 API Library Tests"""
from requests_mock.contrib import fixture
from keystoneclient import session
from openstackclient.api import network_v2 as network
from openstackclient.tests import utils
FAKE_PROJECT = 'xyzpdq'
FAKE_URL = 'http://gopher.com/v2/' + FAKE_PROJECT
class TestNetworkAPIv2(utils.TestCase):
def setUp(self):
super(TestNetworkAPIv2, self).setUp()
sess = session.Session()
self.api = network.APIv2(session=sess, endpoint=FAKE_URL)
self.requests_mock = self.useFixture(fixture.Fixture())
class TestNetwork(TestNetworkAPIv2):
LIST_NETWORK_RESP = [
{'id': '1', 'name': 'p1', 'description': 'none', 'enabled': True},
{'id': '2', 'name': 'p2', 'description': 'none', 'enabled': False},
{'id': '3', 'name': 'p3', 'description': 'none', 'enabled': True},
]
def test_network_list_no_options(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/networks',
json={'networks': self.LIST_NETWORK_RESP},
status_code=200,
)
ret = self.api.network_list()
self.assertEqual(self.LIST_NETWORK_RESP, ret)

14
openstackclient/tests/network/common.py
View File

@ -14,16 +14,26 @@
import argparse
import mock
from openstackclient.api import network_v2
from openstackclient.tests import utils
class FakeNetworkClient(object):
pass
class TestNetworkBase(utils.TestCommand):
def setUp(self):
super(TestNetworkBase, self).setUp()
self.app = mock.Mock(name='app')
self.app.client_manager = mock.Mock(name='client_manager')
self.namespace = argparse.Namespace()
self.app.client_manager.network = FakeNetworkClient()
self.app.client_manager.network.api = network_v2.APIv2(
session=mock.Mock(),
service_type="network",
)
self.api = self.app.client_manager.network.api
given_show_options = [
'-f',
'shell',

279
openstackclient/tests/network/v2/test_network.py
View File

@ -22,16 +22,38 @@ RESOURCE = 'network'
RESOURCES = 'networks'
FAKE_ID = 'iditty'
FAKE_NAME = 'noo'
FAKE_PROJECT = 'yaa'
RECORD = {
'id': FAKE_ID,
'name': FAKE_NAME,
'admin_state_up': True,
'router:external': True,
'status': 'ACTIVE',
'subnets': ['a', 'b'],
'tenant_id': FAKE_PROJECT,
}
COLUMNS = ['id', 'name', 'subnets']
RESPONSE = {RESOURCE: RECORD}
FILTERED = [('id', 'name', 'router:external', 'subnets'),
(FAKE_ID, FAKE_NAME, True, 'a, b')]
COLUMNS = ['ID', 'Name', 'Subnets']
RESPONSE = {RESOURCE: copy.deepcopy(RECORD)}
FILTERED = [
(
'id',
'name',
'project_id',
'router_type',
'state',
'status',
'subnets',
),
(
FAKE_ID,
FAKE_NAME,
FAKE_PROJECT,
'External',
'UP',
'ACTIVE',
'a, b',
),
]
class TestCreateNetwork(common.TestNetworkBase):
@ -122,7 +144,7 @@ class TestDeleteNetwork(common.TestNetworkBase):
verifylist = [
('networks', [FAKE_NAME]),
]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
lister = mock.Mock(return_value={RESOURCES: [copy.deepcopy(RECORD)]})
self.app.client_manager.network.list_networks = lister
mocker = mock.Mock(return_value=None)
self.app.client_manager.network.delete_network = mocker
@ -135,85 +157,170 @@ class TestDeleteNetwork(common.TestNetworkBase):
self.assertEqual(None, result)
@mock.patch(
'openstackclient.api.network_v2.APIv2.network_list'
)
class TestListNetwork(common.TestNetworkBase):
def test_list_no_options(self):
def setUp(self):
super(TestListNetwork, self).setUp()
# Get the command object to test
self.cmd = network.ListNetwork(self.app, self.namespace)
self.NETWORK_LIST = [
copy.deepcopy(RECORD),
copy.deepcopy(RECORD),
]
def test_network_list_no_options(self, n_mock):
n_mock.return_value = self.NETWORK_LIST
arglist = []
verifylist = [
('long', False),
('dhcp', None),
('external', False),
('dhcp', None),
('long', False),
]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
self.app.client_manager.network.list_networks = lister
cmd = network.ListNetwork(self.app, self.namespace)
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
parsed_args = self.check_parser(cmd, arglist, verifylist)
result = cmd.take_action(parsed_args)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
lister.assert_called_with()
self.assertEqual(COLUMNS, result[0])
self.assertEqual((FAKE_ID, FAKE_NAME, 'a, b'), next(result[1]))
self.assertRaises(StopIteration, next, result[1])
# Set expected values
n_mock.assert_called_with(
external=False,
)
def test_list_long(self):
arglist = ['--long']
self.assertEqual(tuple(COLUMNS), columns)
datalist = [
(FAKE_ID, FAKE_NAME, 'a, b'),
(FAKE_ID, FAKE_NAME, 'a, b'),
]
self.assertEqual(datalist, list(data))
def test_list_external(self, n_mock):
n_mock.return_value = self.NETWORK_LIST
arglist = [
'--external',
]
verifylist = [
('external', True),
('dhcp', None),
('long', False),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
# Set expected values
n_mock.assert_called_with(
external=True,
)
self.assertEqual(tuple(COLUMNS), columns)
datalist = [
(FAKE_ID, FAKE_NAME, 'a, b'),
(FAKE_ID, FAKE_NAME, 'a, b'),
]
self.assertEqual(datalist, list(data))
def test_network_list_long(self, n_mock):
n_mock.return_value = self.NETWORK_LIST
arglist = [
'--long',
]
verifylist = [
('long', True),
('dhcp', None),
('external', False),
]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
self.app.client_manager.network.list_networks = lister
cmd = network.ListNetwork(self.app, self.namespace)
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
parsed_args = self.check_parser(cmd, arglist, verifylist)
result = cmd.take_action(parsed_args)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
lister.assert_called_with()
headings = ['id', 'name', 'router:external', 'subnets']
self.assertEqual(headings, result[0])
data = (FAKE_ID, FAKE_NAME, True, 'a, b')
self.assertEqual(data, next(result[1]))
self.assertRaises(StopIteration, next, result[1])
# Set expected values
n_mock.assert_called_with(
external=False,
)
collist = (
'ID',
'Name',
'Status',
'Project',
'State',
'Shared',
'Subnets',
'Network Type',
'Router Type',
)
self.assertEqual(columns, collist)
dataitem = (
FAKE_ID,
FAKE_NAME,
'ACTIVE',
FAKE_PROJECT,
'UP',
'',
'a, b',
'',
'External',
)
datalist = [
dataitem,
dataitem,
]
self.assertEqual(list(data), datalist)
@mock.patch(
'openstackclient.api.network_v2.APIv2.dhcp_agent_list'
)
class TestListDhcpAgent(common.TestNetworkBase):
def setUp(self):
super(TestListDhcpAgent, self).setUp()
# Get the command object to test
self.cmd = network.ListNetwork(self.app, self.namespace)
self.DHCP_LIST = [
{'id': '1'},
{'id': '2'},
]
def test_list_dhcp(self, n_mock):
n_mock.return_value = self.DHCP_LIST
def test_list_dhcp(self):
arglist = [
'--dhcp',
'dhcpid',
] + self.given_list_options
'--dhcp', 'dhcpid',
]
verifylist = [
('external', False),
('dhcp', 'dhcpid'),
] + self.then_list_options
fake_dhcp_data = [{'id': '1'}, {'id': '2'}]
fake_dhcp_response = {'networks_on_dhcp_agent': fake_dhcp_data}
lister = mock.Mock(return_value=fake_dhcp_response)
netty = self.app.client_manager.network
netty.list_networks_on_dhcp_agent = lister
cmd = network.ListNetwork(self.app, self.namespace)
('long', False),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
parsed_args = self.check_parser(cmd, arglist, verifylist)
result = cmd.take_action(parsed_args)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
lister.assert_called_with(dhcp_agent='dhcpid')
self.assertEqual(['id'], result[0])
self.assertEqual(('1',), next(result[1]))
self.assertEqual(('2',), next(result[1]))
self.assertRaises(StopIteration, next, result[1])
# Set expected values
n_mock.assert_called_with(
dhcp_id='dhcpid',
)
def test_list_external(self):
arglist = ['--external', '-c', 'id']
verifylist = [('external', True)]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
self.app.client_manager.network.list_networks = lister
cmd = network.ListNetwork(self.app, self.namespace)
parsed_args = self.check_parser(cmd, arglist, verifylist)
result = cmd.take_action(parsed_args)
lister.assert_called_with(**{'router:external': True})
self.assertEqual(['id'], result[0])
self.assertEqual((FAKE_ID,), next(result[1]))
self.assertRaises(StopIteration, next, result[1])
self.assertEqual(('ID',), columns)
datalist = [
('1',),
('2',),
]
self.assertEqual(datalist, list(data))
class TestSetNetwork(common.TestNetworkBase):
@ -230,7 +337,7 @@ class TestSetNetwork(common.TestNetworkBase):
('name', 'noob'),
('shared', True),
]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
lister = mock.Mock(return_value={RESOURCES: [copy.deepcopy(RECORD)]})
self.app.client_manager.network.list_networks = lister
mocker = mock.Mock(return_value=None)
self.app.client_manager.network.update_network = mocker
@ -255,7 +362,7 @@ class TestSetNetwork(common.TestNetworkBase):
('admin_state', False),
('shared', False),
]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
lister = mock.Mock(return_value={RESOURCES: [copy.deepcopy(RECORD)]})
self.app.client_manager.network.list_networks = lister
mocker = mock.Mock(return_value=None)
self.app.client_manager.network.update_network = mocker
@ -272,7 +379,7 @@ class TestSetNetwork(common.TestNetworkBase):
def test_set_nothing(self):
arglist = [FAKE_NAME, ]
verifylist = [('identifier', FAKE_NAME), ]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
lister = mock.Mock(return_value={RESOURCES: [copy.deepcopy(RECORD)]})
self.app.client_manager.network.list_networks = lister
mocker = mock.Mock(return_value=None)
self.app.client_manager.network.update_network = mocker
@ -283,37 +390,43 @@ class TestSetNetwork(common.TestNetworkBase):
parsed_args)
@mock.patch(
'openstackclient.api.network_v2.APIv2.find_attr'
)
class TestShowNetwork(common.TestNetworkBase):
def test_show_no_options(self):
def setUp(self):
super(TestShowNetwork, self).setUp()
# Get the command object to test
self.cmd = network.ShowNetwork(self.app, self.namespace)
self.NETWORK_ITEM = copy.deepcopy(RECORD)
def test_show_no_options(self, n_mock):
arglist = [
FAKE_NAME,
]
verifylist = [
('identifier', FAKE_NAME),
]
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
self.app.client_manager.network.list_networks = lister
mocker = mock.Mock(return_value=copy.deepcopy(RESPONSE))
self.app.client_manager.network.show_network = mocker
cmd = network.ShowNetwork(self.app, self.namespace)
n_mock.return_value = copy.deepcopy(RECORD)
self.cmd = network.ShowNetwork(self.app, self.namespace)
parsed_args = self.check_parser(cmd, arglist, verifylist)
result = list(cmd.take_action(parsed_args))
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = list(self.cmd.take_action(parsed_args))
mocker.assert_called_with(FAKE_ID)
n_mock.assert_called_with('networks', FAKE_NAME)
self.assertEqual(FILTERED, result)
def test_show_all_options(self):
def test_show_all_options(self, n_mock):
arglist = [FAKE_NAME] + self.given_show_options
verifylist = [('identifier', FAKE_NAME)] + self.then_show_options
lister = mock.Mock(return_value={RESOURCES: [RECORD]})
self.app.client_manager.network.list_networks = lister
mocker = mock.Mock(return_value=copy.deepcopy(RESPONSE))
self.app.client_manager.network.show_network = mocker
cmd = network.ShowNetwork(self.app, self.namespace)
n_mock.return_value = copy.deepcopy(RECORD)
self.cmd = network.ShowNetwork(self.app, self.namespace)
parsed_args = self.check_parser(cmd, arglist, verifylist)
result = list(cmd.take_action(parsed_args))
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = list(self.cmd.take_action(parsed_args))
mocker.assert_called_with(FAKE_ID)
n_mock.assert_called_with('networks', FAKE_NAME)
self.assertEqual(FILTERED, result)