How to use show_image_tasks method in tempest

Best Python code snippet using tempest_python

test_waiters.py

Source:test_waiters.py

1# Copyright 2014 IBM Corp.2#3#    Licensed under the Apache License, Version 2.0 (the "License"); you may4#    not use this file except in compliance with the License. You may obtain5#    a copy of the License at6#7#         http://www.apache.org/licenses/LICENSE-2.08#9#    Unless required by applicable law or agreed to in writing, software10#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12#    License for the specific language governing permissions and limitations13#    under the License.14import time15from unittest import mock16from oslo_utils.fixture import uuidsentinel as uuids17from tempest.common import waiters18from tempest import exceptions19from tempest.lib import exceptions as lib_exc20from tempest.lib.services.compute import servers_client21from tempest.lib.services.volume.v2 import volumes_client22from tempest.tests import base23import tempest.tests.utils as utils24class TestImageWaiters(base.TestCase):25    def setUp(self):26        super(TestImageWaiters, self).setUp()27        self.client = mock.MagicMock()28        self.client.build_timeout = 129        self.client.build_interval = 130    def test_wait_for_image_status(self):31        self.client.show_image.return_value = ({'status': 'active'})32        start_time = int(time.time())33        waiters.wait_for_image_status(self.client, 'fake_image_id', 'active')34        end_time = int(time.time())35        # Ensure waiter returns before build_timeout36        self.assertLess((end_time - start_time), 10)37    def test_wait_for_image_status_timeout(self):38        time_mock = self.patch('time.time')39        time_mock.side_effect = utils.generate_timeout_series(1)40        self.client.show_image.return_value = ({'status': 'saving'})41        self.assertRaises(lib_exc.TimeoutException,42                          waiters.wait_for_image_status,43                          self.client, 'fake_image_id', 'active')44    def test_wait_for_image_status_error_on_image_create(self):45        self.client.show_image.return_value = ({'status': 'ERROR'})46        self.assertRaises(exceptions.AddImageException,47                          waiters.wait_for_image_status,48                          self.client, 'fake_image_id', 'active')49    def test_wait_for_image_imported_to_stores(self):50        self.client.show_image.return_value = ({'status': 'active',51                                                'stores': 'fake_store'})52        start_time = int(time.time())53        waiters.wait_for_image_imported_to_stores(54            self.client, 'fake_image_id', 'fake_store')55        end_time = int(time.time())56        # Ensure waiter returns before build_timeout57        self.assertLess((end_time - start_time), 10)58    def test_wait_for_image_imported_to_stores_failure(self):59        time_mock = self.patch('time.time')60        client = mock.MagicMock()61        client.build_timeout = 262        self.patch('time.time', side_effect=[0., 1., 2.])63        time_mock.side_effect = utils.generate_timeout_series(1)64        client.show_image.return_value = ({65            'status': 'saving',66            'stores': 'fake_store',67            'os_glance_failed_import': 'fake_os_glance_failed_import'})68        self.assertRaises(lib_exc.OtherRestClientException,69                          waiters.wait_for_image_imported_to_stores,70                          client, 'fake_image_id', 'fake_store')71    def test_wait_for_image_imported_to_stores_timeout(self):72        time_mock = self.patch('time.time')73        client = mock.MagicMock()74        client.build_timeout = 275        self.patch('time.time', side_effect=[0., 1., 2.])76        time_mock.side_effect = utils.generate_timeout_series(1)77        client.show_image.return_value = ({78            'status': 'saving',79            'stores': 'fake_store'})80        self.assertRaises(lib_exc.TimeoutException,81                          waiters.wait_for_image_imported_to_stores,82                          client, 'fake_image_id', 'fake_store')83    def test_wait_for_image_copied_to_stores(self):84        self.client.show_image.return_value = ({85            'status': 'active',86            'os_glance_importing_to_stores': '',87            'os_glance_failed_import': 'fake_os_glance_failed_import'})88        start_time = int(time.time())89        waiters.wait_for_image_copied_to_stores(90            self.client, 'fake_image_id')91        end_time = int(time.time())92        # Ensure waiter returns before build_timeout93        self.assertLess((end_time - start_time), 10)94    def test_wait_for_image_copied_to_stores_timeout(self):95        time_mock = self.patch('time.time')96        self.patch('time.time', side_effect=[0., 1.])97        time_mock.side_effect = utils.generate_timeout_series(1)98        self.client.show_image.return_value = ({99            'status': 'active',100            'os_glance_importing_to_stores': 'processing',101            'os_glance_failed_import': 'fake_os_glance_failed_import'})102        self.assertRaises(lib_exc.TimeoutException,103                          waiters.wait_for_image_copied_to_stores,104                          self.client, 'fake_image_id')105    def test_wait_for_image_tasks_status(self):106        self.client.show_image_tasks.return_value = ({107            'tasks': [{'status': 'success'}]})108        start_time = int(time.time())109        waiters.wait_for_image_tasks_status(110            self.client, 'fake_image_id', 'success')111        end_time = int(time.time())112        # Ensure waiter returns before build_timeout113        self.assertLess((end_time - start_time), 10)114    def test_wait_for_image_tasks_status_timeout(self):115        time_mock = self.patch('time.time')116        self.patch('time.time', side_effect=[0., 1.])117        time_mock.side_effect = utils.generate_timeout_series(1)118        self.client.show_image_tasks.return_value = ({119            'tasks': [120                {'status': 'success'},121                {'status': 'processing'}]})122        self.assertRaises(lib_exc.TimeoutException,123                          waiters.wait_for_image_tasks_status,124                          self.client, 'fake_image_id', 'success')125class TestInterfaceWaiters(base.TestCase):126    build_timeout = 1.127    build_interval = 1128    port_down = {'interfaceAttachment': {'port_state': 'DOWN'}}129    port_active = {'interfaceAttachment': {'port_state': 'ACTIVE'}}130    def mock_client(self, **kwargs):131        return mock.MagicMock(132            build_timeout=self.build_timeout,133            build_interval=self.build_interval,134            **kwargs)135    def test_wait_for_interface_status(self):136        show_interface = mock.Mock(137            side_effect=[self.port_down, self.port_active])138        client = self.mock_client(show_interface=show_interface)139        self.patch('time.time', return_value=0.)140        sleep = self.patch('time.sleep')141        result = waiters.wait_for_interface_status(142            client, 'server_id', 'port_id', 'ACTIVE')143        self.assertIs(self.port_active['interfaceAttachment'], result)144        show_interface.assert_has_calls([mock.call('server_id', 'port_id'),145                                         mock.call('server_id', 'port_id')])146        sleep.assert_called_once_with(client.build_interval)147    def test_wait_for_interface_status_timeout(self):148        show_interface = mock.MagicMock(return_value=self.port_down)149        client = self.mock_client(show_interface=show_interface)150        self.patch('time.time', side_effect=[0., client.build_timeout + 1.])151        sleep = self.patch('time.sleep')152        self.assertRaises(lib_exc.TimeoutException,153                          waiters.wait_for_interface_status,154                          client, 'server_id', 'port_id', 'ACTIVE')155        show_interface.assert_has_calls([mock.call('server_id', 'port_id'),156                                         mock.call('server_id', 'port_id')])157        sleep.assert_called_once_with(client.build_interval)158    def test_wait_for_interface_detach(self):159        no_event = {160            'instanceAction': {161                'events': []162            }163        }164        one_event_without_result = {165            'instanceAction': {166                'events': [167                    {168                        'event': 'compute_detach_interface',169                        'result': None170                    }171                ]172            }173        }174        one_event_successful = {175            'instanceAction': {176                'events': [177                    {178                        'event': 'compute_detach_interface',179                        'result': 'Success'180                    }181                ]182            }183        }184        show_instance_action = mock.MagicMock(185            # there is an extra call to return the result from the waiter186            side_effect=[187                no_event,188                one_event_without_result,189                one_event_successful,190                one_event_successful,191            ]192        )193        client = self.mock_client(show_instance_action=show_instance_action)194        self.patch('time.time', return_value=0.)195        sleep = self.patch('time.sleep')196        result = waiters.wait_for_interface_detach(197            client, mock.sentinel.server_id, mock.sentinel.port_id,198            mock.sentinel.detach_request_id199        )200        self.assertIs(one_event_successful['instanceAction'], result)201        show_instance_action.assert_has_calls(202            # there is an extra call to return the result from the waiter203            [204                mock.call(205                    mock.sentinel.server_id, mock.sentinel.detach_request_id)206            ] * 4207        )208        sleep.assert_has_calls([mock.call(client.build_interval)] * 2)209    def test_wait_for_interface_detach_timeout(self):210        one_event_without_result = {211            'instanceAction': {212                'events': [213                    {214                        'event': 'compute_detach_interface',215                        'result': None216                    }217                ]218            }219        }220        show_instance_action = mock.MagicMock(221            return_value=one_event_without_result)222        client = self.mock_client(show_instance_action=show_instance_action)223        self.patch('time.time', side_effect=[0., client.build_timeout + 1.])224        sleep = self.patch('time.sleep')225        self.assertRaises(226            lib_exc.TimeoutException,227            waiters.wait_for_interface_detach,228            client, mock.sentinel.server_id, mock.sentinel.port_id,229            mock.sentinel.detach_request_id230        )231        show_instance_action.assert_has_calls(232            [233                mock.call(234                    mock.sentinel.server_id, mock.sentinel.detach_request_id)235            ] * 2236        )237        sleep.assert_called_once_with(client.build_interval)238class TestVolumeWaiters(base.TestCase):239    vol_migrating_src_host = {240        'volume': {'migration_status': 'migrating',241                   'os-vol-host-attr:host': 'src_host@backend#type'}}242    vol_migrating_dst_host = {243        'volume': {'migration_status': 'migrating',244                   'os-vol-host-attr:host': 'dst_host@backend#type'}}245    vol_migration_success = {246        'volume': {'migration_status': 'success',247                   'os-vol-host-attr:host': 'dst_host@backend#type'}}248    vol_migration_error = {249        'volume': {'migration_status': 'error',250                   'os-vol-host-attr:host': 'src_host@backend#type'}}251    def test_wait_for_volume_migration_timeout(self):252        show_volume = mock.MagicMock(return_value=self.vol_migrating_src_host)253        client = mock.Mock(spec=volumes_client.VolumesClient,254                           resource_type="volume",255                           build_interval=1,256                           build_timeout=1,257                           show_volume=show_volume)258        self.patch('time.time', side_effect=[0., client.build_timeout + 1.])259        self.patch('time.sleep')260        self.assertRaises(lib_exc.TimeoutException,261                          waiters.wait_for_volume_migration,262                          client, mock.sentinel.volume_id, 'dst_host')263    def test_wait_for_volume_migration_error(self):264        show_volume = mock.MagicMock(side_effect=[265            self.vol_migrating_src_host,266            self.vol_migrating_src_host,267            self.vol_migration_error])268        client = mock.Mock(spec=volumes_client.VolumesClient,269                           resource_type="volume",270                           build_interval=1,271                           build_timeout=1,272                           show_volume=show_volume)273        self.patch('time.time', return_value=0.)274        self.patch('time.sleep')275        self.assertRaises(lib_exc.TempestException,276                          waiters.wait_for_volume_migration,277                          client, mock.sentinel.volume_id, 'dst_host')278    def test_wait_for_volume_migration_success_and_dst(self):279        show_volume = mock.MagicMock(side_effect=[280            self.vol_migrating_src_host,281            self.vol_migrating_dst_host,282            self.vol_migration_success])283        client = mock.Mock(spec=volumes_client.VolumesClient,284                           resource_type="volume",285                           build_interval=1,286                           build_timeout=1,287                           show_volume=show_volume)288        self.patch('time.time', return_value=0.)289        self.patch('time.sleep')290        waiters.wait_for_volume_migration(291            client, mock.sentinel.volume_id, 'dst_host')292        # Assert that we wait until migration_status is success and dst_host is293        # part of the returned os-vol-host-attr:host.294        show_volume.assert_has_calls([mock.call(mock.sentinel.volume_id),295                                      mock.call(mock.sentinel.volume_id),296                                      mock.call(mock.sentinel.volume_id)])297    @mock.patch.object(time, 'sleep')298    def test_wait_for_volume_status_error_restoring(self, mock_sleep):299        # Tests that the wait method raises VolumeRestoreErrorException if300        # the volume status is 'error_restoring'.301        client = mock.Mock(spec=volumes_client.VolumesClient,302                           resource_type="volume",303                           build_interval=1)304        volume1 = {'volume': {'status': 'restoring-backup'}}305        volume2 = {'volume': {'status': 'error_restoring'}}306        mock_show = mock.Mock(side_effect=(volume1, volume2))307        client.show_volume = mock_show308        volume_id = '7532b91e-aa0a-4e06-b3e5-20c0c5ee1caa'309        self.assertRaises(exceptions.VolumeRestoreErrorException,310                          waiters.wait_for_volume_resource_status,311                          client, volume_id, 'available')312        mock_show.assert_has_calls([mock.call(volume_id),313                                    mock.call(volume_id)])314        mock_sleep.assert_called_once_with(1)315    @mock.patch.object(time, 'sleep')316    def test_wait_for_volume_status_error_extending(self, mock_sleep):317        # Tests that the wait method raises VolumeExtendErrorException if318        # the volume status is 'error_extending'.319        client = mock.Mock(spec=volumes_client.VolumesClient,320                           resource_type="volume",321                           build_interval=1)322        volume1 = {'volume': {'status': 'extending'}}323        volume2 = {'volume': {'status': 'error_extending'}}324        mock_show = mock.Mock(side_effect=(volume1, volume2))325        client.show_volume = mock_show326        volume_id = '7532b91e-aa0a-4e06-b3e5-20c0c5ee1caa'327        self.assertRaises(exceptions.VolumeExtendErrorException,328                          waiters.wait_for_volume_resource_status,329                          client, volume_id, 'available')330        mock_show.assert_has_calls([mock.call(volume_id),331                                    mock.call(volume_id)])332        mock_sleep.assert_called_once_with(1)333    def test_wait_for_volume_attachment_create(self):334        vol_detached = {'volume': {'attachments': []}}335        vol_attached = {'volume': {'attachments': [336                       {'id': uuids.volume_id,337                        'attachment_id': uuids.attachment_id,338                        'server_id': uuids.server_id,339                        'volume_id': uuids.volume_id}]}}340        show_volume = mock.MagicMock(side_effect=[341            vol_detached, vol_detached, vol_attached])342        client = mock.Mock(spec=volumes_client.VolumesClient,343                           build_interval=1,344                           build_timeout=5,345                           show_volume=show_volume)346        self.patch('time.time')347        self.patch('time.sleep')348        att = waiters.wait_for_volume_attachment_create(349            client, uuids.volume_id, uuids.server_id)350        assert att == vol_attached['volume']['attachments'][0]351        # Assert that show volume is called until the attachment is removed.352        show_volume.assert_has_calls([mock.call(uuids.volume_id),353                                      mock.call(uuids.volume_id),354                                      mock.call(uuids.volume_id)])355    def test_wait_for_volume_attachment(self):356        vol_detached = {'volume': {'attachments': []}}357        vol_attached = {'volume': {'attachments': [358                       {'attachment_id': uuids.attachment_id}]}}359        show_volume = mock.MagicMock(side_effect=[360            vol_attached, vol_attached, vol_detached])361        client = mock.Mock(spec=volumes_client.VolumesClient,362                           build_interval=1,363                           build_timeout=5,364                           show_volume=show_volume)365        self.patch('time.time')366        self.patch('time.sleep')367        waiters.wait_for_volume_attachment_remove(client, uuids.volume_id,368                                                  uuids.attachment_id)369        # Assert that show volume is called until the attachment is removed.370        show_volume.assert_has_calls([mock.call(uuids.volume_id),371                                      mock.call(uuids.volume_id),372                                      mock.call(uuids.volume_id)])373    def test_wait_for_volume_attachment_timeout(self):374        show_volume = mock.MagicMock(return_value={375            'volume': {'attachments': [376                {'attachment_id': uuids.attachment_id}]}})377        client = mock.Mock(spec=volumes_client.VolumesClient,378                           build_interval=1,379                           build_timeout=1,380                           show_volume=show_volume)381        self.patch('time.time', side_effect=[0., client.build_timeout + 1.])382        self.patch('time.sleep')383        # Assert that a timeout is raised if the attachment remains.384        self.assertRaises(lib_exc.TimeoutException,385                          waiters.wait_for_volume_attachment_remove,386                          client, uuids.volume_id, uuids.attachment_id)387    def test_wait_for_volume_attachment_not_present(self):388        show_volume = mock.MagicMock(return_value={389            'volume': {'attachments': []}})390        client = mock.Mock(spec=volumes_client.VolumesClient,391                           build_interval=1,392                           build_timeout=1,393                           show_volume=show_volume)394        self.patch('time.time', side_effect=[0., client.build_timeout + 1.])395        self.patch('time.sleep')396        waiters.wait_for_volume_attachment_remove(client, uuids.volume_id,397                                                  uuids.attachment_id)398        # Assert that show volume is only called once before we return399        show_volume.assert_called_once_with(uuids.volume_id)400    def test_wait_for_volume_attachment_remove_from_server(self):401        volume_attached = {402            "volumeAttachments": [{"volumeId": uuids.volume_id}]}403        volume_not_attached = {"volumeAttachments": []}404        mock_list_volume_attachments = mock.Mock(405            side_effect=[volume_attached, volume_not_attached])406        mock_client = mock.Mock(407            spec=servers_client.ServersClient,408            build_interval=1,409            build_timeout=1,410            list_volume_attachments=mock_list_volume_attachments)411        self.patch(412            'time.time',413            side_effect=[0., 0.5, mock_client.build_timeout + 1.])414        self.patch('time.sleep')415        waiters.wait_for_volume_attachment_remove_from_server(416            mock_client, uuids.server_id, uuids.volume_id)417        # Assert that list_volume_attachments is called until the attachment is418        # removed.419        mock_list_volume_attachments.assert_has_calls([420            mock.call(uuids.server_id),421            mock.call(uuids.server_id)])422    def test_wait_for_volume_attachment_remove_from_server_timeout(self):423        volume_attached = {424            "volumeAttachments": [{"volumeId": uuids.volume_id}]}425        mock_list_volume_attachments = mock.Mock(426            side_effect=[volume_attached, volume_attached])427        mock_get_console_output = mock.Mock(428            return_value={'output': 'output'})429        mock_client = mock.Mock(430            spec=servers_client.ServersClient,431            build_interval=1,432            build_timeout=1,433            list_volume_attachments=mock_list_volume_attachments,434            get_console_output=mock_get_console_output)435        self.patch(436            'time.time',437            side_effect=[0., 0.5, mock_client.build_timeout + 1.])438        self.patch('time.sleep')439        self.assertRaises(440            lib_exc.TimeoutException,441            waiters.wait_for_volume_attachment_remove_from_server,442            mock_client, uuids.server_id, uuids.volume_id)443        # Assert that list_volume_attachments is called until the attachment is444        # removed.445        mock_list_volume_attachments.assert_has_calls([446            mock.call(uuids.server_id),447            mock.call(uuids.server_id)])448        # Assert that we fetch console output449        mock_get_console_output.assert_called_once_with(uuids.server_id)450    def test_wait_for_volume_attachment_remove_from_server_not_found(self):451        mock_list_volume_attachments = mock.Mock(452            side_effect=lib_exc.NotFound)453        mock_client = mock.Mock(454            spec=servers_client.ServersClient,455            list_volume_attachments=mock_list_volume_attachments)456        # Assert that nothing is raised when lib_exc_NotFound is raised457        # by the client call to list_volume_attachments458        waiters.wait_for_volume_attachment_remove_from_server(459            mock_client, mock.sentinel.server_id, mock.sentinel.volume_id)460        # Assert that list_volume_attachments was actually called461        mock_list_volume_attachments.assert_called_once_with(462            mock.sentinel.server_id)463    @mock.patch('os.system')464    def test_wait_for_ping_host_alive(self, mock_ping):465        mock_ping.return_value = 0466        # Assert that nothing is raised as the host is alive467        waiters.wait_for_ping('127.0.0.1', 10, 1)468    @mock.patch('os.system')469    def test_wait_for_ping_host_eventually_alive(self, mock_ping):470        mock_ping.side_effect = [1, 1, 0]471        # Assert that nothing is raised when the host is eventually alive472        waiters.wait_for_ping('127.0.0.1', 10, 1)473    @mock.patch('os.system')474    def test_wait_for_ping_timeout(self, mock_ping):475        mock_ping.return_value = 1476        # Assert that TimeoutException is raised when the host is dead477        self.assertRaises(478            lib_exc.TimeoutException,479            waiters.wait_for_ping,480            '127.0.0.1',481            .1,482            .1483        )484    def test_wait_for_ssh(self):485        mock_ssh_client = mock.Mock()486        mock_ssh_client.validate_authentication.return_value = True487        # Assert that nothing is raised when validate_authentication returns488        waiters.wait_for_ssh(mock_ssh_client, .1)489        mock_ssh_client.validate_authentication.assert_called_once()490    def test_wait_for_ssh_eventually_up(self):491        mock_ssh_client = mock.Mock()492        timeout = lib_exc.SSHTimeout(493            host='foo',494            username='bar',495            password='fizz'496        )497        mock_ssh_client.validate_authentication.side_effect = [498            timeout,499            timeout,500            True501        ]502        # Assert that nothing is raised if validate_authentication passes503        # before the timeout504        waiters.wait_for_ssh(mock_ssh_client, 10)505    def test_wait_for_ssh_timeout(self):506        mock_ssh_client = mock.Mock()507        timeout = lib_exc.SSHTimeout(508            host='foo',509            username='bar',510            password='fizz'511        )512        mock_ssh_client.validate_authentication.side_effect = timeout513        # Assert that TimeoutException is raised when validate_authentication514        # doesn't pass in time.515        self.assertRaises(516            lib_exc.TimeoutException,517            waiters.wait_for_ssh,518            mock_ssh_client,519            .1520        )521class TestServerFloatingIPWaiters(base.TestCase):522    def test_wait_for_server_floating_ip_associate_timeout(self):523        mock_server = {'server': {'id': 'fake_uuid', 'addresses': {}}}524        mock_client = mock.Mock(525            spec=servers_client.ServersClient,526            build_timeout=1, build_interval=1,527            show_server=lambda id: mock_server)528        fake_server = {'id': 'fake-uuid'}529        fake_fip = {'floating_ip_address': 'fake_address'}530        self.assertRaises(531            lib_exc.TimeoutException,532            waiters.wait_for_server_floating_ip, mock_client, fake_server,533            fake_fip)534    def test_wait_for_server_floating_ip_disassociate_timeout(self):535        mock_addresses = {'shared': [{'OS-EXT-IPS:type': 'floating',536                                      'addr': 'fake_address'}]}537        mock_server = {'server': {'id': 'fake_uuid',538                                  'addresses': mock_addresses}}539        mock_client = mock.Mock(540            spec=servers_client.ServersClient,541            build_timeout=1, build_interval=1,542            show_server=lambda id: mock_server)543        fake_server = {'id': 'fake-uuid'}544        fake_fip = {'floating_ip_address': 'fake_address'}545        self.assertRaises(546            lib_exc.TimeoutException,547            waiters.wait_for_server_floating_ip, mock_client, fake_server,...

test_images_client.py

Source:test_images_client.py

...239    def test_list_images_with_str_body(self):240        self._test_list_images()241    def test_list_images_with_bytes_body(self):242        self._test_list_images(bytes_body=True)243    def test_show_image_tasks(self):244        self.check_service_client_function(245            self.client.show_image_tasks,246            'tempest.lib.common.rest_client.RestClient.get',247            self.FAKE_SHOW_IMAGE_TASKS,248            True,...

images_client.py

Source:images_client.py

...97        resp, body = self.get(url)98        self.expected_success(200, resp.status)99        body = json.loads(body)100        return rest_client.ResponseBody(resp, body)101    def show_image_tasks(self, image_id):102        """Show image tasks."""103        url = 'images/%s/tasks' % image_id104        resp, body = self.get(url)105        self.expected_success(200, resp.status)106        body = json.loads(body)107        return rest_client.ResponseBody(resp, body)108    def is_resource_deleted(self, id):109        try:110            self.show_image(id)111        except lib_exc.NotFound:112            return True113        return False114    def is_resource_active(self, id):115        try:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.