tempest/api/compute/admin/test_servers.py

# 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_lib import decorators

from tempest.api.compute import base
from tempest.common.utils import data_utils
from tempest import test


class ServersAdminTestJSON(base.BaseV2ComputeAdminTest):

    """
    Tests Servers API using admin privileges
    """

    _host_key = 'OS-EXT-SRV-ATTR:host'

    @classmethod
    def resource_setup(cls):
        super(ServersAdminTestJSON, cls).resource_setup()
        cls.client = cls.os_adm.servers_client
        cls.non_admin_client = cls.servers_client
        cls.flavors_client = cls.os_adm.flavors_client

        cls.s1_name = data_utils.rand_name('server')
        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')
        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
        body = self.client.list_servers_with_detail()
        servers = body['servers']
        self.assertEqual([], servers)

    @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'}
        self.client.reset_state(self.s1_id, state='error')
        body = self.non_admin_client.list_servers(params)
        # Reset server's state to 'active'
        self.client.reset_state(self.s1_id, state='active')
        # Verify server's state
        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_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': ''}
        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_by_admin_with_specified_tenant(self):
        # In nova v2, tenant_id is ignored unless all_tenants is specified

        # List the primary tenant but get nothing due to odd specified behavior
        tenant_id = self.non_admin_client.tenant_id
        params = {'tenant_id': tenant_id}
        body = self.client.list_servers_with_detail(params)
        servers = body['servers']
        self.assertEqual([], servers)

        # List the admin tenant which has no servers
        admin_tenant_id = self.client.tenant_id
        params = {'all_tenants': '', 'tenant_id': admin_tenant_id}
        body = self.client.list_servers_with_detail(params)
        servers = body['servers']
        self.assertEqual([], servers)

    @test.attr(type='gate')
    def test_list_servers_filter_by_exist_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
        test_server = self.client.create_server(name, image_id, flavor)
        self.addCleanup(self.client.delete_server, test_server['id'])
        self.client.wait_for_server_status(test_server['id'], 'ACTIVE')
        server = self.client.get_server(test_server['id'])
        self.assertEqual(server['status'], 'ACTIVE')
        hostname = server[self._host_key]
        params = {'host': hostname}
        body = self.client.list_servers(params)
        servers = body['servers']
        nonexistent_params = {'host': 'nonexistent_host'}
        nonexistent_body = self.client.list_servers(nonexistent_params)
        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'
        self.client.reset_state(self.s1_id)

        # Verify server's state
        server = self.client.get_server(self.s1_id)
        self.assertEqual(server['status'], 'ERROR')

        # Reset server's state to 'active'
        self.client.reset_state(self.s1_id, state='active')

        # Verify server's state
        server = self.client.get_server(self.s1_id)
        self.assertEqual(server['status'], 'ACTIVE')

    @test.attr(type='gate')
    @decorators.skip_because(bug="1240043")
    def test_get_server_diagnostics_by_admin(self):
        # Retrieve server diagnostics by admin user
        diagnostic = self.client.get_server_diagnostics(self.s1_id)
        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_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
        self.client.reset_state(self.s1_id, state='error')
        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
        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):
        # Reset Network of a Server
        server = self.create_test_server(wait_until='ACTIVE')
        self.client.reset_network(server['id'])
        # Inject the Network Info into Server
        self.client.inject_network_info(server['id'])

    @test.attr(type='gate')
    def test_create_server_with_scheduling_hint(self):
        # Create a server with scheduler hints.
        hints = {
            'same_host': self.s1_id
        }
        self.create_test_server(sched_hints=hints,
                                wait_until='ACTIVE')