Merge "Cannot boot vm if quantum plugin does not support L3 api"
This commit is contained in:
@@ -720,7 +720,15 @@ class API(base.Base):
|
|||||||
|
|
||||||
def _get_floating_ips_by_fixed_and_port(self, client, fixed_ip, port):
|
def _get_floating_ips_by_fixed_and_port(self, client, fixed_ip, port):
|
||||||
"""Get floatingips from fixed ip and port."""
|
"""Get floatingips from fixed ip and port."""
|
||||||
data = client.list_floatingips(fixed_ip_address=fixed_ip, port_id=port)
|
try:
|
||||||
|
data = client.list_floatingips(fixed_ip_address=fixed_ip,
|
||||||
|
port_id=port)
|
||||||
|
# If a quantum plugin does not implement the L3 API a 404 from
|
||||||
|
# list_floatingips will be raised.
|
||||||
|
except quantumv2.exceptions.QuantumClientException as e:
|
||||||
|
if e.status_code == 404:
|
||||||
|
return []
|
||||||
|
raise
|
||||||
return data['floatingips']
|
return data['floatingips']
|
||||||
|
|
||||||
def release_floating_ip(self, context, address,
|
def release_floating_ip(self, context, address,
|
||||||
|
@@ -1210,6 +1210,18 @@ class TestQuantumv2(test.TestCase):
|
|||||||
self.mox.ReplayAll()
|
self.mox.ReplayAll()
|
||||||
api.remove_fixed_ip_from_instance(self.context, self.instance, address)
|
api.remove_fixed_ip_from_instance(self.context, self.instance, address)
|
||||||
|
|
||||||
|
def test_list_floating_ips_without_l3_support(self):
|
||||||
|
api = quantumapi.API()
|
||||||
|
QuantumNotFound = quantumv2.exceptions.QuantumClientException(
|
||||||
|
status_code=404)
|
||||||
|
self.moxed_client.list_floatingips(
|
||||||
|
fixed_ip_address='1.1.1.1', port_id=1).AndRaise(QuantumNotFound)
|
||||||
|
self.mox.ReplayAll()
|
||||||
|
quantumv2.get_client('fake')
|
||||||
|
floatingips = api._get_floating_ips_by_fixed_and_port(
|
||||||
|
self.moxed_client, '1.1.1.1', 1)
|
||||||
|
self.assertEqual(floatingips, [])
|
||||||
|
|
||||||
|
|
||||||
class TestQuantumv2ModuleMethods(test.TestCase):
|
class TestQuantumv2ModuleMethods(test.TestCase):
|
||||||
def test_ensure_requested_network_ordering_no_preference_ids(self):
|
def test_ensure_requested_network_ordering_no_preference_ids(self):
|
||||||
|
Reference in New Issue
Block a user