# 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 time import pytest import allure from tests.api.volume import base from common import waiters from configs import config from lib import exceptions CONF = config.CONF class TestServerAttachSsdNvmeVolumes(base.BaseVolumeTest): """Test Server Attach and Detach with Volumes scenario test cases""" create_default_network = True @classmethod def setup_clients(cls): super(TestServerAttachSsdNvmeVolumes, cls).setup_clients() # If there is vpc_network, add ports client for testing. if CONF.network.vpc_network_id: cls.ports_client = cls.os_primary.ports_client @classmethod def resource_setup(cls): super(TestServerAttachSsdNvmeVolumes, cls).resource_setup() # Create a test shared volume for attach/detach tests if not CONF.common.availability_zone_list or len(CONF.common.availability_zone_list) < 1: raise cls.skipException("can't resource setup, there is no availability_zone_list") cls.volume = cls.create_volume(availability_zone=CONF.common.availability_zone_list[0]) def _create_multi_volumes(self, vol_type, size, create_num=5): created_vols = [] for _ in range(0, create_num): volume = self.create_volume(volume_type=vol_type, size=size) if not volume["id"]: return None created_vols.append(volume) return created_vols def _attach_multi_volumes(self, vol_list, server_id): for volume in vol_list: try: self.attach_volume( server_id=server_id, volume_id=volume["id"], wait_for_detach=False, is_cleanup=False, ) except exceptions.TimeoutException: return ( False, f"Failed to attach volume({volume['id']}) to server({server_id})", ) return True, "" def _detach_multi_volumes(self, vol_list, server_id): for volume in vol_list: resp = self.servers_client.detach_volume(server_id, volume_id=volume["id"]) if resp.response["status"] != "202": return ( False, f"Failed to detached volume({volume['id']}) from server({server_id})", ) if not self._wait_volume_status( volume_id=volume["id"], wait_status="available" ): return ( False, f"Failed to wait volume({volume['id']}) status to 'available' after volume detach", ) return True, "" def _wait_volume_status(self, volume_id, wait_status): try: waiters.wait_for_volume_resource_status( self.volumes_client, volume_id, wait_status ) except exceptions.TimeoutException: return False return True def _verify_create_vpc_server(self, network_id): with allure.step("Create server with selecting the VPC network"): server_with_vpc = self.create_server( networks=[{"uuid": network_id}], wait_until="ACTIVE", ) with allure.step("Verify server is using selected VPC network"): # Need to use the Neutron API to get network_id info connected to the server.. ports = self.ports_client.list_ports(device_id=server_with_vpc["id"])[ "ports" ] # Initialize a variable to track if we have found a matching network_id matching_network_id_found = False # Iterate over the ports to check for a matching network_id for port in ports: # If the network ID of the attached port matches the expected VPC network ID, set the flag to True if port["network_id"] == network_id: matching_network_id_found = True break # No need to check further ports # After checking all ports, assert that we found a matching network_id assert ( matching_network_id_found ), "None of the attached ports have a network ID that matches the expected VPC network ID." return server_with_vpc def _verify_create_volumes_and_attach_to_server(self, server, vol_type): with allure.step(f"Create multi volumes with VolumeType:{vol_type}"): volume_list = self._create_multi_volumes(vol_type=vol_type, size=10) assert volume_list, "Failed to create multi volumes" with allure.step( f"Verify Attach multi volumes to server({server['id']}) VolumeType:{vol_type}" ): is_success, fail_msg = self._attach_multi_volumes( vol_list=volume_list, server_id=server["id"] ) assert is_success, fail_msg return volume_list @pytest.mark.positive @pytest.mark.skipif(not CONF.common.availability_zone_list or len(CONF.common.availability_zone_list) < 2, reason='Multi AZ for volume not available.') @pytest.mark.skipif(not CONF.volume.volume_type_ssd, reason='SSD volume type not available.') def test_attach_detach_ssd_volume_to_instance_with_multi_az(self): """Test attaching and detaching SSD type volume to instance""" vol_type = CONF.volume.volume_type_ssd for vol_az in CONF.common.availability_zone_list: # Create ssd volume with az with allure.step(f"Create a volume with AZ:{vol_az}, VolumeType:{vol_type}"): self.volume = self.create_volume(availability_zone=vol_az, volume_type=vol_type) assert self.volume['id'], "Failed to create volume for test" # Create a server with az with allure.step(f"Create a server with AZ:{vol_az}, VolumeType:{vol_type}"): server = self.create_server(availability_zone=vol_az) assert server['id'], "Failed to create server for test" # Volume is attached and detached successfully from an instance with allure.step(f"Validate Attached volume to instance with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.attach_volume(server['id'], volumeId=self.volume['id']) assert resp.response['status'] == '200', "Failed to attached volume to instance" with allure.step(f"Wait Volume status to 'in-use' after volume attach with AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'in-use') except exceptions.TimeoutException: assert False, "Failed to wait volume status to 'in-use' after volume attach" # Wait reason: There is a problem if run Detach immediately after status 'in-use' changed. time.sleep(5) with allure.step(f"Validate Detached volume to instance with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.detach_volume(server['id'], volume_id=self.volume['id']) assert resp.response['status'] == '202', "Failed to detached volume to instance" with allure.step(f"Wait Volume status to 'available' after volume detach with AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'available') except exceptions.TimeoutException: assert False, "Failed to wait volume status to 'available' after volume detach" # Re-attach volume to test volume return 'available' status after instance deleted with allure.step(f"Attached volume to instance with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.attach_volume(server['id'], volumeId=self.volume['id']) assert resp.response['status'] == '200', "Failed to attached volume to instance" try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'in-use') except exceptions.TimeoutException: assert False, "Failed to wait volume status to 'in-use' after volume attach" with allure.step(f"Delete instance with attach volume AZ:{vol_az}, VolumeType:{vol_type}"): self.servers_client.delete_server(server['id']) try: waiters.wait_for_server_termination(self.servers_client, server['id']) except exceptions.TimeoutException: assert False, "Failed to instance server with attach volume" with allure.step(f"Verify volume return to 'available' status AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'available') except exceptions.TimeoutException: assert False, "Failed volume return to 'available' status" @pytest.mark.positive @pytest.mark.skipif(not CONF.common.availability_zone_list or len(CONF.common.availability_zone_list) < 2, reason='Multi AZ for volume not available.') @pytest.mark.skipif(not CONF.volume.volume_type_nvme, reason='NVME volume type not available.') def test_attach_detach_nvme_volume_to_instance_with_multi_az(self): """Test attaching and detaching NVME type volume to instance""" vol_type = CONF.volume.volume_type_nvme for vol_az in CONF.common.availability_zone_list: # Create nvme volume with az with allure.step(f"Create a volume with AZ:{vol_az}, VolumeType:{vol_type}"): self.volume = self.create_volume(availability_zone=vol_az, volume_type=vol_type) assert self.volume['id'], "Failed to create volume for test" # Create a server with az with allure.step(f"Create a server with AZ:{vol_az}, VolumeType:{vol_type}"): server = self.create_server(availability_zone=vol_az) assert server['id'], "Failed to create server for test" # Volume is attached and detached successfully from an instance with allure.step(f"Validate Attached volume to instance with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.attach_volume(server['id'], volumeId=self.volume['id']) assert resp.response['status'] == '200', "Failed to attached volume to instance" with allure.step(f"Wait Volume status to 'in-use' after volume attach with AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'in-use') except exceptions.TimeoutException: assert False, "Failed to wait volume status to 'in-use' after volume attach" # Wait reason: There is a problem if run Detach immediately after status 'in-use' changed. time.sleep(5) with allure.step(f"Validate Detached volume to instance with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.detach_volume(server['id'], volume_id=self.volume['id']) assert resp.response['status'] == '202', "Failed to detached volume to instance" with allure.step(f"Wait Volume status to 'available' after volume detach with AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'available') except exceptions.TimeoutException: assert False, "Failed to wait volume status to 'available' after volume detach" # Re-attach volume to test volume return 'available' status after instance deleted with allure.step(f"Attached volume to instance with AZ:{vol_az}, VolumeType:{vol_type}"): resp = self.servers_client.attach_volume(server['id'], volumeId=self.volume['id']) assert resp.response['status'] == '200', "Failed to attached volume to instance" try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'in-use') except exceptions.TimeoutException: assert False, "Failed to wait volume status to 'in-use' after volume attach" with allure.step(f"Delete instance with attach volume AZ:{vol_az}, VolumeType:{vol_type}"): self.servers_client.delete_server(server['id']) try: waiters.wait_for_server_termination(self.servers_client, server['id']) except exceptions.TimeoutException: assert False, "Failed to instance server with attach volume" with allure.step(f"Verify volume return to 'available' status AZ:{vol_az}, VolumeType:{vol_type}"): try: waiters.wait_for_volume_resource_status(self.volumes_client, self.volume['id'], 'available') except exceptions.TimeoutException: assert False, "Failed volume return to 'available' status" @pytest.mark.positive @pytest.mark.skipif( not CONF.volume.volume_type_ssd, reason="SSD volume type not available." ) def test_attach_detach_ssd_volume_to_another_vm_with_multi_volumes(self): """Test attaching and detaching SSD type volume to another vm with multi volumes""" vol_type = CONF.volume.volume_type_ssd with allure.step(f"Create multi volumes with VolumeType:{vol_type}"): volume_list = self._create_multi_volumes(vol_type=vol_type, size=10) assert volume_list, "Failed to create multi volumes" with allure.step("Create test servers"): server_1 = self.create_server() server_2 = self.create_server() assert server_1["id"], "Failed to create server_1 for test" assert server_2["id"], "Failed to create server_2 for test" with allure.step( f"Validate Attach multi volumes to vm({server_1['id']}) VolumeType:{vol_type}" ): is_success, fail_msg = self._attach_multi_volumes( vol_list=volume_list, server_id=server_1["id"] ) assert is_success, fail_msg # Wait reason: There is a problem if run Detach immediately after status 'in-use' changed. time.sleep(5) with allure.step( f"Validate Detach multi volumes from vm({server_1['id']}) with VolumeType:{vol_type}" ): is_success, fail_msg = self._detach_multi_volumes( vol_list=volume_list, server_id=server_1["id"] ) assert is_success, fail_msg with allure.step( f"Verify attach multi volumes to another vm({server_2['id']}) VolumeType:{vol_type}" ): is_success, fail_msg = self._attach_multi_volumes( vol_list=volume_list, server_id=server_2["id"] ) assert is_success, fail_msg with allure.step( f"Delete another vm({server_2['id']}) with attached multi volumes VolumeType:{vol_type}" ): self.servers_client.delete_server(server_2["id"]) try: waiters.wait_for_server_termination(self.servers_client, server_2["id"]) except exceptions.TimeoutException: assert ( False ), f"Failed to another vm({server_2['id']}) with attach volume" with allure.step( f"Verify multi volumes return to 'available' VolumeType:{vol_type}" ): for volume in volume_list: is_success = self._wait_volume_status( volume_id=volume["id"], wait_status="available" ) assert ( is_success ), f"Failed to volume({volume['id']})return 'available' status" @pytest.mark.positive @pytest.mark.skipif( not CONF.volume.volume_type_nvme, reason="NVME volume type not available." ) def test_attach_detach_nvme_volume_to_another_vm_with_multi_volumes(self): """Test attaching and detaching NVME type volume to another vm with multi volumes""" vol_type = CONF.volume.volume_type_nvme with allure.step(f"Create multi volumes with VolumeType:{vol_type}"): volume_list = self._create_multi_volumes(vol_type=vol_type, size=10) assert volume_list, "Failed to create multi volumes" with allure.step("Create test servers"): server_1 = self.create_server() server_2 = self.create_server() assert server_1["id"], "Failed to create server_1 for test" assert server_2["id"], "Failed to create server_2 for test" with allure.step( f"Validate Attach multi volumes to vm({server_1['id']}) VolumeType:{vol_type}" ): is_success, fail_msg = self._attach_multi_volumes( vol_list=volume_list, server_id=server_1["id"] ) assert is_success, fail_msg # Wait reason: There is a problem if run Detach immediately after status 'in-use' changed. time.sleep(5) with allure.step( f"Validate Detach multi volumes from vm({server_1['id']}) with VolumeType:{vol_type}" ): is_success, fail_msg = self._detach_multi_volumes( vol_list=volume_list, server_id=server_1["id"] ) assert is_success, fail_msg with allure.step( f"Verify Attach multi volumes to another vm({server_2['id']}) VolumeType:{vol_type}" ): is_success, fail_msg = self._attach_multi_volumes( vol_list=volume_list, server_id=server_2["id"] ) assert is_success, fail_msg with allure.step( f"Delete another vm({server_2['id']}) with attach multi volumes VolumeType:{vol_type}" ): self.servers_client.delete_server(server_2["id"]) try: waiters.wait_for_server_termination(self.servers_client, server_2["id"]) except exceptions.TimeoutException: assert ( False ), f"Failed to delete vm({server_2['id']}) attached to the volume" with allure.step( f"Verify multi volumes return to 'available' VolumeType:{vol_type}" ): for volume in volume_list: is_success = self._wait_volume_status( volume_id=volume["id"], wait_status="available" ) assert ( is_success ), f"Failed to volume({volume['id']})return 'available' status" @pytest.mark.positive @pytest.mark.skipif( not CONF.network.vpc_network_id, reason="there is no predefined vpc network" ) @pytest.mark.skipif( not CONF.volume.volume_type_ssd, reason="SSD volume type not available." ) def test_create_server_with_vpc_network_and_attach_detach_ssd_volume(self): """Test creating server with the selecting VPC network and SSD Volume attach and detach by server delete""" # Get predefined VPC network id network_id = CONF.network.vpc_network_id vol_type = CONF.volume.volume_type_ssd with allure.step("Verify create server with selecting the VPC network"): server_with_vpc = self._verify_create_vpc_server(network_id=network_id) with allure.step( f"Verify create multi volumes and attach to vpc server{server_with_vpc['id']} with VolumeType:{vol_type}" ): volume_list = self._verify_create_volumes_and_attach_to_server( server=server_with_vpc, vol_type=vol_type ) with allure.step(f"Delete vpc server with attach VolumeType:{vol_type}"): self.servers_client.delete_server(server_with_vpc["id"]) try: waiters.wait_for_server_termination( self.servers_client, server_with_vpc["id"] ) except exceptions.TimeoutException: assert False, "Failed to delete vpc server with attach volumes" with allure.step( f"Verify multi volumes return to 'available' VolumeType:{vol_type}" ): for volume in volume_list: is_success = self._wait_volume_status( volume_id=volume["id"], wait_status="available" ) assert ( is_success ), f"Failed to volume({volume['id']})return 'available' status" @pytest.mark.positive @pytest.mark.skipif( not CONF.network.vpc_network_id, reason="there is no predefined vpc network" ) @pytest.mark.skipif( not CONF.volume.volume_type_nvme, reason="NVME volume type not available." ) def test_create_server_with_vpc_network_and_attach_detach_nvme_volume(self): """Test creating server with the selecting VPC network and NVME Volume attach and detach by server delete""" # Get predefined VPC network id network_id = CONF.network.vpc_network_id vol_type = CONF.volume.volume_type_nvme with allure.step("Verify create server with selecting the vpc network"): server_with_vpc = self._verify_create_vpc_server(network_id=network_id) with allure.step( f"Verify create multi volumes and attach to vpc server{server_with_vpc['id']} with VolumeType:{vol_type}" ): volume_list = self._verify_create_volumes_and_attach_to_server( server=server_with_vpc, vol_type=vol_type ) with allure.step(f"Delete vpc server with attach VolumeType:{vol_type}"): self.servers_client.delete_server(server_with_vpc["id"]) try: waiters.wait_for_server_termination( self.servers_client, server_with_vpc["id"] ) except exceptions.TimeoutException: assert False, "Failed to delete vpc server with attach volumes" with allure.step( f"Verify multi volumes return to 'available' VolumeType:{vol_type}" ): for volume in volume_list: is_success = self._wait_volume_status( volume_id=volume["id"], wait_status="available" ) assert ( is_success ), f"Failed to volume({volume['id']})return 'available' status"