# Copyright 2012 OpenStack Foundation
# All Rights Reserved.
#
# 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.
import testtools
from tempest.api.compute import base
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
CONF = config.CONF
[docs]
class ServersTestJSON(base.BaseV2ComputeTest):
"""Test servers API"""
create_default_network = True
credentials = ['primary', 'project_reader']
@classmethod
def setup_clients(cls):
super(ServersTestJSON, cls).setup_clients()
cls.client = cls.servers_client
if CONF.enforce_scope.nova:
cls.reader_client = cls.os_project_reader.servers_client
else:
cls.reader_client = cls.client
[docs]
@decorators.idempotent_id('b92d5ec7-b1dd-44a2-87e4-45e888c46ef0')
@testtools.skipUnless(CONF.compute_feature_enabled.
enable_instance_password,
'Instance password not available.')
def test_create_server_with_admin_password(self):
"""Test creating server with admin password
If an admin password is provided on server creation, the server's
root password should be set to that password.
"""
server = self.create_test_server(adminPass='testpassword')
self.addCleanup(self.delete_server, server['id'])
# Verify the password is set correctly in the response
self.assertEqual('testpassword', server['adminPass'])
[docs]
@decorators.idempotent_id('8fea6be7-065e-47cf-89b8-496e6f96c699')
def test_create_with_existing_server_name(self):
"""Test creating a server with already existing name is allowed"""
# TODO(sdague): clear out try, we do cleanup one layer up
server_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix,
name=self.__class__.__name__ + '-server')
server = self.create_test_server(name=server_name,
wait_until='ACTIVE')
id1 = server['id']
self.addCleanup(self.delete_server, id1)
server = self.create_test_server(name=server_name,
wait_until='ACTIVE')
id2 = server['id']
self.addCleanup(self.delete_server, id2)
self.assertNotEqual(id1, id2, "Did not create a new server")
server = self.reader_client.show_server(id1)['server']
name1 = server['name']
server = self.reader_client.show_server(id2)['server']
name2 = server['name']
self.assertEqual(name1, name2)
[docs]
@decorators.idempotent_id('f9e15296-d7f9-4e62-b53f-a04e89160833')
def test_create_specify_keypair(self):
"""Test creating server with keypair"""
key_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name='key')
self.keypairs_client.create_keypair(name=key_name)
self.addCleanup(self.keypairs_client.delete_keypair, key_name)
self.keypairs_client.list_keypairs()
server = self.create_test_server(key_name=key_name,
wait_until='ACTIVE')
self.addCleanup(self.delete_server, server['id'])
server = self.reader_client.show_server(server['id'])['server']
self.assertEqual(key_name, server['key_name'])
def _update_server_name(self, server_id, status, prefix_name='server'):
# The server name should be changed to the provided value
new_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix, name=prefix_name)
# Update the server with a new name
self.client.update_server(server_id,
name=new_name)
waiters.wait_for_server_status(self.client, server_id, status)
# Verify the name of the server has changed
server = self.reader_client.show_server(server_id)['server']
self.assertEqual(new_name, server['name'])
return server
[docs]
@decorators.idempotent_id('5e6ccff8-349d-4852-a8b3-055df7988dd2')
def test_update_server_name(self):
"""Test updating server name to the provided value"""
server = self.create_test_server(wait_until='ACTIVE')
self.addCleanup(self.delete_server, server['id'])
# Update instance name with non-ASCII characters
prefix_name = '\u00CD\u00F1st\u00E1\u00F1c\u00E9'
self._update_server_name(server['id'], 'ACTIVE', prefix_name)
# stop server and check server name update again
self.client.stop_server(server['id'])
waiters.wait_for_server_status(self.client, server['id'], 'SHUTOFF')
# Update instance name with non-ASCII characters
updated_server = self._update_server_name(server['id'],
'SHUTOFF',
prefix_name)
self.assertNotIn('progress', updated_server)
[docs]
@decorators.idempotent_id('89b90870-bc13-4b73-96af-f9d4f2b70077')
def test_update_access_server_address(self):
"""Test updating server's access addresses to the provided value"""
server = self.create_test_server(wait_until='ACTIVE')
self.addCleanup(self.delete_server, server['id'])
# Update the IPv4 and IPv6 access addresses
self.client.update_server(server['id'],
accessIPv4='1.1.1.1',
accessIPv6='::babe:202:202')
waiters.wait_for_server_status(self.client, server['id'], 'ACTIVE')
# Verify the access addresses have been updated
server = self.reader_client.show_server(server['id'])['server']
self.assertEqual('1.1.1.1', server['accessIPv4'])
self.assertEqual('::babe:202:202', server['accessIPv6'])
[docs]
@decorators.idempotent_id('38fb1d02-c3c5-41de-91d3-9bc2025a75eb')
def test_create_server_with_ipv6_addr_only(self):
"""Test creating server with ipv6 address only(no ipv4 address)"""
server = self.create_test_server(accessIPv6='2001:2001::3',
wait_until='ACTIVE')
self.addCleanup(self.delete_server, server['id'])
server = self.reader_client.show_server(server['id'])['server']
self.assertEqual('2001:2001::3', server['accessIPv6'])
[docs]
@decorators.related_bug('1730756')
@decorators.idempotent_id('defbaca5-d611-49f5-ae21-56ee25d2db49')
def test_create_server_specify_multibyte_character_name(self):
"""Test creating server with multi character name
prefix character is:
http://unicode.org/cldr/utility/character.jsp?a=20A1
We use a string with 3 byte utf-8 character due to nova
will return 400(Bad Request) if we attempt to send a name which has
4 byte utf-8 character.
"""
utf8_name = data_utils.rand_name(
prefix=CONF.resource_name_prefix,
name=b'\xe2\x82\xa1'.decode('utf-8'))
self.create_test_server(name=utf8_name, wait_until='ACTIVE')
[docs]
class ServerShowV247Test(base.BaseV2ComputeTest):
"""Test servers API with compute microversion greater than 2.46"""
min_microversion = '2.47'
max_microversion = 'latest'
# NOTE(gmann): This test tests the server APIs response schema
# Along with 2.47 microversion schema this test class tests the
# other microversions 2.9, 2.19 and 2.26 server APIs response schema
# also. 2.47 APIs schema are on top of 2.9->2.19->2.26 schema so
# below tests cover all of the schema.
credentials = ['primary', 'project_reader']
@classmethod
def setup_clients(cls):
super(ServerShowV247Test, cls).setup_clients()
if CONF.enforce_scope.nova:
cls.reader_client = cls.os_project_reader.servers_client
else:
cls.reader_client = cls.servers_client
[docs]
@decorators.idempotent_id('88b0bdb2-494c-11e7-a919-92ebcb67fe33')
def test_show_server(self):
"""Test getting server detail"""
server = self.create_test_server()
# All fields will be checked by API schema
self.reader_client.show_server(server['id'])
[docs]
@decorators.idempotent_id('8de397c2-57d0-4b90-aa30-e5d668f21a8b')
def test_update_rebuild_list_server(self):
"""Test update/rebuild/list server"""
server = self.create_test_server()
# Checking update API response schema
self.servers_client.update_server(server['id'])
waiters.wait_for_server_status(self.servers_client, server['id'],
'ACTIVE')
# Checking rebuild API response schema
self.servers_client.rebuild_server(server['id'], self.image_ref_alt)
waiters.wait_for_server_status(self.servers_client,
server['id'], 'ACTIVE')
# Checking list details API response schema
self.servers_client.list_servers(detail=True)
[docs]
class ServerShowV263Test(base.BaseV2ComputeTest):
"""Test servers API with compute microversion greater than 2.62"""
min_microversion = '2.63'
max_microversion = 'latest'
credentials = ['primary', 'project_reader']
@classmethod
def setup_clients(cls):
super(ServerShowV263Test, cls).setup_clients()
if CONF.enforce_scope.nova:
cls.reader_client = cls.os_project_reader.servers_client
else:
cls.reader_client = cls.servers_client
[docs]
@testtools.skipUnless(CONF.compute.certified_image_ref,
'``[compute]/certified_image_ref`` required to test '
'image certificate validation.')
@testtools.skipUnless(CONF.compute.certified_image_trusted_certs,
'``[compute]/certified_image_trusted_certs`` '
'required to test image certificate validation.')
@decorators.idempotent_id('71b8e3d5-11d2-494f-b917-b094a4afed3c')
def test_show_update_rebuild_list_server(self):
"""Test show/update/rebuild/list server"""
trusted_certs = CONF.compute.certified_image_trusted_certs
server = self.create_test_server(
image_id=CONF.compute.certified_image_ref,
trusted_image_certificates=trusted_certs,
wait_until='ACTIVE')
# Check show API response schema
self.reader_client.show_server(server['id'])['server']
# Check update API response schema
self.servers_client.update_server(server['id'])
waiters.wait_for_server_status(self.servers_client,
server['id'], 'ACTIVE')
# Check rebuild API response schema
self.servers_client.rebuild_server(
server['id'], CONF.compute.certified_image_ref)
waiters.wait_for_server_status(self.servers_client,
server['id'], 'ACTIVE')
# Check list details API response schema
params = {'trusted_image_certificates': trusted_certs}
servers = self.servers_client.list_servers(
detail=True, **params)['servers']
self.assertNotEmpty(servers)
[docs]
class ServersListShow296Test(base.BaseV2ComputeTest):
"""Test compute server with microversion >= than 2.96
This test tests the Server APIs response schema for 2.96 microversion.
No specific assert or behaviour verification is needed.
"""
min_microversion = '2.96'
max_microversion = 'latest'
[docs]
@decorators.idempotent_id('4eee1ffe-9e00-4c99-a431-0d3e0f323a8f')
def test_list_show_server_296(self):
server = self.create_test_server()
# Checking list API response schema.
self.servers_client.list_servers(detail=True)
# Checking show API response schema
self.servers_client.show_server(server['id'])
