71322c7bbc
This is a big patch because there are more than one issue happening at the same time and we have to fix all of them to fix any of them. Force nova microversion to 2.0 The current use of novaclient is to get the latest microversion. So far this has not been a problem, as shade deals with different payloads across clouds all the time. However, the latest microversion to nova broke shade's expectations about how usage reports work. Actual microversion support is coming soon to shade, but is too much of a task for a gate fix. In the meantime, pin to 2.0 which is available on all of the clouds. Produce some debug details about nova usage objects Capture novaclient debug logging In chasing down the usage issue, we were missing the REST interactions we needed to be effective in chasing down the problem. novaclient passes its own logger to keystoneauth Session, so we needed to include it in the debug logging setup. Also, add a helper function to make adding things like this easier. Consume cirros qcow2 image if it's there The move from ami to qcow2 for cirros broke shade's finding of it as a candidate image. Move pick_image into the base class so that we can include add_on_exception and error messages everywhere consistently. Add image list to debug output on failure. When we can't find a sensible image, add the list of images to the test output so that we can examine them. Change-Id: Ifae65e6cdf48921eaa379b803913277affbfe22a
293 lines
11 KiB
Python
293 lines
11 KiB
Python
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
|
|
#
|
|
# 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.
|
|
|
|
"""
|
|
test_floating_ip
|
|
----------------------------------
|
|
|
|
Functional tests for floating IP resource.
|
|
"""
|
|
|
|
import pprint
|
|
|
|
from novaclient import exceptions as nova_exc
|
|
from testtools import content
|
|
|
|
from shade import _utils
|
|
from shade import meta
|
|
from shade.exc import OpenStackCloudException
|
|
from shade.tests.functional import base
|
|
from shade.tests.functional.util import pick_flavor
|
|
|
|
|
|
class TestFloatingIP(base.BaseFunctionalTestCase):
|
|
timeout = 60
|
|
|
|
def setUp(self):
|
|
super(TestFloatingIP, self).setUp()
|
|
self.nova = self.user_cloud.nova_client
|
|
if self.user_cloud.has_service('network'):
|
|
self.neutron = self.user_cloud.neutron_client
|
|
self.flavor = pick_flavor(self.nova.flavors.list())
|
|
if self.flavor is None:
|
|
self.assertFalse('no sensible flavor available')
|
|
self.image = self.pick_image()
|
|
|
|
# Generate a random name for these tests
|
|
self.new_item_name = self.getUniqueString()
|
|
|
|
self.addCleanup(self._cleanup_network)
|
|
self.addCleanup(self._cleanup_servers)
|
|
|
|
def _cleanup_network(self):
|
|
exception_list = list()
|
|
|
|
# Delete stale networks as well as networks created for this test
|
|
if self.user_cloud.has_service('network'):
|
|
# Delete routers
|
|
for r in self.user_cloud.list_routers():
|
|
try:
|
|
if r['name'].startswith(self.new_item_name):
|
|
# ToDo: update_router currently won't allow removing
|
|
# external_gateway_info
|
|
router = {
|
|
'external_gateway_info': None
|
|
}
|
|
self.neutron.update_router(
|
|
router=r['id'], body={'router': router})
|
|
# ToDo: Shade currently doesn't have methods for this
|
|
for s in self.user_cloud.list_subnets():
|
|
if s['name'].startswith(self.new_item_name):
|
|
try:
|
|
self.neutron.remove_interface_router(
|
|
router=r['id'],
|
|
body={'subnet_id': s['id']})
|
|
except Exception:
|
|
pass
|
|
self.user_cloud.delete_router(name_or_id=r['id'])
|
|
except Exception as e:
|
|
exception_list.append(str(e))
|
|
continue
|
|
# Delete subnets
|
|
for s in self.user_cloud.list_subnets():
|
|
if s['name'].startswith(self.new_item_name):
|
|
try:
|
|
self.user_cloud.delete_subnet(name_or_id=s['id'])
|
|
except Exception as e:
|
|
exception_list.append(str(e))
|
|
continue
|
|
# Delete networks
|
|
for n in self.user_cloud.list_networks():
|
|
if n['name'].startswith(self.new_item_name):
|
|
try:
|
|
self.user_cloud.delete_network(name_or_id=n['id'])
|
|
except Exception as e:
|
|
exception_list.append(str(e))
|
|
continue
|
|
|
|
if exception_list:
|
|
# Raise an error: we must make users aware that something went
|
|
# wrong
|
|
raise OpenStackCloudException('\n'.join(exception_list))
|
|
|
|
def _cleanup_servers(self):
|
|
exception_list = list()
|
|
|
|
# Delete stale servers as well as server created for this test
|
|
for i in self.nova.servers.list():
|
|
if i.name.startswith(self.new_item_name):
|
|
self.nova.servers.delete(i)
|
|
for _ in _utils._iterate_timeout(
|
|
self.timeout, "Timeout deleting servers"):
|
|
try:
|
|
self.nova.servers.get(server=i)
|
|
except nova_exc.NotFound:
|
|
break
|
|
except Exception as e:
|
|
exception_list.append(str(e))
|
|
continue
|
|
|
|
if exception_list:
|
|
# Raise an error: we must make users aware that something went
|
|
# wrong
|
|
raise OpenStackCloudException('\n'.join(exception_list))
|
|
|
|
def _cleanup_ips(self, server):
|
|
|
|
exception_list = list()
|
|
|
|
fixed_ip = meta.get_server_private_ip(server)
|
|
|
|
for ip in self.user_cloud.list_floating_ips():
|
|
if (ip.get('fixed_ip', None) == fixed_ip
|
|
or ip.get('fixed_ip_address', None) == fixed_ip):
|
|
try:
|
|
self.user_cloud.delete_floating_ip(ip['id'])
|
|
except Exception as e:
|
|
exception_list.append(str(e))
|
|
continue
|
|
|
|
if exception_list:
|
|
# Raise an error: we must make users aware that something went
|
|
# wrong
|
|
raise OpenStackCloudException('\n'.join(exception_list))
|
|
|
|
def _setup_networks(self):
|
|
if self.user_cloud.has_service('network'):
|
|
# Create a network
|
|
self.test_net = self.user_cloud.create_network(
|
|
name=self.new_item_name + '_net')
|
|
# Create a subnet on it
|
|
self.test_subnet = self.user_cloud.create_subnet(
|
|
subnet_name=self.new_item_name + '_subnet',
|
|
network_name_or_id=self.test_net['id'],
|
|
cidr='10.24.4.0/24',
|
|
enable_dhcp=True
|
|
)
|
|
# Create a router
|
|
self.test_router = self.user_cloud.create_router(
|
|
name=self.new_item_name + '_router')
|
|
# Attach the router to an external network
|
|
ext_nets = self.user_cloud.search_networks(
|
|
filters={'router:external': True})
|
|
self.user_cloud.update_router(
|
|
name_or_id=self.test_router['id'],
|
|
ext_gateway_net_id=ext_nets[0]['id'])
|
|
# Attach the router to the internal subnet
|
|
self.neutron.add_interface_router(
|
|
router=self.test_router['id'],
|
|
body={'subnet_id': self.test_subnet['id']})
|
|
|
|
# Select the network for creating new servers
|
|
self.nic = {'net-id': self.test_net['id']}
|
|
self.addDetail(
|
|
'networks-neutron',
|
|
content.text_content(pprint.pformat(
|
|
self.user_cloud.list_networks())))
|
|
else:
|
|
# ToDo: remove once we have list/get methods for nova networks
|
|
nets = self.user_cloud.nova_client.networks.list()
|
|
self.addDetail(
|
|
'networks-nova',
|
|
content.text_content(pprint.pformat(
|
|
nets)))
|
|
self.nic = {'net-id': nets[0].id}
|
|
|
|
def test_private_ip(self):
|
|
self._setup_networks()
|
|
|
|
new_server = self.user_cloud.get_openstack_vars(
|
|
self.user_cloud.create_server(
|
|
wait=True, name=self.new_item_name + '_server',
|
|
image=self.image,
|
|
flavor=self.flavor, nics=[self.nic]))
|
|
|
|
self.addDetail(
|
|
'server', content.text_content(pprint.pformat(new_server)))
|
|
self.assertNotEqual(new_server['private_v4'], '')
|
|
|
|
def test_add_auto_ip(self):
|
|
self._setup_networks()
|
|
|
|
new_server = self.user_cloud.create_server(
|
|
wait=True, name=self.new_item_name + '_server',
|
|
image=self.image,
|
|
flavor=self.flavor, nics=[self.nic])
|
|
|
|
# ToDo: remove the following iteration when create_server waits for
|
|
# the IP to be attached
|
|
ip = None
|
|
for _ in _utils._iterate_timeout(
|
|
self.timeout, "Timeout waiting for IP address to be attached"):
|
|
ip = meta.get_server_external_ipv4(self.user_cloud, new_server)
|
|
if ip is not None:
|
|
break
|
|
new_server = self.user_cloud.get_server(new_server.id)
|
|
|
|
self.addCleanup(self._cleanup_ips, new_server)
|
|
|
|
def test_detach_ip_from_server(self):
|
|
self._setup_networks()
|
|
|
|
new_server = self.user_cloud.create_server(
|
|
wait=True, name=self.new_item_name + '_server',
|
|
image=self.image,
|
|
flavor=self.flavor, nics=[self.nic])
|
|
|
|
# ToDo: remove the following iteration when create_server waits for
|
|
# the IP to be attached
|
|
ip = None
|
|
for _ in _utils._iterate_timeout(
|
|
self.timeout, "Timeout waiting for IP address to be attached"):
|
|
ip = meta.get_server_external_ipv4(self.user_cloud, new_server)
|
|
if ip is not None:
|
|
break
|
|
new_server = self.user_cloud.get_server(new_server.id)
|
|
|
|
self.addCleanup(self._cleanup_ips, new_server)
|
|
|
|
f_ip = self.user_cloud.get_floating_ip(
|
|
id=None, filters={'floating_ip_address': ip})
|
|
self.user_cloud.detach_ip_from_server(
|
|
server_id=new_server.id, floating_ip_id=f_ip['id'])
|
|
|
|
def test_list_floating_ips(self):
|
|
fip_admin = self.operator_cloud.create_floating_ip()
|
|
self.addCleanup(self.operator_cloud.delete_floating_ip, fip_admin.id)
|
|
fip_user = self.user_cloud.create_floating_ip()
|
|
self.addCleanup(self.user_cloud.delete_floating_ip, fip_user.id)
|
|
|
|
# Get all the floating ips.
|
|
fip_id_list = [
|
|
fip.id for fip in self.operator_cloud.list_floating_ips()
|
|
]
|
|
if self.user_cloud.has_service('network'):
|
|
# Neutron returns all FIP for all projects by default
|
|
self.assertIn(fip_admin.id, fip_id_list)
|
|
self.assertIn(fip_user.id, fip_id_list)
|
|
|
|
# Ask Neutron for only a subset of all the FIPs.
|
|
filtered_fip_id_list = [
|
|
fip.id for fip in self.operator_cloud.list_floating_ips(
|
|
{'tenant_id': self.user_cloud.current_project_id}
|
|
)
|
|
]
|
|
self.assertNotIn(fip_admin.id, filtered_fip_id_list)
|
|
self.assertIn(fip_user.id, filtered_fip_id_list)
|
|
|
|
else:
|
|
self.assertIn(fip_admin.id, fip_id_list)
|
|
# By default, Nova returns only the FIPs that belong to the
|
|
# project which made the listing request.
|
|
self.assertNotIn(fip_user.id, fip_id_list)
|
|
self.assertRaisesRegex(
|
|
ValueError, "Nova-network don't support server-side.*",
|
|
self.operator_cloud.list_floating_ips, filters={'foo': 'bar'}
|
|
)
|
|
|
|
def test_search_floating_ips(self):
|
|
fip_user = self.user_cloud.create_floating_ip()
|
|
self.addCleanup(self.user_cloud.delete_floating_ip, fip_user.id)
|
|
|
|
self.assertIn(
|
|
fip_user['id'],
|
|
[fip.id for fip in self.user_cloud.search_floating_ips(
|
|
filters={"attached": False})]
|
|
)
|
|
self.assertNotIn(
|
|
fip_user['id'],
|
|
[fip.id for fip in self.user_cloud.search_floating_ips(
|
|
filters={"attached": True})]
|
|
)
|