openstack/shade
Code Issues Proposed changes

Merge "Add searching for Neutron API extensions"

Browse Source
This commit is contained in:
Jenkins 2017-07-05 15:09:22 +00:00 committed by Gerrit Code Review
commit a4036394e8
2 changed files with 112 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
shade/openstackcloud.py
View File

@ -1446,6 +1446,20 @@ class OpenStackCloud(
keypairs = self.list_keypairs()
return _utils._filter_list(keypairs, name_or_id, filters)
@_utils.cache_on_arguments()
def _neutron_extensions(self):
extensions = set()
for extension in self._network_client.get(
'/extensions.json',
error_message="Error fetching extension list for neutron"):
extensions.add(extension['alias'])
return extensions
def _has_neutron_extension(self, extension_alias):
return extension_alias in self._neutron_extensions()
def search_networks(self, name_or_id=None, filters=None):
"""Search networks

98
shade/tests/unit/test_shade.py
View File

@ -304,6 +304,104 @@ class TestShade(base.RequestsMockTestCase):
self.assert_calls()
def test__neutron_extensions(self):
body = [
{
"updated": "2014-06-1T10:00:00-00:00",
"name": "Distributed Virtual Router",
"links": [],
"alias": "dvr",
"description":
"Enables configuration of Distributed Virtual Routers."
},
{
"updated": "2013-07-23T10:00:00-00:00",
"name": "Allowed Address Pairs",
"links": [],
"alias": "allowed-address-pairs",
"description": "Provides allowed address pairs"
},
]
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'network', 'public', append=['v2.0', 'extensions.json']),
json=dict(extensions=body))
])
extensions = self.cloud._neutron_extensions()
self.assertEqual(set(['dvr', 'allowed-address-pairs']), extensions)
self.assert_calls()
def test__neutron_extensions_fails(self):
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'network', 'public', append=['v2.0', 'extensions.json']),
status_code=404)
])
with testtools.ExpectedException(
exc.OpenStackCloudURINotFound,
"Error fetching extension list for neutron"
):
self.cloud._neutron_extensions()
self.assert_calls()
def test__has_neutron_extension(self):
body = [
{
"updated": "2014-06-1T10:00:00-00:00",
"name": "Distributed Virtual Router",
"links": [],
"alias": "dvr",
"description":
"Enables configuration of Distributed Virtual Routers."
},
{
"updated": "2013-07-23T10:00:00-00:00",
"name": "Allowed Address Pairs",
"links": [],
"alias": "allowed-address-pairs",
"description": "Provides allowed address pairs"
},
]
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'network', 'public', append=['v2.0', 'extensions.json']),
json=dict(extensions=body))
])
self.assertTrue(self.cloud._has_neutron_extension('dvr'))
self.assert_calls()
def test__has_neutron_extension_missing(self):
body = [
{
"updated": "2014-06-1T10:00:00-00:00",
"name": "Distributed Virtual Router",
"links": [],
"alias": "dvr",
"description":
"Enables configuration of Distributed Virtual Routers."
},
{
"updated": "2013-07-23T10:00:00-00:00",
"name": "Allowed Address Pairs",
"links": [],
"alias": "allowed-address-pairs",
"description": "Provides allowed address pairs"
},
]
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'network', 'public', append=['v2.0', 'extensions.json']),
json=dict(extensions=body))
])
self.assertFalse(self.cloud._has_neutron_extension('invalid'))
self.assert_calls()
def test_range_search(self):
filters = {"key1": "min", "key2": "20"}
retval = self.cloud.range_search(RANGE_DATA, filters)