# Copyright 2013 IBM Corp. # # 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. from tempest.api.compute import base from tempest.common.utils import data_utils from tempest import test class ServersAdminV3Test(base.BaseV3ComputeAdminTest): """ Tests Servers API using admin privileges """ _host_key = 'os-extended-server-attributes:host' @classmethod def resource_setup(cls): super(ServersAdminV3Test, cls).resource_setup() cls.client = cls.servers_admin_client cls.non_admin_client = cls.servers_client cls.flavors_client = cls.flavors_admin_client cls.s1_name = data_utils.rand_name('server') resp, server = cls.create_test_server(name=cls.s1_name, wait_until='ACTIVE') cls.s1_id = server['id'] cls.s2_name = data_utils.rand_name('server') resp, server = cls.create_test_server(name=cls.s2_name, wait_until='ACTIVE') cls.s2_id = server['id'] @test.attr(type='gate') def test_list_servers_by_admin(self): # Listing servers by admin user returns empty list by default resp, body = self.client.list_servers_with_detail() servers = body['servers'] self.assertEqual('200', resp['status']) self.assertEqual([], servers) @test.attr(type='gate') def test_list_servers_by_admin_with_all_tenants(self): # Listing servers by admin user with all tenants parameter # Here should be listed all servers params = {'all_tenants': ''} resp, body = self.client.list_servers_with_detail(params) servers = body['servers'] servers_name = map(lambda x: x['name'], servers) self.assertIn(self.s1_name, servers_name) self.assertIn(self.s2_name, servers_name) @test.attr(type='gate') def test_list_servers_filter_by_existent_host(self): # Filter the list of servers by existent host name = data_utils.rand_name('server') flavor = self.flavor_ref image_id = self.image_ref resp, test_server = self.client.create_server( name, image_id, flavor) self.assertEqual('202', resp['status']) self.addCleanup(self.client.delete_server, test_server['id']) self.client.wait_for_server_status(test_server['id'], 'ACTIVE') resp, server = self.client.get_server(test_server['id']) self.assertEqual(server['status'], 'ACTIVE') hostname = server[self._host_key] params = {'host': hostname} resp, body = self.client.list_servers(params) self.assertEqual('200', resp['status']) servers = body['servers'] nonexistent_params = {'host': 'nonexistent_host'} resp, nonexistent_body = self.client.list_servers( nonexistent_params) self.assertEqual('200', resp['status']) nonexistent_servers = nonexistent_body['servers'] self.assertIn(test_server['id'], map(lambda x: x['id'], servers)) self.assertNotIn(test_server['id'], map(lambda x: x['id'], nonexistent_servers)) @test.attr(type='gate') def test_reset_state_server(self): # Reset server's state to 'error' resp, server = self.client.reset_state(self.s1_id) self.assertEqual(202, resp.status) # Verify server's state resp, server = self.client.get_server(self.s1_id) self.assertEqual(server['status'], 'ERROR') # Reset server's state to 'active' resp, server = self.client.reset_state(self.s1_id, state='active') self.assertEqual(202, resp.status) # Verify server's state resp, server = self.client.get_server(self.s1_id) self.assertEqual(server['status'], 'ACTIVE') @test.attr(type='gate') @test.skip_because(bug="1240043") def test_get_server_diagnostics_by_admin(self): # Retrieve server diagnostics by admin user resp, diagnostic = self.client.get_server_diagnostics(self.s1_id) self.assertEqual(200, resp.status) basic_attrs = ['rx_packets', 'rx_errors', 'rx_drop', 'tx_packets', 'tx_errors', 'tx_drop', 'read_req', 'write_req', 'cpu', 'memory'] for key in basic_attrs: self.assertIn(key, str(diagnostic.keys())) @test.attr(type='gate') def test_list_servers_filter_by_error_status(self): # Filter the list of servers by server error status params = {'status': 'error'} resp, server = self.client.reset_state(self.s1_id, state='error') resp, body = self.non_admin_client.list_servers(params) # Reset server's state to 'active' resp, server = self.client.reset_state(self.s1_id, state='active') # Verify server's state resp, server = self.client.get_server(self.s1_id) self.assertEqual(server['status'], 'ACTIVE') servers = body['servers'] # Verify error server in list result self.assertIn(self.s1_id, map(lambda x: x['id'], servers)) self.assertNotIn(self.s2_id, map(lambda x: x['id'], servers)) @test.attr(type='gate') def test_rebuild_server_in_error_state(self): # The server in error state should be rebuilt using the provided # image and changed to ACTIVE state # resetting vm state require admin privilege resp, server = self.client.reset_state(self.s1_id, state='error') self.assertEqual(202, resp.status) resp, rebuilt_server = self.non_admin_client.rebuild( self.s1_id, self.image_ref_alt) self.addCleanup(self.non_admin_client.wait_for_server_status, self.s1_id, 'ACTIVE') self.addCleanup(self.non_admin_client.rebuild, self.s1_id, self.image_ref) # Verify the properties in the initial response are correct self.assertEqual(self.s1_id, rebuilt_server['id']) rebuilt_image_id = rebuilt_server['image']['id'] self.assertEqual(self.image_ref_alt, rebuilt_image_id) self.assertEqual(self.flavor_ref, rebuilt_server['flavor']['id']) self.non_admin_client.wait_for_server_status(rebuilt_server['id'], 'ACTIVE', raise_on_error=False) # Verify the server properties after rebuilding resp, server = self.non_admin_client.get_server(rebuilt_server['id']) rebuilt_image_id = server['image']['id'] self.assertEqual(self.image_ref_alt, rebuilt_image_id) @test.attr(type='gate') def test_reset_network_inject_network_info(self): resp, server = self.create_test_server(wait_until='ACTIVE') # Reset Network of a Server resp, server_body = self.client.reset_network(server['id']) self.assertEqual(202, resp.status) # Inject the Network Info into Server resp, server = self.client.inject_network_info(server['id']) self.assertEqual(202, resp.status)