Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use return_value method in refurb

Best Python code snippet using refurb_python

tests.py

Source:tests.py

1# Copyright 2012 Nebula, Inc.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 copy15import mock16import six17from django.conf import settings18from django.forms import widgets19from django.template.defaultfilters import slugify20from django.test.utils import override_settings21from django.urls import reverse22from django.utils.http import urlunquote23from openstack_dashboard import api24from openstack_dashboard.api import cinder25from openstack_dashboard.dashboards.project.volumes \26    import tables as volume_tables27from openstack_dashboard.test import helpers as test28from openstack_dashboard.usage import quotas29DETAIL_URL = ('horizon:project:volumes:detail')30INDEX_URL = reverse('horizon:project:volumes:index')31SEARCH_OPTS = dict(status=api.cinder.VOLUME_STATE_AVAILABLE)32ATTACHMENT_ID = '6061364b-6612-48a9-8fee-1a38fe072547'33class VolumeIndexViewTests(test.ResetImageAPIVersionMixin, test.TestCase):34    @test.create_mocks({35        api.nova: ['server_get', 'server_list'],36        api.cinder: ['volume_backup_supported',37                     'volume_snapshot_list',38                     'volume_list_paged',39                     'tenant_absolute_limits',40                     'group_list'],41    })42    def _test_index(self, with_attachments=False, with_groups=False):43        vol_snaps = self.cinder_volume_snapshots.list()44        volumes = self.cinder_volumes.list()45        if with_attachments:46            server = self.servers.first()47        else:48            for volume in volumes:49                volume.attachments = []50        self.mock_volume_backup_supported.return_value = False51        if with_groups:52            self.mock_group_list.return_value = self.cinder_groups.list()53            volumes = self.cinder_group_volumes.list()54        self.mock_volume_list_paged.return_value = [volumes, False, False]55        if with_attachments:56            self.mock_server_get.return_value = server57            self.mock_server_list.return_value = [self.servers.list(), False]58            self.mock_volume_snapshot_list.return_value = vol_snaps59        self.mock_tenant_absolute_limits.return_value = \60            self.cinder_limits['absolute']61        res = self.client.get(INDEX_URL)62        if with_attachments:63            self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),64                                                          search_opts=None)65            self.mock_volume_snapshot_list.assert_called_once()66        if with_groups:67            self.mock_group_list.assert_called_once_with(test.IsHttpRequest(),68                                                         search_opts=None)69        self.mock_volume_backup_supported.assert_called_with(70            test.IsHttpRequest())71        self.mock_volume_list_paged.assert_called_once_with(72            test.IsHttpRequest(), marker=None, search_opts=None,73            sort_dir='desc', paginate=True)74        self.mock_tenant_absolute_limits.assert_called_with(75            test.IsHttpRequest())76        self.assertEqual(res.status_code, 200)77        self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')78    def test_index_with_volume_attachments(self):79        self._test_index(True)80    def test_index_no_volume_attachments(self):81        self._test_index(False)82    def test_index_with_volume_groups(self):83        self._test_index(with_groups=True)84    @test.create_mocks({85        api.nova: ['server_get', 'server_list'],86        cinder: ['tenant_absolute_limits',87                 'volume_list_paged',88                 'volume_backup_supported',89                 'volume_snapshot_list'],90    })91    def _test_index_paginated(self, marker, sort_dir, volumes, url,92                              has_more, has_prev):93        backup_supported = True94        vol_snaps = self.cinder_volume_snapshots.list()95        server = self.servers.first()96        self.mock_volume_backup_supported.return_value = backup_supported97        self.mock_volume_list_paged.return_value = [volumes,98                                                    has_more, has_prev]99        self.mock_volume_snapshot_list.return_value = vol_snaps100        self.mock_server_list.return_value = [self.servers.list(), False]101        self.mock_server_get.return_value = server102        self.mock_tenant_absolute_limits.return_value = \103            self.cinder_limits['absolute']104        res = self.client.get(urlunquote(url))105        self.assertEqual(2, self.mock_volume_backup_supported.call_count)106        self.mock_volume_list_paged.assert_called_once_with(107            test.IsHttpRequest(), marker=marker, sort_dir=sort_dir,108            search_opts=None, paginate=True)109        self.mock_volume_snapshot_list.assert_called_once_with(110            test.IsHttpRequest(), search_opts=None)111        self.mock_tenant_absolute_limits.assert_called_with(112            test.IsHttpRequest())113        self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),114                                                      search_opts=None)115        self.assertEqual(res.status_code, 200)116        self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')117        return res118    def ensure_attachments_exist(self, volumes):119        volumes = copy.copy(volumes)120        for volume in volumes:121            if not volume.attachments:122                volume.attachments.append({123                    "id": "1", "server_id": '1', "device": "/dev/hda",124                    "attachment_id": ATTACHMENT_ID})125        return volumes126    @override_settings(API_RESULT_PAGE_SIZE=2)127    def test_index_paginated(self):128        volumes = self.ensure_attachments_exist(self.cinder_volumes.list())129        size = settings.API_RESULT_PAGE_SIZE130        # get first page131        expected_volumes = volumes[:size]132        url = INDEX_URL133        res = self._test_index_paginated(None, "desc", expected_volumes, url,134                                         True, False)135        result = res.context['volumes_table'].data136        self.assertItemsEqual(result, expected_volumes)137        # get second page138        expected_volumes = volumes[size:2 * size]139        marker = expected_volumes[0].id140        next = volume_tables.VolumesTable._meta.pagination_param141        url = "?".join([INDEX_URL, "=".join([next, marker])])142        res = self._test_index_paginated(marker, "desc", expected_volumes, url,143                                         True, True)144        result = res.context['volumes_table'].data145        self.assertItemsEqual(result, expected_volumes)146        # get last page147        expected_volumes = volumes[-size:]148        marker = expected_volumes[0].id149        next = volume_tables.VolumesTable._meta.pagination_param150        url = "?".join([INDEX_URL, "=".join([next, marker])])151        res = self._test_index_paginated(marker, "desc", expected_volumes, url,152                                         False, True)153        result = res.context['volumes_table'].data154        self.assertItemsEqual(result, expected_volumes)155    @override_settings(API_RESULT_PAGE_SIZE=2)156    def test_index_paginated_prev_page(self):157        volumes = self.ensure_attachments_exist(self.cinder_volumes.list())158        size = settings.API_RESULT_PAGE_SIZE159        # prev from some page160        expected_volumes = volumes[size:2 * size]161        marker = expected_volumes[0].id162        prev = volume_tables.VolumesTable._meta.prev_pagination_param163        url = "?".join([INDEX_URL, "=".join([prev, marker])])164        res = self._test_index_paginated(marker, "asc", expected_volumes, url,165                                         True, True)166        result = res.context['volumes_table'].data167        self.assertItemsEqual(result, expected_volumes)168        # back to first page169        expected_volumes = volumes[:size]170        marker = expected_volumes[0].id171        prev = volume_tables.VolumesTable._meta.prev_pagination_param172        url = "?".join([INDEX_URL, "=".join([prev, marker])])173        res = self._test_index_paginated(marker, "asc", expected_volumes, url,174                                         True, False)175        result = res.context['volumes_table'].data176        self.assertItemsEqual(result, expected_volumes)177class VolumeViewTests(test.ResetImageAPIVersionMixin, test.TestCase):178    def tearDown(self):179        for volume in self.cinder_volumes.list():180            # VolumeTableMixIn._set_volume_attributes mutates data181            # and cinder_volumes.list() doesn't deep copy182            for att in volume.attachments:183                if 'instance' in att:184                    del att['instance']185        super(VolumeViewTests, self).tearDown()186    @test.create_mocks({187        cinder: ['volume_create', 'volume_snapshot_list',188                 'volume_type_list', 'volume_type_default',189                 'volume_list', 'availability_zone_list',190                 'extension_supported', 'group_list'],191        quotas: ['tenant_quota_usages'],192        api.glance: ['image_list_detailed'],193    })194    def test_create_volume(self):195        volume = self.cinder_volumes.first()196        volume_type = self.cinder_volume_types.first()197        az = self.cinder_availability_zones.first().zoneName198        formData = {'name': u'A Volume I Am Making',199                    'description': u'This is a volume I am making for a test.',200                    'method': u'CreateForm',201                    'type': volume_type.name,202                    'size': 50,203                    'snapshot_source': '',204                    'availability_zone': az}205        self.mock_volume_type_default.return_value = \206            self.cinder_volume_types.first()207        self.mock_volume_type_list.return_value = \208            self.cinder_volume_types.list()209        self.mock_tenant_quota_usages.return_value = \210            self.cinder_quota_usages.first()211        self.mock_volume_snapshot_list.return_value = \212            self.cinder_volume_snapshots.list()213        self.mock_image_list_detailed.return_value = [[], False, False]214        self.mock_availability_zone_list.return_value = \215            self.cinder_availability_zones.list()216        self.mock_extension_supported.return_value = True217        self.mock_volume_list.return_value = self.cinder_volumes.list()218        self.mock_volume_create.return_value = volume219        self.mock_group_list.return_value = []220        url = reverse('horizon:project:volumes:create')221        res = self.client.post(url, formData)222        self.assertNoFormErrors(res)223        redirect_url = INDEX_URL224        self.assertRedirectsNoFollow(res, redirect_url)225        self.mock_volume_type_default.assert_called_once()226        self.mock_volume_type_list.assert_called_once()227        self.mock_volume_snapshot_list.assert_called_once_with(228            test.IsHttpRequest(), search_opts=SEARCH_OPTS)229        self.mock_availability_zone_list.assert_called_once()230        self.mock_extension_supported.assert_called_once_with(231            test.IsHttpRequest(), 'AvailabilityZones')232        self.mock_volume_list.assert_called_once_with(test.IsHttpRequest(),233                                                      search_opts=SEARCH_OPTS)234        self.mock_volume_create.assert_called_once_with(235            test.IsHttpRequest(), formData['size'], formData['name'],236            formData['description'], formData['type'], metadata={},237            snapshot_id=None, group_id=None, image_id=None,238            availability_zone=formData['availability_zone'], source_volid=None)239        self.mock_image_list_detailed.assert_called_with(240            test.IsHttpRequest(),241            filters={'visibility': 'shared', 'status': 'active'})242        self.mock_tenant_quota_usages.assert_called_once_with(243            test.IsHttpRequest(),244            targets=('volumes', 'gigabytes'))245        self.mock_group_list.assert_called_with(test.IsHttpRequest())246    @test.create_mocks({247        quotas: ['tenant_quota_usages'],248        api.glance: ['image_list_detailed'],249        cinder: ['extension_supported',250                 'availability_zone_list',251                 'volume_list',252                 'volume_type_default',253                 'volume_type_list',254                 'volume_snapshot_list',255                 'volume_create',256                 'group_list'],257    })258    def test_create_volume_without_name(self):259        volume = self.cinder_volumes.first()260        volume_type = self.cinder_volume_types.first()261        az = self.cinder_availability_zones.first().zoneName262        formData = {'name': '',263                    'description': u'This is a volume I am making for a test.',264                    'method': u'CreateForm',265                    'type': volume_type.name,266                    'size': 50,267                    'snapshot_source': '',268                    'availability_zone': az}269        self.mock_volume_type_list.return_value = \270            self.cinder_volume_types.list()271        self.mock_tenant_quota_usages.return_value = \272            self.cinder_quota_usages.first()273        self.mock_volume_snapshot_list.return_value = \274            self.cinder_volume_snapshots.list()275        self.mock_image_list_detailed.return_value = [self.images.list(),276                                                      False, False]277        self.mock_availability_zone_list.return_value = \278            self.cinder_availability_zones.list()279        self.mock_extension_supported.return_value = True280        self.mock_volume_type_default.return_value = \281            self.cinder_volume_types.first()282        self.mock_volume_list.return_value = self.cinder_volumes.list()283        self.mock_volume_create.return_value = volume284        self.mock_group_list.return_value = []285        url = reverse('horizon:project:volumes:create')286        res = self.client.post(url, formData)287        redirect_url = INDEX_URL288        self.assertRedirectsNoFollow(res, redirect_url)289        self.mock_volume_type_list.assert_called_once()290        self.mock_volume_snapshot_list.assert_called_once_with(291            test.IsHttpRequest(), search_opts=SEARCH_OPTS)292        self.mock_image_list_detailed.assert_called_with(293            test.IsHttpRequest(),294            filters={'visibility': 'shared', 'status': 'active'})295        self.mock_availability_zone_list.assert_called_once()296        self.mock_extension_supported.assert_called_once_with(297            test.IsHttpRequest(), 'AvailabilityZones')298        self.mock_volume_type_default.assert_called_once()299        self.mock_volume_list.assert_called_once()300        self.mock_volume_create.assert_called_once_with(301            test.IsHttpRequest(), formData['size'], formData['name'],302            formData['description'], formData['type'], metadata={},303            snapshot_id=None, group_id=None, image_id=None,304            availability_zone=formData['availability_zone'], source_volid=None)305        self.mock_group_list.assert_called_once_with(test.IsHttpRequest())306    @test.create_mocks({307        quotas: ['tenant_quota_usages'],308        api.glance: ['image_list_detailed'],309        cinder: ['extension_supported',310                 'availability_zone_list',311                 'volume_list',312                 'volume_type_default',313                 'volume_type_list',314                 'volume_snapshot_list',315                 'volume_create',316                 'group_list'],317    })318    def test_create_volume_dropdown(self):319        volume = self.cinder_volumes.first()320        formData = {'name': u'A Volume I Am Making',321                    'description': u'This is a volume I am making for a test.',322                    'method': u'CreateForm',323                    'size': 50,324                    'type': '',325                    'volume_source_type': 'no_source_type',326                    'snapshot_source': self.cinder_volume_snapshots.first().id,327                    'image_source': self.images.first().id}328        self.mock_volume_type_default.return_value = \329            self.cinder_volume_types.first()330        self.mock_volume_type_list.return_value = \331            self.cinder_volume_types.list()332        self.mock_volume_snapshot_list.return_value = \333            self.cinder_volume_snapshots.list()334        self.mock_image_list_detailed.return_value = \335            [self.images.list(), False, False]336        self.mock_volume_list.return_value = self.cinder_volumes.list()337        self.mock_tenant_quota_usages.return_value = \338            self.cinder_quota_usages.first()339        self.mock_extension_supported.return_value = True340        self.mock_availability_zone_list.return_value = \341            self.cinder_availability_zones.list()342        self.mock_group_list.return_value = []343        self.mock_volume_create.return_value = volume344        url = reverse('horizon:project:volumes:create')345        res = self.client.post(url, formData)346        redirect_url = INDEX_URL347        self.assertRedirectsNoFollow(res, redirect_url)348        self.mock_volume_type_default.assert_called_once()349        self.mock_volume_type_list.assert_called_once()350        self.mock_volume_snapshot_list.assert_called_once_with(351            test.IsHttpRequest(), search_opts=SEARCH_OPTS)352        self.mock_image_list_detailed.assert_called_with(353            test.IsHttpRequest(),354            filters={'visibility': 'shared', 'status': 'active'})355        self.mock_volume_list.assert_called_once_with(test.IsHttpRequest(),356                                                      search_opts=SEARCH_OPTS)357        self.mock_tenant_quota_usages.assert_called_once()358        self.mock_extension_supported.assert_called_once_with(359            test.IsHttpRequest(), 'AvailabilityZones')360        self.mock_availability_zone_list.assert_called_once()361        self.mock_volume_create.assert_called_once_with(362            test.IsHttpRequest(), formData['size'], formData['name'],363            formData['description'], '', metadata={}, snapshot_id=None,364            group_id=None, image_id=None, availability_zone=None,365            source_volid=None)366        self.mock_group_list.assert_called_with(test.IsHttpRequest())367    @test.create_mocks({368        quotas: ['tenant_quota_usages'],369        cinder: ['volume_type_list',370                 'volume_type_default',371                 'volume_get',372                 'volume_snapshot_get',373                 'volume_create',374                 'group_list'],375    })376    def test_create_volume_from_snapshot(self):377        volume = self.cinder_volumes.first()378        snapshot = self.cinder_volume_snapshots.first()379        formData = {'name': u'A Volume I Am Making',380                    'description': u'This is a volume I am making for a test.',381                    'method': u'CreateForm',382                    'size': 50,383                    'type': '',384                    'snapshot_source': snapshot.id}385        self.mock_volume_type_default.return_value = \386            self.cinder_volume_types.first()387        self.mock_volume_type_list.return_value = \388            self.cinder_volume_types.list()389        self.mock_tenant_quota_usages.return_value = \390            self.cinder_quota_usages.first()391        self.mock_volume_snapshot_get.return_value = snapshot392        self.mock_volume_get.return_value = self.cinder_volumes.first()393        self.mock_volume_create.return_value = volume394        self.mock_group_list.return_value = []395        # get snapshot from url396        url = reverse('horizon:project:volumes:create')397        res = self.client.post("?".join([url,398                                         "snapshot_id=" + str(snapshot.id)]),399                               formData)400        redirect_url = INDEX_URL401        self.assertRedirectsNoFollow(res, redirect_url)402        self.mock_volume_type_default.assert_called_once()403        self.mock_volume_type_list.assert_called_once()404        self.mock_tenant_quota_usages.assert_called_once()405        self.mock_volume_snapshot_get.assert_called_once_with(406            test.IsHttpRequest(), str(snapshot.id))407        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),408                                                     snapshot.volume_id)409        self.mock_volume_create.assert_called_once_with(410            test.IsHttpRequest(), formData['size'], formData['name'],411            formData['description'], '', metadata={}, snapshot_id=snapshot.id,412            group_id=None, image_id=None, availability_zone=None,413            source_volid=None)414        self.mock_group_list.assert_called_once_with(test.IsHttpRequest())415    @test.create_mocks({416        quotas: ['tenant_quota_usages'],417        api.glance: ['image_list_detailed'],418        cinder: ['extension_supported',419                 'volume_snapshot_list',420                 'volume_snapshot_get',421                 'availability_zone_list',422                 'volume_type_list',423                 'volume_list',424                 'volume_type_default',425                 'volume_get',426                 'volume_create',427                 'group_list'],428    })429    def test_create_volume_from_volume(self):430        volume = self.cinder_volumes.first()431        formData = {'name': u'A copy of a volume',432                    'description': u'This is a volume I am making for a test.',433                    'method': u'CreateForm',434                    'size': 50,435                    'type': '',436                    'volume_source_type': 'volume_source',437                    'volume_source': volume.id}438        self.mock_volume_type_default.return_value = \439            self.cinder_volume_types.first()440        self.mock_volume_list.return_value = self.cinder_volumes.list()441        self.mock_volume_type_list.return_value = \442            self.cinder_volume_types.list()443        self.mock_volume_snapshot_list.return_value = \444            self.cinder_volume_snapshots.list()445        self.mock_tenant_quota_usages.return_value = \446            self.cinder_quota_usages.first()447        self.mock_volume_get.return_value = self.cinder_volumes.first()448        self.mock_extension_supported.return_value = True449        self.mock_availability_zone_list.return_value = \450            self.cinder_availability_zones.list()451        self.mock_image_list_detailed.return_value = \452            [self.images.list(), False, False]453        self.mock_volume_create.return_value = volume454        self.mock_group_list.return_value = []455        url = reverse('horizon:project:volumes:create')456        redirect_url = INDEX_URL457        res = self.client.post(url, formData)458        self.assertNoFormErrors(res)459        self.assertMessageCount(info=1)460        self.assertRedirectsNoFollow(res, redirect_url)461        self.mock_volume_type_default.assert_called_once()462        self.mock_volume_list.assert_called_once_with(test.IsHttpRequest(),463                                                      search_opts=SEARCH_OPTS)464        self.mock_volume_type_list.assert_called_once()465        self.mock_volume_snapshot_list.assert_called_once_with(466            test.IsHttpRequest(), search_opts=SEARCH_OPTS)467        self.mock_tenant_quota_usages.assert_called_once()468        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),469                                                     volume.id)470        self.mock_extension_supported.assert_called_once_with(471            test.IsHttpRequest(), 'AvailabilityZones')472        self.mock_availability_zone_list.assert_called_once()473        self.mock_image_list_detailed.assert_called_with(474            test.IsHttpRequest(),475            filters={'visibility': 'shared', 'status': 'active'})476        self.mock_volume_create.assert_called_once_with(477            test.IsHttpRequest(), formData['size'], formData['name'],478            formData['description'], None, metadata={}, snapshot_id=None,479            group_id=None, image_id=None, availability_zone=None,480            source_volid=volume.id)481        self.mock_group_list.assert_called_once_with(test.IsHttpRequest())482    @test.create_mocks({483        quotas: ['tenant_quota_usages'],484        api.glance: ['image_list_detailed'],485        cinder: ['extension_supported',486                 'availability_zone_list',487                 'volume_type_list',488                 'volume_list',489                 'volume_type_default',490                 'volume_get',491                 'volume_snapshot_get',492                 'volume_snapshot_list',493                 'volume_create',494                 'group_list'],495    })496    def test_create_volume_from_snapshot_dropdown(self):497        volume = self.cinder_volumes.first()498        snapshot = self.cinder_volume_snapshots.first()499        formData = {'name': u'A Volume I Am Making',500                    'description': u'This is a volume I am making for a test.',501                    'method': u'CreateForm',502                    'size': 50,503                    'type': '',504                    'volume_source_type': 'snapshot_source',505                    'snapshot_source': snapshot.id}506        self.mock_volume_type_list.return_value = \507            self.cinder_volume_types.list()508        self.mock_volume_snapshot_list.return_value = \509            self.cinder_volume_snapshots.list()510        self.mock_image_list_detailed.return_value = [self.images.list(),511                                                      False, False]512        self.mock_volume_type_default.return_value = \513            self.cinder_volume_types.first()514        self.mock_volume_list.return_value = self.cinder_volumes.list()515        self.mock_tenant_quota_usages.return_value = \516            self.cinder_quota_usages.first()517        self.mock_volume_snapshot_get.return_value = snapshot518        self.mock_extension_supported.return_value = True519        self.mock_availability_zone_list.return_value = \520            self.cinder_availability_zones.list()521        self.mock_volume_create.return_value = volume522        self.mock_group_list.return_value = []523        # get snapshot from dropdown list524        url = reverse('horizon:project:volumes:create')525        res = self.client.post(url, formData)526        redirect_url = INDEX_URL527        self.assertRedirectsNoFollow(res, redirect_url)528        self.mock_volume_type_list.assert_called_once()529        self.mock_volume_snapshot_list.assert_called_once_with(530            test.IsHttpRequest(), search_opts=SEARCH_OPTS)531        self.mock_image_list_detailed.assert_called_with(532            test.IsHttpRequest(),533            filters={'visibility': 'shared', 'status': 'active'})534        self.mock_volume_type_default.assert_called_once()535        self.mock_volume_list.assert_called_once_with(test.IsHttpRequest(),536                                                      search_opts=SEARCH_OPTS)537        self.mock_tenant_quota_usages.assert_called_once()538        self.mock_volume_snapshot_get.assert_called_once_with(539            test.IsHttpRequest(), str(snapshot.id))540        self.mock_extension_supported.assert_called_once_with(541            test.IsHttpRequest(), 'AvailabilityZones')542        self.mock_availability_zone_list.assert_called_once()543        self.mock_volume_create.assert_called_once_with(544            test.IsHttpRequest(), formData['size'], formData['name'],545            formData['description'], '', metadata={}, snapshot_id=snapshot.id,546            group_id=None, image_id=None, availability_zone=None,547            source_volid=None)548        self.mock_group_list.assert_called_once_with(test.IsHttpRequest())549    @test.create_mocks({550        quotas: ['tenant_quota_usages'],551        api.glance: ['image_list_detailed'],552        cinder: ['volume_snapshot_get',553                 'volume_type_list',554                 'volume_type_default',555                 'volume_get',556                 'group_list'],557    })558    def test_create_volume_from_snapshot_invalid_size(self):559        snapshot = self.cinder_volume_snapshots.first()560        formData = {'name': u'A Volume I Am Making',561                    'description': u'This is a volume I am making for a test.',562                    'method': u'CreateForm',563                    'size': 20, 'snapshot_source': snapshot.id}564        self.mock_volume_type_list.return_value = \565            self.cinder_volume_types.list()566        self.mock_volume_type_default.return_value = \567            self.cinder_volume_types.first()568        self.mock_tenant_quota_usages.return_value = \569            self.cinder_quota_usages.first()570        self.mock_volume_snapshot_get.return_value = snapshot571        self.mock_volume_get.return_value = self.cinder_volumes.first()572        self.mock_group_list.return_value = []573        url = reverse('horizon:project:volumes:create')574        res = self.client.post("?".join([url,575                                         "snapshot_id=" + str(snapshot.id)]),576                               formData, follow=True)577        self.assertEqual(res.redirect_chain, [])578        self.assertFormError(res, 'form', None,579                             "The volume size cannot be less than the "580                             "snapshot size (40GiB)")581        self.assertEqual(3, self.mock_volume_type_list.call_count)582        self.assertEqual(2, self.mock_volume_type_default.call_count)583        self.mock_volume_snapshot_get.assert_called_with(test.IsHttpRequest(),584                                                         str(snapshot.id))585        self.mock_volume_get.assert_called_with(test.IsHttpRequest(),586                                                snapshot.volume_id)587        self.mock_group_list.assert_called_with(test.IsHttpRequest())588    @test.create_mocks({589        quotas: ['tenant_quota_usages'],590        api.glance: ['image_get'],591        cinder: ['extension_supported',592                 'availability_zone_list',593                 'volume_type_default',594                 'volume_type_list',595                 'volume_create',596                 'group_list'],597    })598    def test_create_volume_from_image(self):599        volume = self.cinder_volumes.first()600        image = self.images.first()601        formData = {'name': u'A Volume I Am Making',602                    'description': u'This is a volume I am making for a test.',603                    'method': u'CreateForm',604                    'size': 40,605                    'type': '',606                    'image_source': image.id}607        self.mock_volume_type_default.return_value = \608            self.cinder_volume_types.first()609        self.mock_volume_type_list.ret = self.cinder_volume_types.list()610        self.mock_tenant_quota_usages.return_value = \611            self.cinder_quota_usages.first()612        self.mock_image_get.return_value = image613        self.mock_extension_supported.return_value = True614        self.mock_availability_zone_list.return_value = \615            self.cinder_availability_zones.list()616        self.mock_group_list.return_value = []617        self.mock_volume_create.return_value = volume618        # get image from url619        url = reverse('horizon:project:volumes:create')620        res = self.client.post("?".join([url,621                                         "image_id=" + str(image.id)]),622                               formData)623        redirect_url = INDEX_URL624        self.assertRedirectsNoFollow(res, redirect_url)625        self.mock_volume_type_default.assert_called_once()626        self.mock_volume_type_list.assert_called_once()627        self.mock_tenant_quota_usages.assert_called_once()628        self.mock_image_get.assert_called_once_with(test.IsHttpRequest(),629                                                    str(image.id))630        self.mock_extension_supported.assert_called_once_with(631            test.IsHttpRequest(), 'AvailabilityZones')632        self.mock_availability_zone_list.assert_called_once()633        self.mock_volume_create.assert_called_once_with(634            test.IsHttpRequest(), formData['size'], formData['name'],635            formData['description'], '', metadata={}, snapshot_id=None,636            group_id=None, image_id=image.id, availability_zone=None,637            source_volid=None)638        self.mock_group_list.assert_called_with(test.IsHttpRequest())639    @test.create_mocks({640        quotas: ['tenant_quota_usages'],641        api.glance: ['image_list_detailed',642                     'image_get'],643        cinder: ['extension_supported',644                 'availability_zone_list',645                 'volume_snapshot_list',646                 'volume_list',647                 'volume_type_list',648                 'volume_type_default',649                 'volume_create',650                 'group_list'],651    })652    def test_create_volume_from_image_dropdown(self):653        volume = self.cinder_volumes.first()654        image = self.images.first()655        formData = {'name': u'A Volume I Am Making',656                    'description': u'This is a volume I am making for a test.',657                    'method': u'CreateForm',658                    'size': 30,659                    'type': '',660                    'volume_source_type': 'image_source',661                    'snapshot_source': self.cinder_volume_snapshots.first().id,662                    'image_source': image.id}663        self.mock_volume_type_list.return_value = \664            self.cinder_volume_types.list()665        self.mock_volume_snapshot_list.return_value = \666            self.cinder_volume_snapshots.list()667        self.mock_image_list_detailed.return_value = [self.images.list(),668                                                      False, False]669        self.mock_volume_type_default.return_value = \670            self.cinder_volume_types.first()671        self.mock_volume_list.return_value = self.cinder_volumes.list()672        self.mock_tenant_quota_usages.return_value = \673            self.cinder_quota_usages.first()674        self.mock_image_get.return_value = image675        self.mock_extension_supported.return_value = True676        self.mock_availability_zone_list.return_value = \677            self.cinder_availability_zones.list()678        self.mock_group_list.return_value = []679        self.mock_volume_create.return_value = volume680        # get image from dropdown list681        url = reverse('horizon:project:volumes:create')682        res = self.client.post(url, formData)683        redirect_url = INDEX_URL684        self.assertRedirectsNoFollow(res, redirect_url)685        self.mock_volume_type_list.assert_called_once()686        self.mock_volume_snapshot_list.assert_called_once_with(687            test.IsHttpRequest(), search_opts=SEARCH_OPTS)688        self.mock_image_list_detailed.assert_called_with(689            test.IsHttpRequest(),690            filters={'visibility': 'shared', 'status': 'active'})691        self.mock_volume_type_default.assert_called_once()692        self.mock_volume_list.assert_called_once_with(test.IsHttpRequest(),693                                                      search_opts=SEARCH_OPTS)694        self.mock_tenant_quota_usages.assert_called_once()695        self.mock_image_get.assert_called_with(test.IsHttpRequest(),696                                               str(image.id))697        self.mock_extension_supported.assert_called_once_with(698            test.IsHttpRequest(), 'AvailabilityZones')699        self.mock_availability_zone_list.assert_called_once()700        self.mock_volume_create.assert_called_once_with(701            test.IsHttpRequest(), formData['size'], formData['name'],702            formData['description'], '', metadata={}, snapshot_id=None,703            group_id=None, image_id=image.id, availability_zone=None,704            source_volid=None)705        self.mock_group_list.assert_called_with(test.IsHttpRequest())706    @test.create_mocks({707        quotas: ['tenant_quota_usages'],708        api.glance: ['image_get'],709        cinder: ['extension_supported',710                 'availability_zone_list',711                 'volume_type_list',712                 'volume_type_default',713                 'group_list'],714    })715    def test_create_volume_from_image_under_image_size(self):716        image = self.images.first()717        formData = {'name': u'A Volume I Am Making',718                    'description': u'This is a volume I am making for a test.',719                    'method': u'CreateForm',720                    'size': 1, 'image_source': image.id}721        self.mock_volume_type_list.return_value = \722            self.cinder_volume_types.list()723        self.mock_volume_type_default.return_value = \724            self.cinder_volume_types.first()725        self.mock_tenant_quota_usages.return_value = \726            self.cinder_quota_usages.first()727        self.mock_image_get.return_value = image728        self.mock_extension_supported.return_value = True729        self.mock_group_list.return_value = []730        url = reverse('horizon:project:volumes:create')731        res = self.client.post("?".join([url,732                                         "image_id=" + str(image.id)]),733                               formData, follow=True)734        self.assertEqual(res.redirect_chain, [])735        msg = (u"The volume size cannot be less than the "736               u"image size (20.0\xa0GB)")737        self.assertFormError(res, 'form', None, msg)738        self.assertEqual(3, self.mock_volume_type_list.call_count)739        self.assertEqual(2, self.mock_volume_type_default.call_count)740        self.assertEqual(2, self.mock_tenant_quota_usages.call_count)741        self.mock_image_get.assert_called_with(test.IsHttpRequest(),742                                               str(image.id))743        self.mock_extension_supported.assert_called_with(test.IsHttpRequest(),744                                                         'AvailabilityZones')745        self.mock_group_list.assert_called_with(test.IsHttpRequest())746    @test.create_mocks({747        quotas: ['tenant_quota_usages'],748        api.glance: ['image_get'],749        cinder: ['extension_supported',750                 'availability_zone_list',751                 'volume_type_list',752                 'volume_type_default',753                 'group_list'],754    })755    def _test_create_volume_from_image_under_image_min_disk_size(self, image):756        formData = {'name': u'A Volume I Am Making',757                    'description': u'This is a volume I am making for a test.',758                    'method': u'CreateForm',759                    'size': 5, 'image_source': image.id}760        self.mock_volume_type_list.return_value = \761            self.cinder_volume_types.list()762        self.mock_volume_type_default.return_value = \763            self.cinder_volume_types.first()764        self.mock_tenant_quota_usages.return_value = \765            self.cinder_quota_usages.first()766        self.mock_image_get.return_value = image767        self.mock_extension_supported.return_value = True768        self.mock_availability_zone_list.return_value = \769            self.cinder_availability_zones.list()770        self.mock_group_list.return_value = []771        url = reverse('horizon:project:volumes:create')772        res = self.client.post("?".join([url,773                                         "image_id=" + str(image.id)]),774                               formData, follow=True)775        self.assertEqual(res.redirect_chain, [])776        self.assertFormError(res, 'form', None,777                             "The volume size cannot be less than the "778                             "image minimum disk size (30GiB)")779        self.assertEqual(3, self.mock_volume_type_list.call_count)780        self.assertEqual(2, self.mock_volume_type_default.call_count)781        self.assertEqual(2, self.mock_availability_zone_list.call_count)782        self.mock_image_get.assert_called_with(test.IsHttpRequest(),783                                               str(image.id))784        self.mock_extension_supported.assert_called_with(test.IsHttpRequest(),785                                                         'AvailabilityZones')786        self.mock_group_list.assert_called_with(test.IsHttpRequest())787    def test_create_volume_from_image_under_image_min_disk_size(self):788        image = self.images.get(name="protected_images")789        image.min_disk = 30790        self._test_create_volume_from_image_under_image_min_disk_size(image)791    def test_create_volume_from_image_under_image_prop_min_disk_size_v2(self):792        image = self.imagesV2.get(name="protected_images")793        self._test_create_volume_from_image_under_image_min_disk_size(image)794    @test.create_mocks({795        quotas: ['tenant_quota_usages'],796        api.glance: ['image_list_detailed'],797        cinder: ['extension_supported',798                 'availability_zone_list',799                 'volume_list',800                 'volume_type_list',801                 'volume_type_default',802                 'volume_snapshot_list',803                 'group_list'],804    })805    def test_create_volume_gb_used_over_alloted_quota(self):806        formData = {'name': u'This Volume Is Huge!',807                    'description': u'This is a volume that is just too big!',808                    'method': u'CreateForm',809                    'size': 5000}810        usage_limit = self.cinder_quota_usages.first()811        usage_limit.add_quota(api.base.Quota('volumes', 6))812        usage_limit.tally('volumes', len(self.cinder_volumes.list()))813        usage_limit.add_quota(api.base.Quota('gigabytes', 100))814        usage_limit.tally('gigabytes', 80)815        self.mock_volume_type_list.return_value = \816            self.cinder_volume_types.list()817        self.mock_volume_type_default.return_value = \818            self.cinder_volume_types.first()819        self.mock_tenant_quota_usages.return_value = usage_limit820        self.mock_volume_snapshot_list.return_value = \821            self.cinder_volume_snapshots.list()822        self.mock_image_list_detailed.return_value = [self.images.list(),823                                                      False, False]824        self.mock_volume_list.return_value = self.cinder_volumes.list()825        self.mock_extension_supported.return_value = True826        self.mock_availability_zone_list.return_value = \827            self.cinder_availability_zones.list()828        self.mock_group_list.return_value = []829        url = reverse('horizon:project:volumes:create')830        res = self.client.post(url, formData)831        expected_error = [u'A volume of 5000GiB cannot be created as you only'832                          ' have 20GiB of your quota available.']833        self.assertEqual(res.context['form'].errors['__all__'], expected_error)834        self.assertEqual(3, self.mock_volume_type_list.call_count)835        self.assertEqual(2, self.mock_volume_type_default.call_count)836        self.assertEqual(2, self.mock_volume_list.call_count)837        self.assertEqual(2, self.mock_availability_zone_list.call_count)838        self.assertEqual(2, self.mock_tenant_quota_usages.call_count)839        self.mock_volume_snapshot_list.assert_called_with(840            test.IsHttpRequest(), search_opts=SEARCH_OPTS)841        self.mock_image_list_detailed.assert_called_with(842            test.IsHttpRequest(),843            filters={'visibility': 'shared', 'status': 'active'})844        self.mock_extension_supported.assert_called_with(test.IsHttpRequest(),845                                                         'AvailabilityZones')846        self.mock_group_list.assert_called_with(test.IsHttpRequest())847    @test.create_mocks({848        quotas: ['tenant_quota_usages'],849        api.glance: ['image_list_detailed'],850        cinder: ['extension_supported',851                 'availability_zone_list',852                 'volume_list',853                 'volume_type_list',854                 'volume_type_default',855                 'volume_snapshot_list',856                 'group_list'],857    })858    def test_create_volume_number_over_alloted_quota(self):859        formData = {'name': u'Too Many...',860                    'description': u'We have no volumes left!',861                    'method': u'CreateForm',862                    'size': 10}863        usage_limit = self.cinder_quota_usages.first()864        usage_limit.add_quota(api.base.Quota('volumes',865                                             len(self.cinder_volumes.list())))866        usage_limit.tally('volumes', len(self.cinder_volumes.list()))867        usage_limit.add_quota(api.base.Quota('gigabytes', 100))868        usage_limit.tally('gigabytes', 20)869        self.mock_volume_type_list.return_value = \870            self.cinder_volume_types.list()871        self.mock_volume_type_default.return_value = \872            self.cinder_volume_types.first()873        self.mock_tenant_quota_usages.return_value = usage_limit874        self.mock_volume_snapshot_list.return_value = \875            self.cinder_volume_snapshots.list()876        self.mock_image_list_detailed.return_value = [self.images.list(),877                                                      False, False]878        self.mock_volume_list.return_value = self.cinder_volumes.list()879        self.mock_extension_supported.return_value = True880        self.mock_availability_zone_list.return_value = \881            self.cinder_availability_zones.list()882        self.mock_group_list.return_value = []883        url = reverse('horizon:project:volumes:create')884        res = self.client.post(url, formData)885        expected_error = [u'You are already using all of your available'886                          ' volumes.']887        self.assertEqual(res.context['form'].errors['__all__'], expected_error)888        self.assertEqual(3, self.mock_volume_type_list.call_count)889        self.assertEqual(2, self.mock_volume_type_default.call_count)890        self.assertEqual(2, self.mock_availability_zone_list.call_count)891        self.mock_volume_snapshot_list.assert_called_with(892            test.IsHttpRequest(), search_opts=SEARCH_OPTS)893        self.mock_image_list_detailed.assert_called_with(894            test.IsHttpRequest(),895            filters={'visibility': 'shared', 'status': 'active'})896        self.mock_volume_list.assert_called_with(test.IsHttpRequest(),897                                                 search_opts=SEARCH_OPTS)898        self.mock_extension_supported.assert_called_with(test.IsHttpRequest(),899                                                         'AvailabilityZones')900        self.mock_group_list.assert_called_with(test.IsHttpRequest())901    @test.create_mocks({902        cinder: ['volume_create', 'volume_snapshot_list',903                 'volume_type_list', 'volume_type_default',904                 'volume_list', 'availability_zone_list',905                 'extension_supported', 'group_list'],906        quotas: ['tenant_quota_usages'],907        api.glance: ['image_list_detailed'],908    })909    def test_create_volume_with_group(self):910        volume = self.cinder_volumes.first()911        volume_type = self.cinder_volume_types.first()912        az = self.cinder_availability_zones.first().zoneName913        volume_group = self.cinder_groups.list()[0]914        formData = {'name': u'A Volume I Am Making',915                    'description': u'This is a volume I am making for a test.',916                    'method': u'CreateForm',917                    'type': volume_type.name,918                    'size': 50,919                    'snapshot_source': '',920                    'availability_zone': az,921                    'group': volume_group.id}922        self.mock_volume_type_default.return_value = \923            self.cinder_volume_types.first()924        self.mock_volume_type_list.return_value = \925            self.cinder_volume_types.list()926        self.mock_tenant_quota_usages.return_value = \927            self.cinder_quota_usages.first()928        self.mock_volume_snapshot_list.return_value = \929            self.cinder_volume_snapshots.list()930        self.mock_image_list_detailed.return_value = [[], False, False]931        self.mock_availability_zone_list.return_value = \932            self.cinder_availability_zones.list()933        self.mock_extension_supported.return_value = True934        self.mock_volume_list.return_value = self.cinder_volumes.list()935        self.mock_volume_create.return_value = volume936        self.mock_group_list.return_value = self.cinder_groups.list()937        url = reverse('horizon:project:volumes:create')938        res = self.client.post(url, formData)939        self.assertNoFormErrors(res)940        redirect_url = INDEX_URL941        self.assertRedirectsNoFollow(res, redirect_url)942        self.mock_volume_type_default.assert_called_once()943        self.mock_volume_type_list.assert_called_once()944        self.mock_volume_snapshot_list.assert_called_once_with(945            test.IsHttpRequest(), search_opts=SEARCH_OPTS)946        self.mock_availability_zone_list.assert_called_once()947        self.mock_extension_supported.assert_called_once_with(948            test.IsHttpRequest(), 'AvailabilityZones')949        self.mock_volume_list.assert_called_once_with(test.IsHttpRequest(),950                                                      search_opts=SEARCH_OPTS)951        self.mock_volume_create.assert_called_once_with(952            test.IsHttpRequest(), formData['size'], formData['name'],953            formData['description'], formData['type'], metadata={},954            snapshot_id=None, group_id=volume_group.id, image_id=None,955            availability_zone=formData['availability_zone'], source_volid=None)956        self.mock_image_list_detailed.assert_called_with(957            test.IsHttpRequest(),958            filters={'visibility': 'shared', 'status': 'active'})959        self.mock_tenant_quota_usages.assert_called_once_with(960            test.IsHttpRequest(),961            targets=('volumes', 'gigabytes'))962        self.mock_group_list.assert_called_with(test.IsHttpRequest())963    @test.create_mocks({964        api.nova: ['server_list'],965        cinder: ['volume_delete',966                 'volume_snapshot_list',967                 'volume_list_paged',968                 'tenant_absolute_limits'],969    })970    def test_delete_volume(self):971        volumes = self.cinder_volumes.list()972        volume = self.cinder_volumes.first()973        formData = {'action':974                    'volumes__delete__%s' % volume.id}975        self.mock_volume_list_paged.return_value = [volumes, False, False]976        self.mock_volume_snapshot_list.return_value = []977        self.mock_server_list.return_value = [self.servers.list(), False]978        self.mock_volume_list_paged.return_value = [volumes, False, False]979        self.mock_tenant_absolute_limits.return_value = \980            self.cinder_limits['absolute']981        url = INDEX_URL982        res = self.client.post(url, formData, follow=True)983        self.assertIn("Scheduled deletion of Volume: Volume name",984                      [m.message for m in res.context['messages']])985        self.mock_volume_list_paged.assert_called_with(986            test.IsHttpRequest(), marker=None,987            paginate=True, sort_dir='desc',988            search_opts=None)989        self.assertEqual(2, self.mock_volume_snapshot_list.call_count)990        self.mock_volume_delete.assert_called_once_with(test.IsHttpRequest(),991                                                        volume.id)992        self.mock_server_list.assert_called_with(test.IsHttpRequest(),993                                                 search_opts=None)994        self.assertEqual(8, self.mock_tenant_absolute_limits.call_count)995    @mock.patch.object(quotas, 'tenant_quota_usages')996    @mock.patch.object(cinder, 'tenant_absolute_limits')997    @mock.patch.object(cinder, 'volume_get')998    def test_delete_volume_with_snap_no_action_item(self, mock_get,999                                                    mock_limits,1000                                                    mock_quotas):1001        volume = self.cinder_volumes.get(name='Volume name')1002        setattr(volume, 'has_snapshot', True)1003        limits = self.cinder_limits['absolute']1004        mock_get.return_value = volume1005        mock_limits.return_value = limits1006        mock_quotas.return_value = self.cinder_quota_usages.first()1007        url = (INDEX_URL +1008               "?action=row_update&table=volumes&obj_id=" + volume.id)1009        res = self.client.get(url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')1010        self.assertEqual(res.status_code, 200)1011        mock_quotas.assert_called_once_with(test.IsHttpRequest(),1012                                            targets=('volumes', 'gigabytes'))1013        self.assert_mock_multiple_calls_with_same_arguments(1014            mock_limits, 2,1015            mock.call(test.IsHttpRequest()))1016        self.assertNotContains(res, 'Delete Volume')1017        self.assertNotContains(res, 'delete')1018    @mock.patch.object(api.nova, 'server_list')1019    @mock.patch.object(cinder, 'volume_get')1020    @override_settings(OPENSTACK_HYPERVISOR_FEATURES={'can_set_mount_point':1021                                                      True})1022    def test_edit_attachments(self, mock_get, mock_server_list):1023        volume = self.cinder_volumes.first()1024        servers = [s for s in self.servers.list()1025                   if s.tenant_id == self.request.user.tenant_id]1026        volume.attachments = [{'id': volume.id,1027                               'volume_id': volume.id,1028                               'volume_name': volume.name,1029                               "attachment_id": ATTACHMENT_ID,1030                               'instance': servers[0],1031                               'device': '/dev/vdb',1032                               'server_id': servers[0].id}]1033        mock_get.return_value = volume1034        mock_server_list.return_value = [servers, False]1035        url = reverse('horizon:project:volumes:attach',1036                      args=[volume.id])1037        res = self.client.get(url)1038        msg = 'Volume %s on instance %s' % (volume.name, servers[0].name)1039        self.assertContains(res, msg)1040        # Asserting length of 2 accounts for the one instance option,1041        # and the one 'Choose Instance' option.1042        form = res.context['form']1043        self.assertEqual(len(form.fields['instance']._choices),1044                         1)1045        self.assertEqual(res.status_code, 200)1046        self.assertIsInstance(form.fields['device'].widget,1047                              widgets.TextInput)1048        self.assertFalse(form.fields['device'].required)1049        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1050        mock_server_list.assert_called_once()1051    @mock.patch.object(api.nova, 'server_list')1052    @mock.patch.object(cinder, 'volume_get')1053    @override_settings(OPENSTACK_HYPERVISOR_FEATURES={'can_set_mount_point':1054                                                      True})1055    def test_edit_attachments_auto_device_name(self, mock_get,1056                                               mock_server_list):1057        volume = self.cinder_volumes.first()1058        servers = [s for s in self.servers.list()1059                   if s.tenant_id == self.request.user.tenant_id]1060        volume.attachments = [{'id': volume.id,1061                               'volume_id': volume.id,1062                               'volume_name': volume.name,1063                               "attachment_id": ATTACHMENT_ID,1064                               'instance': servers[0],1065                               'device': '',1066                               'server_id': servers[0].id}]1067        mock_get.return_value = volume1068        mock_server_list.return_value = [servers, False]1069        url = reverse('horizon:project:volumes:attach',1070                      args=[volume.id])1071        res = self.client.get(url)1072        form = res.context['form']1073        self.assertIsInstance(form.fields['device'].widget,1074                              widgets.TextInput)1075        self.assertFalse(form.fields['device'].required)1076        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1077        mock_server_list.assert_called_once()1078    @mock.patch.object(api.nova, 'server_list')1079    @mock.patch.object(cinder, 'volume_get')1080    def test_edit_attachments_cannot_set_mount_point(self, mock_get,1081                                                     mock_server_list):1082        volume = self.cinder_volumes.first()1083        url = reverse('horizon:project:volumes:attach',1084                      args=[volume.id])1085        res = self.client.get(url)1086        # Assert the device field is hidden.1087        form = res.context['form']1088        self.assertIsInstance(form.fields['device'].widget,1089                              widgets.HiddenInput)1090        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1091        mock_server_list.assert_called_once()1092    @mock.patch.object(api.nova, 'server_list')1093    @mock.patch.object(cinder, 'volume_get')1094    def test_edit_attachments_attached_volume(self, mock_get,1095                                              mock_server_list):1096        servers = [s for s in self.servers.list()1097                   if s.tenant_id == self.request.user.tenant_id]1098        server = servers[0]1099        volume = self.cinder_volumes.list()[0]1100        mock_get.return_value = volume1101        mock_server_list.return_value = [servers, False]1102        url = reverse('horizon:project:volumes:attach',1103                      args=[volume.id])1104        res = self.client.get(url)1105        self.assertEqual(res.context['form'].fields['instance']._choices[0][1],1106                         "Select an instance")1107        self.assertEqual(len(res.context['form'].fields['instance'].choices),1108                         2)1109        self.assertEqual(res.context['form'].fields['instance']._choices[1][0],1110                         server.id)1111        self.assertEqual(res.status_code, 200)1112        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1113        mock_server_list.assert_called_once()1114    @mock.patch.object(quotas, 'tenant_quota_usages')1115    @mock.patch.object(cinder, 'tenant_absolute_limits')1116    @mock.patch.object(cinder, 'volume_get')1117    def test_create_snapshot_button_attributes(self, mock_get,1118                                               mock_limits,1119                                               mock_quotas):1120        limits = {'maxTotalSnapshots': 2}1121        limits['totalSnapshotsUsed'] = 11122        volume = self.cinder_volumes.first()1123        mock_get.return_value = volume1124        mock_limits.return_value = limits1125        mock_quotas.return_value = self.cinder_quota_usages.first()1126        res_url = (INDEX_URL +1127                   "?action=row_update&table=volumes&obj_id=" + volume.id)1128        res = self.client.get(res_url, {},1129                              HTTP_X_REQUESTED_WITH='XMLHttpRequest')1130        action_name = ('%(table)s__row_%(id)s__action_%(action)s' %1131                       {'table': 'volumes', 'id': volume.id,1132                        'action': 'snapshots'})1133        content = res.content.decode('utf-8')1134        self.assertIn(action_name, content)1135        self.assertIn('Create Snapshot', content)1136        self.assertIn(reverse('horizon:project:volumes:create_snapshot',1137                              args=[volume.id]),1138                      content)1139        self.assertNotIn('disabled', content)1140        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1141        mock_quotas.assert_called_once_with(test.IsHttpRequest(),1142                                            targets=('volumes', 'gigabytes'))1143        self.assert_mock_multiple_calls_with_same_arguments(1144            mock_limits, 2,1145            mock.call(test.IsHttpRequest()))1146    @mock.patch.object(quotas, 'tenant_quota_usages')1147    @mock.patch.object(cinder, 'tenant_absolute_limits')1148    @mock.patch.object(cinder, 'volume_get')1149    def test_create_snapshot_button_disabled_when_quota_exceeded(1150            self, mock_get, mock_limits, mock_quotas):1151        limits = {'maxTotalSnapshots': 1}1152        limits['totalSnapshotsUsed'] = limits['maxTotalSnapshots']1153        volume = self.cinder_volumes.first()1154        mock_get.return_value = volume1155        mock_limits.return_value = limits1156        mock_quotas.return_value = self.cinder_quota_usages.first()1157        res_url = (INDEX_URL +1158                   "?action=row_update&table=volumes&obj_id=" + volume.id)1159        res = self.client.get(res_url, {},1160                              HTTP_X_REQUESTED_WITH='XMLHttpRequest')1161        action_name = ('%(table)s__row_%(id)s__action_%(action)s' %1162                       {'table': 'volumes', 'id': volume.id,1163                        'action': 'snapshots'})1164        content = res.content.decode('utf-8')1165        self.assertIn(action_name, content)1166        self.assertIn('Create Snapshot (Quota exceeded)', content)1167        self.assertIn(reverse('horizon:project:volumes:create_snapshot',1168                              args=[volume.id]),1169                      content)1170        self.assertIn('disabled', content,1171                      'The create snapshot button should be disabled')1172        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1173        mock_quotas.assert_called_once_with(test.IsHttpRequest(),1174                                            targets=('volumes', 'gigabytes'))1175        self.assert_mock_multiple_calls_with_same_arguments(1176            mock_limits, 2,1177            mock.call(test.IsHttpRequest()))1178    @test.create_mocks({1179        api.nova: ['server_list'],1180        cinder: ['volume_backup_supported',1181                 'volume_snapshot_list',1182                 'volume_list_paged',1183                 'tenant_absolute_limits'],1184    })1185    def test_create_button_attributes(self):1186        limits = self.cinder_limits['absolute']1187        limits['maxTotalVolumes'] = 101188        limits['totalVolumesUsed'] = 11189        volumes = self.cinder_volumes.list()1190        self.mock_volume_backup_supported.return_value = True1191        self.mock_volume_list_paged.return_value = [volumes, False, False]1192        self.mock_volume_snapshot_list.return_value = []1193        self.mock_server_list.return_value = [self.servers.list(), False]1194        self.mock_tenant_absolute_limits.return_value = limits1195        res = self.client.get(INDEX_URL)1196        self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')1197        volumes = res.context['volumes_table'].data1198        self.assertItemsEqual(volumes, self.cinder_volumes.list())1199        create_action = self.getAndAssertTableAction(res, 'volumes', 'create')1200        self.assertEqual(set(['ajax-modal', 'ajax-update', 'btn-create']),1201                         set(create_action.classes))1202        self.assertEqual('Create Volume',1203                         six.text_type(create_action.verbose_name))1204        self.assertEqual('horizon:project:volumes:create',1205                         create_action.url)1206        self.assertEqual((('volume', 'volume:create'),),1207                         create_action.policy_rules)1208        self.assertEqual(5, self.mock_volume_backup_supported.call_count)1209        self.mock_volume_list_paged.assert_called_once_with(1210            test.IsHttpRequest(), sort_dir='desc', marker=None,1211            paginate=True, search_opts=None)1212        self.mock_volume_snapshot_list.assert_called_once_with(1213            test.IsHttpRequest(), search_opts=None)1214        self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),1215                                                      search_opts=None)1216        self.assertEqual(9, self.mock_tenant_absolute_limits.call_count)1217    @test.create_mocks({1218        api.nova: ['server_list'],1219        cinder: ['volume_backup_supported',1220                 'volume_snapshot_list',1221                 'volume_list_paged',1222                 'tenant_absolute_limits'],1223    })1224    def test_create_button_disabled_when_quota_exceeded(self):1225        limits = self.cinder_limits['absolute']1226        limits['totalVolumesUsed'] = limits['maxTotalVolumes']1227        volumes = self.cinder_volumes.list()1228        self.mock_volume_backup_supported.return_value = True1229        self.mock_volume_list_paged.return_value = [volumes, False, False]1230        self.mock_volume_snapshot_list.return_value = []1231        self.mock_server_list.return_value = [self.servers.list(), False]1232        self.mock_tenant_absolute_limits.return_value = limits1233        res = self.client.get(INDEX_URL)1234        self.assertTemplateUsed(res, 'horizon/common/_data_table_view.html')1235        volumes = res.context['volumes_table'].data1236        self.assertItemsEqual(volumes, self.cinder_volumes.list())1237        create_action = self.getAndAssertTableAction(res, 'volumes', 'create')1238        self.assertIn('disabled', create_action.classes,1239                      'The create button should be disabled')1240        self.assertEqual(5, self.mock_volume_backup_supported.call_count)1241        self.mock_volume_list_paged.assert_called_once_with(1242            test.IsHttpRequest(), marker=None,1243            paginate=True, sort_dir='desc',1244            search_opts=None)1245        self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),1246                                                      search_opts=None)1247        self.assertEqual(9, self.mock_tenant_absolute_limits.call_count)1248    @test.create_mocks({1249        api.nova: ['server_get'],1250        cinder: ['message_list',1251                 'volume_snapshot_list',1252                 'volume_get',1253                 'tenant_absolute_limits'],1254    })1255    def test_detail_view(self):1256        volume = self.cinder_volumes.first()1257        server = self.servers.first()1258        snapshots = self.cinder_volume_snapshots.list()1259        volume.attachments = [{"server_id": server.id,1260                               "attachment_id": ATTACHMENT_ID}]1261        self.mock_volume_get.return_value = volume1262        self.mock_volume_snapshot_list.return_value = snapshots1263        self.mock_server_get.return_value = server1264        self.mock_tenant_absolute_limits.return_value = \1265            self.cinder_limits['absolute']1266        self.mock_message_list.return_value = []1267        url = reverse('horizon:project:volumes:detail',1268                      args=[volume.id])1269        res = self.client.get(url)1270        self.assertTemplateUsed(res, 'horizon/common/_detail.html')1271        self.assertEqual(res.context['volume'].id, volume.id)1272        self.assertNoMessages()1273        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),1274                                                     volume.id)1275        self.mock_volume_snapshot_list.assert_called_once_with(1276            test.IsHttpRequest(), search_opts={'volume_id': volume.id})1277        self.mock_server_get.assert_called_once_with(test.IsHttpRequest(),1278                                                     server.id)1279        self.mock_tenant_absolute_limits.assert_called_once()1280        self.mock_message_list.assert_called_once_with(1281            test.IsHttpRequest(),1282            {1283                'resource_uuid': '11023e92-8008-4c8b-8059-7f2293ff3887',1284                'resource_type': 'volume',1285            })1286    @mock.patch.object(cinder, 'volume_get_encryption_metadata')1287    @mock.patch.object(cinder, 'volume_get')1288    def test_encryption_detail_view_encrypted(self, mock_get, mock_encryption):1289        enc_meta = self.cinder_volume_encryption.first()1290        volume = self.cinder_volumes.get(name='my_volume2')1291        mock_encryption.return_value = enc_meta1292        mock_get.return_value = volume1293        url = reverse('horizon:project:volumes:encryption_detail',1294                      args=[volume.id])1295        res = self.client.get(url)1296        self.assertContains(res,1297                            "Volume Encryption Details: %s" % volume.name,1298                            2, 200)1299        self.assertContains(res, "<dd>%s</dd>" % volume.volume_type, 1, 200)1300        self.assertContains(res, "<dd>%s</dd>" % enc_meta.provider, 1, 200)1301        self.assertContains(res, "<dd>%s</dd>" % enc_meta.control_location, 1,1302                            200)1303        self.assertContains(res, "<dd>%s</dd>" % enc_meta.cipher, 1, 200)1304        self.assertContains(res, "<dd>%s</dd>" % enc_meta.key_size, 1, 200)1305        self.assertNoMessages()1306        mock_encryption.assert_called_once_with(test.IsHttpRequest(),1307                                                volume.id)1308        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1309    @mock.patch.object(cinder, 'volume_get_encryption_metadata')1310    @mock.patch.object(cinder, 'volume_get')1311    def test_encryption_detail_view_unencrypted(self, mock_get,1312                                                mock_encryption):1313        enc_meta = self.cinder_volume_encryption.list()[1]1314        volume = self.cinder_volumes.get(name='my_volume2')1315        mock_encryption.return_value = enc_meta1316        mock_get.return_value = volume1317        url = reverse('horizon:project:volumes:encryption_detail',1318                      args=[volume.id])1319        res = self.client.get(url)1320        self.assertContains(res,1321                            "Volume Encryption Details: %s" % volume.name,1322                            2, 200)1323        self.assertContains(res, "<h3>Volume is Unencrypted</h3>", 1, 200)1324        self.assertNoMessages()1325        mock_encryption.assert_called_once_with(test.IsHttpRequest(),1326                                                volume.id)1327        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1328    @mock.patch.object(quotas, 'tenant_quota_usages')1329    @mock.patch.object(cinder, 'tenant_absolute_limits')1330    @mock.patch.object(cinder, 'volume_get')1331    def test_get_data(self, mock_get, mock_limits, mock_quotas):1332        volume = self.cinder_volumes.get(name='v2_volume')1333        volume._apiresource.name = ""1334        mock_get.return_value = volume1335        mock_limits.return_value = self.cinder_limits['absolute']1336        mock_quotas.return_value = self.cinder_quota_usages.first()1337        url = (INDEX_URL +1338               "?action=row_update&table=volumes&obj_id=" + volume.id)1339        res = self.client.get(url, {},1340                              HTTP_X_REQUESTED_WITH='XMLHttpRequest')1341        self.assertEqual(res.status_code, 200)1342        self.assertEqual(volume.name, volume.id)1343        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1344        mock_quotas.assert_called_once_with(test.IsHttpRequest(),1345                                            targets=('volumes', 'gigabytes'))1346        self.assert_mock_multiple_calls_with_same_arguments(1347            mock_limits, 2,1348            mock.call(test.IsHttpRequest()))1349    @test.create_mocks({1350        api.nova: ['server_get'],1351        cinder: ['tenant_absolute_limits',1352                 'volume_get',1353                 'volume_snapshot_list',1354                 'message_list'],1355    })1356    def test_detail_view_snapshot_tab(self):1357        volume = self.cinder_volumes.first()1358        server = self.servers.first()1359        snapshots = self.cinder_volume_snapshots.list()1360        this_volume_snapshots = [snapshot for snapshot in snapshots1361                                 if snapshot.volume_id == volume.id]1362        volume.attachments = [{"server_id": server.id,1363                               "attachment_id": ATTACHMENT_ID}]1364        self.mock_volume_get.return_value = volume1365        self.mock_server_get.return_value = server1366        self.mock_tenant_absolute_limits.return_value = \1367            self.cinder_limits['absolute']1368        self.mock_message_list.return_value = []1369        self.mock_volume_snapshot_list.return_value = this_volume_snapshots1370        url = '?'.join([reverse(DETAIL_URL, args=[volume.id]),1371                        '='.join(['tab', 'volume_details__snapshots_tab'])])1372        res = self.client.get(url)1373        self.assertTemplateUsed(res, 'horizon/common/_detail.html')1374        self.assertEqual(res.context['volume'].id, volume.id)1375        self.assertEqual(len(res.context['table'].data),1376                         len(this_volume_snapshots))1377        self.assertNoMessages()1378        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),1379                                                     volume.id)1380        self.mock_volume_snapshot_list.assert_called_once_with(1381            test.IsHttpRequest(), search_opts={'volume_id': volume.id})1382        self.mock_tenant_absolute_limits.assert_called_once()1383        self.mock_message_list.assert_called_once_with(1384            test.IsHttpRequest(),1385            {1386                'resource_uuid': volume.id,1387                'resource_type': 'volume'1388            })1389    @mock.patch.object(cinder, 'volume_get')1390    def test_detail_view_with_exception(self, mock_get):1391        volume = self.cinder_volumes.first()1392        server = self.servers.first()1393        volume.attachments = [{"server_id": server.id,1394                               "attachment_id": ATTACHMENT_ID}]1395        mock_get.side_effect = self.exceptions.cinder1396        url = reverse('horizon:project:volumes:detail',1397                      args=[volume.id])1398        res = self.client.get(url)1399        self.assertRedirectsNoFollow(res, INDEX_URL)1400        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1401    @test.create_mocks({cinder: ['volume_update',1402                                 'volume_set_bootable',1403                                 'volume_get']})1404    def test_update_volume(self):1405        volume = self.cinder_volumes.get(name="my_volume")1406        self.mock_volume_get.return_value = volume1407        formData = {'method': 'UpdateForm',1408                    'name': volume.name,1409                    'description': volume.description,1410                    'bootable': False}1411        url = reverse('horizon:project:volumes:update',1412                      args=[volume.id])1413        res = self.client.post(url, formData)1414        self.assertRedirectsNoFollow(res, INDEX_URL)1415        self.mock_volume_get.assert_called_once_with(1416            test.IsHttpRequest(), volume.id)1417        self.mock_volume_update.assert_called_once_with(1418            test.IsHttpRequest(), volume.id, volume.name, volume.description)1419        self.mock_volume_set_bootable.assert_called_once_with(1420            test.IsHttpRequest(), volume.id, False)1421    @test.create_mocks({cinder: ['volume_update',1422                                 'volume_set_bootable',1423                                 'volume_get']})1424    def test_update_volume_without_name(self):1425        volume = self.cinder_volumes.get(name="my_volume")1426        self.mock_volume_get.return_value = volume1427        formData = {'method': 'UpdateForm',1428                    'name': '',1429                    'description': volume.description,1430                    'bootable': False}1431        url = reverse('horizon:project:volumes:update',1432                      args=[volume.id])1433        res = self.client.post(url, formData)1434        self.assertRedirectsNoFollow(res, INDEX_URL)1435        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),1436                                                     volume.id)1437        self.mock_volume_update.assert_called_once_with(1438            test.IsHttpRequest(), volume.id, '', volume.description)1439        self.mock_volume_set_bootable.assert_called_once_with(1440            test.IsHttpRequest(), volume.id, False)1441    @test.create_mocks({cinder: ['volume_update',1442                                 'volume_set_bootable',1443                                 'volume_get']})1444    def test_update_volume_bootable_flag(self):1445        volume = self.cinder_bootable_volumes.get(name="my_volume")1446        self.mock_volume_get.return_value = volume1447        formData = {'method': 'UpdateForm',1448                    'name': volume.name,1449                    'description': 'update bootable flag',1450                    'bootable': True}1451        url = reverse('horizon:project:volumes:update',1452                      args=[volume.id])1453        res = self.client.post(url, formData)1454        self.assertRedirectsNoFollow(res, INDEX_URL)1455        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),1456                                                     volume.id)1457        self.mock_volume_update.assert_called_once_with(1458            test.IsHttpRequest(), volume.id, volume.name,1459            'update bootable flag')1460        self.mock_volume_set_bootable.assert_called_once_with(1461            test.IsHttpRequest(), volume.id, True)1462    @mock.patch.object(cinder, 'volume_upload_to_image')1463    @mock.patch.object(cinder, 'volume_get')1464    def test_upload_to_image(self, mock_get, mock_upload):1465        volume = self.cinder_volumes.get(name='v2_volume')1466        loaded_resp = {'container_format': 'bare',1467                       'disk_format': 'raw',1468                       'id': '741fe2ac-aa2f-4cec-82a9-4994896b43fb',1469                       'image_id': '2faa080b-dd56-4bf0-8f0a-0d4627d8f306',1470                       'image_name': 'test',1471                       'size': '2',1472                       'status': 'uploading'}1473        form_data = {'id': volume.id,1474                     'name': volume.name,1475                     'image_name': 'testimage',1476                     'force': True,1477                     'container_format': 'bare',1478                     'disk_format': 'raw'}1479        mock_get.return_value = volume1480        mock_upload.return_value = loaded_resp1481        url = reverse('horizon:project:volumes:upload_to_image',1482                      args=[volume.id])1483        res = self.client.post(url, form_data)1484        self.assertNoFormErrors(res)1485        self.assertMessageCount(info=1)1486        redirect_url = INDEX_URL1487        self.assertRedirectsNoFollow(res, redirect_url)1488        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1489        mock_upload.assert_called_once_with(test.IsHttpRequest(),1490                                            form_data['id'],1491                                            form_data['force'],1492                                            form_data['image_name'],1493                                            form_data['container_format'],1494                                            form_data['disk_format'])1495    @mock.patch.object(quotas, 'tenant_quota_usages')1496    @mock.patch.object(cinder, 'volume_extend')1497    @mock.patch.object(cinder, 'volume_get')1498    def test_extend_volume(self, mock_get, mock_extend, mock_quotas):1499        volume = self.cinder_volumes.first()1500        formData = {'name': u'A Volume I Am Making',1501                    'orig_size': volume.size,1502                    'new_size': 120}1503        mock_get.return_value = volume1504        mock_quotas.return_value = self.cinder_quota_usages.first()1505        mock_extend.return_value = volume1506        url = reverse('horizon:project:volumes:extend',1507                      args=[volume.id])1508        res = self.client.post(url, formData)1509        redirect_url = INDEX_URL1510        self.assertRedirectsNoFollow(res, redirect_url)1511        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1512        mock_quotas.assert_called_once()1513        mock_extend.assert_called_once_with(test.IsHttpRequest(), volume.id,1514                                            formData['new_size'])1515    @mock.patch.object(quotas, 'tenant_quota_usages')1516    @mock.patch.object(cinder, 'volume_get')1517    def test_extend_volume_with_wrong_size(self, mock_get, mock_quotas):1518        volume = self.cinder_volumes.first()1519        formData = {'name': u'A Volume I Am Making',1520                    'orig_size': volume.size,1521                    'new_size': 10}1522        mock_get.return_value = volume1523        mock_quotas.return_value = self.cinder_quota_usages.first()1524        url = reverse('horizon:project:volumes:extend',1525                      args=[volume.id])1526        res = self.client.post(url, formData)1527        self.assertFormErrors(res, 1,1528                              "New size must be greater than "1529                              "current size.")1530        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1531        mock_quotas.assert_called_once()1532    @mock.patch.object(quotas, 'tenant_quota_usages')1533    @mock.patch.object(cinder, 'tenant_absolute_limits')1534    @mock.patch.object(cinder, 'volume_get')1535    def test_retype_volume_supported_action_item(self, mock_get,1536                                                 mock_limits, mock_quotas):1537        volume = self.cinder_volumes.get(name='v2_volume')1538        limits = self.cinder_limits['absolute']1539        mock_get.return_value = volume1540        mock_limits.return_value = limits1541        mock_quotas.return_value = self.cinder_quota_usages.first()1542        url = (INDEX_URL +1543               "?action=row_update&table=volumes&obj_id=" + volume.id)1544        res = self.client.get(url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')1545        self.assertEqual(res.status_code, 200)1546        self.assertContains(res, 'Change Volume Type')1547        self.assertContains(res, 'retype')1548        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1549        mock_quotas.assert_called_once_with(test.IsHttpRequest(),1550                                            targets=('volumes', 'gigabytes'))1551        self.assert_mock_multiple_calls_with_same_arguments(1552            mock_limits, 2,1553            mock.call(test.IsHttpRequest()))1554    @test.create_mocks({1555        cinder: ['volume_type_list',1556                 'volume_retype',1557                 'volume_get']1558    })1559    def test_retype_volume(self):1560        volume = self.cinder_volumes.get(name='my_volume2')1561        volume_type = self.cinder_volume_types.get(name='vol_type_1')1562        form_data = {'id': volume.id,1563                     'name': volume.name,1564                     'volume_type': volume_type.name,1565                     'migration_policy': 'on-demand'}1566        self.mock_volume_get.return_value = volume1567        self.mock_volume_type_list.return_value = \1568            self.cinder_volume_types.list()1569        self.mock_volume_retype.return_value = True1570        url = reverse('horizon:project:volumes:retype',1571                      args=[volume.id])1572        res = self.client.post(url, form_data)1573        self.assertNoFormErrors(res)1574        redirect_url = INDEX_URL1575        self.assertRedirectsNoFollow(res, redirect_url)1576        self.mock_volume_get.assert_called_once_with(test.IsHttpRequest(),1577                                                     volume.id)1578        self.mock_volume_type_list.assert_called_once()1579        self.mock_volume_retype.assert_called_once_with(1580            test.IsHttpRequest(), volume.id,1581            form_data['volume_type'], form_data['migration_policy'])1582    def test_encryption_false(self):1583        self._test_encryption(False)1584    def test_encryption_true(self):1585        self._test_encryption(True)1586    @test.create_mocks({1587        api.nova: ['server_list'],1588        cinder: ['volume_backup_supported',1589                 'volume_list_paged',1590                 'volume_snapshot_list',1591                 'tenant_absolute_limits'],1592    })1593    def _test_encryption(self, encryption):1594        volumes = self.cinder_volumes.list()1595        for volume in volumes:1596            volume.encrypted = encryption1597        limits = self.cinder_limits['absolute']1598        self.mock_volume_backup_supported.return_value = False1599        self.mock_volume_list_paged.return_value = [self.cinder_volumes.list(),1600                                                    False, False]1601        self.mock_volume_snapshot_list.return_value = \1602            self.cinder_volume_snapshots.list()1603        self.mock_server_list.return_value = [self.servers.list(), False]1604        self.mock_tenant_absolute_limits.return_value = limits1605        res = self.client.get(INDEX_URL)1606        rows = res.context['volumes_table'].get_rows()1607        column_value = 'Yes' if encryption else 'No'1608        for row in rows:1609            self.assertEqual(row.cells['encryption'].data, column_value)1610        self.assertEqual(10, self.mock_volume_backup_supported.call_count)1611        self.mock_volume_list_paged.assert_called_once_with(1612            test.IsHttpRequest(), marker=None,1613            sort_dir='desc', search_opts=None,1614            paginate=True)1615        self.mock_volume_snapshot_list.assert_called_once_with(1616            test.IsHttpRequest(), search_opts=None)1617        self.assertEqual(13, self.mock_tenant_absolute_limits.call_count)1618    @mock.patch.object(quotas, 'tenant_quota_usages')1619    @mock.patch.object(cinder, 'volume_get')1620    def test_extend_volume_with_size_out_of_quota(self, mock_get, mock_quotas):1621        volume = self.cinder_volumes.first()1622        usage_limit = self.cinder_quota_usages.first()1623        usage_limit.add_quota(api.base.Quota('gigabytes', 100))1624        usage_limit.tally('gigabytes', 20)1625        usage_limit.tally('volumes', len(self.cinder_volumes.list()))1626        formData = {'name': u'A Volume I Am Making',1627                    'orig_size': volume.size,1628                    'new_size': 1000}1629        mock_quotas.return_value = usage_limit1630        mock_get.return_value = volume1631        url = reverse('horizon:project:volumes:extend',1632                      args=[volume.id])1633        res = self.client.post(url, formData)1634        self.assertFormError(res, "form", "new_size",1635                             "Volume cannot be extended to 1000GiB as "1636                             "the maximum size it can be extended to is "1637                             "120GiB.")1638        mock_get.assert_called_once_with(test.IsHttpRequest(), volume.id)1639        self.assertEqual(2, mock_quotas.call_count)1640    @test.create_mocks({1641        api.nova: ['server_list'],1642        cinder: ['volume_backup_supported',1643                 'volume_list_paged',1644                 'volume_snapshot_list',1645                 'tenant_absolute_limits'],1646    })1647    def test_create_transfer_availability(self):1648        limits = self.cinder_limits['absolute']1649        self.mock_volume_backup_supported.return_value = False1650        self.mock_volume_list_paged.return_value = [self.cinder_volumes.list(),1651                                                    False, False]1652        self.mock_volume_snapshot_list.return_value = []1653        self.mock_server_list.return_value = [self.servers.list(), False]1654        self.mock_tenant_absolute_limits.return_value = limits1655        res = self.client.get(INDEX_URL)1656        table = res.context['volumes_table']1657        # Verify that the create transfer action is present if and only if1658        # the volume is available1659        for vol in table.data:1660            actions = [a.name for a in table.get_row_actions(vol)]1661            self.assertEqual('create_transfer' in actions,1662                             vol.status == 'available')1663        self.assertEqual(10, self.mock_volume_backup_supported.call_count)1664        self.mock_volume_list_paged.assert_called_once_with(1665            test.IsHttpRequest(), marker=None,1666            sort_dir='desc', search_opts=None,1667            paginate=True)1668        self.mock_volume_snapshot_list.assert_called_once_with(1669            test.IsHttpRequest(), search_opts=None)1670        self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),1671                                                      search_opts=None)1672        self.assertEqual(13, self.mock_tenant_absolute_limits.call_count)1673    @mock.patch.object(cinder, 'transfer_get')1674    @mock.patch.object(cinder, 'transfer_create')1675    def test_create_transfer(self, mock_transfer_create, mock_transfer_get):1676        volumes = self.cinder_volumes.list()1677        volToTransfer = [v for v in volumes if v.status == 'available'][0]1678        formData = {'volume_id': volToTransfer.id,1679                    'name': u'any transfer name'}1680        transfer = self.cinder_volume_transfers.first()1681        mock_transfer_create.return_value = transfer1682        mock_transfer_get.return_value = transfer1683        # Create a transfer for the first available volume1684        url = reverse('horizon:project:volumes:create_transfer',1685                      args=[volToTransfer.id])1686        res = self.client.post(url, formData)1687        self.assertNoFormErrors(res)1688        mock_transfer_create.assert_called_once_with(test.IsHttpRequest(),1689                                                     formData['volume_id'],1690                                                     formData['name'])1691        mock_transfer_get.assert_called_once_with(test.IsHttpRequest(),1692                                                  transfer.id)1693    @test.create_mocks({1694        api.nova: ['server_list'],1695        cinder: ['volume_backup_supported',1696                 'volume_list_paged',1697                 'volume_snapshot_list',1698                 'transfer_delete',1699                 'tenant_absolute_limits'],1700    })1701    def test_delete_transfer(self):1702        transfer = self.cinder_volume_transfers.first()1703        volumes = []1704        # Attach the volume transfer to the relevant volume1705        for v in self.cinder_volumes.list():1706            if v.id == transfer.volume_id:1707                v.status = 'awaiting-transfer'1708                v.transfer = transfer1709            volumes.append(v)1710        formData = {'action':1711                    'volumes__delete_transfer__%s' % transfer.volume_id}1712        self.mock_volume_backup_supported.return_value = False1713        self.mock_volume_list_paged.return_value = [volumes, False, False]1714        self.mock_volume_snapshot_list.return_value = []1715        self.mock_server_list.return_value = [self.servers.list(), False]1716        self.mock_tenant_absolute_limits.return_value = \1717            self.cinder_limits['absolute']1718        url = INDEX_URL1719        res = self.client.post(url, formData, follow=True)1720        self.assertNoFormErrors(res)1721        self.assertIn('Successfully deleted volume transfer "test transfer"',1722                      [m.message for m in res.context['messages']])1723        self.assertEqual(5, self.mock_volume_backup_supported.call_count)1724        self.mock_volume_list_paged.assert_called_once_with(1725            test.IsHttpRequest(), marker=None,1726            search_opts=None, sort_dir='desc',1727            paginate=True)1728        self.mock_volume_snapshot_list.assert_called_once_with(1729            test.IsHttpRequest(), search_opts=None)1730        self.mock_transfer_delete.assert_called_once_with(test.IsHttpRequest(),1731                                                          transfer.id)1732        self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),1733                                                      search_opts=None)1734        self.assertEqual(8, self.mock_tenant_absolute_limits.call_count)1735    @test.create_mocks({1736        api.nova: ['server_list'],1737        cinder: ['volume_list_paged',1738                 'volume_snapshot_list',1739                 'tenant_absolute_limits',1740                 'transfer_accept']1741    })1742    def test_accept_transfer(self):1743        transfer = self.cinder_volume_transfers.first()1744        self.mock_tenant_absolute_limits.return_value = \1745            self.cinder_limits['absolute']1746        formData = {'transfer_id': transfer.id, 'auth_key': transfer.auth_key}1747        url = reverse('horizon:project:volumes:accept_transfer')1748        res = self.client.post(url, formData, follow=True)1749        self.assertNoFormErrors(res)1750        self.mock_transfer_accept.assert_called_once_with(test.IsHttpRequest(),1751                                                          transfer.id,1752                                                          transfer.auth_key)1753        self.assertEqual(3, self.mock_tenant_absolute_limits.call_count)1754        self.mock_server_list.assert_called_once()1755        self.mock_volume_list_paged.assert_called_once()1756        self.mock_volume_snapshot_list.assert_called_once()1757        self.mock_transfer_accept.assert_called_once()1758    @mock.patch.object(cinder, 'transfer_get')1759    def test_download_transfer_credentials(self, mock_transfer):1760        transfer = self.cinder_volume_transfers.first()1761        filename = "{}.txt".format(slugify(transfer.id))1762        url = reverse('horizon:project:volumes:'1763                      'download_transfer_creds',1764                      kwargs={'transfer_id': transfer.id,1765                              'auth_key': transfer.auth_key})1766        res = self.client.get(url)1767        self.assertTrue(res.has_header('content-disposition'))1768        self.assertTrue(res.has_header('content-type'))1769        self.assertEqual(res.get('content-disposition'),1770                         'attachment; filename={}'.format(filename))1771        self.assertEqual(res.get('content-type'), 'application/text')1772        self.assertIn(transfer.id, res.content.decode('utf-8'))1773        self.assertIn(transfer.auth_key, res.content.decode('utf-8'))1774        mock_transfer.assert_called_once_with(test.IsHttpRequest(),1775                                              transfer.id)1776    @test.create_mocks({1777        api.nova: ['server_list'],1778        cinder: ['volume_backup_supported',1779                 'volume_list_paged',1780                 'volume_snapshot_list',1781                 'tenant_absolute_limits',1782                 'volume_get'],1783    })1784    def test_create_backup_availability(self):1785        limits = self.cinder_limits['absolute']1786        self.mock_volume_backup_supported.return_value = True1787        self.mock_volume_list_paged.return_value = [self.cinder_volumes.list(),1788                                                    False, False]1789        self.mock_volume_snapshot_list.return_value = []1790        self.mock_server_list.return_value = [self.servers.list(), False]1791        self.mock_tenant_absolute_limits.return_value = limits1792        res = self.client.get(INDEX_URL)1793        table = res.context['volumes_table']1794        # Verify that the create backup action is present if and only if1795        # the volume is available or in-use1796        for vol in table.data:1797            actions = [a.name for a in table.get_row_actions(vol)]1798            self.assertEqual('backups' in actions,1799                             vol.status in ('available', 'in-use'))1800        self.assertEqual(10, self.mock_volume_backup_supported.call_count)1801        self.mock_volume_list_paged.assert_called_once_with(1802            test.IsHttpRequest(), marker=None,1803            sort_dir='desc', search_opts=None,1804            paginate=True)1805        self.mock_volume_snapshot_list.assert_called_once_with(1806            test.IsHttpRequest(), search_opts=None)1807        self.mock_server_list.assert_called_once_with(test.IsHttpRequest(),1808                                                      search_opts=None)...

test_nova.py

Source:test_nova.py

1# Copyright 2014, Rackspace, US, Inc.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain 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,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import json15from json import loads as to_json16import uuid17from django.conf import settings18import mock19from openstack_dashboard import api20from openstack_dashboard.api.base import Quota21from openstack_dashboard.api.rest import nova22from openstack_dashboard.test import helpers as test23from openstack_dashboard.usage import quotas24# NOTE(flwang): mock.Mock and mock.MagicMock do not support sort, so the test25# case involved sorted will fail. This fake class is for the flavor test cases26# related to sort.27class FakeFlavor(object):28    def __init__(self, id, ram=1):29        self.id = id30        self.ram = ram31        self.extras = {}32    def to_dict(self):33        return {"id": self.id}34class NovaRestTestCase(test.TestCase):35    #36    # Snapshots37    #38    @test.create_mocks({api.nova: ['snapshot_create']})39    def test_snapshots_create(self):40        body = '{"instance_id": "1234", "name": "foo"}'41        request = self.mock_rest_request(body=body)42        self.mock_snapshot_create.return_value = {'id': 'abcd', 'name': 'foo'}43        response = nova.Snapshots().post(request)44        self.assertStatusCode(response, 200)45        self.assertEqual(response.json, {'id': 'abcd', 'name': 'foo'})46        self.mock_snapshot_create.assert_called_once_with(request,47                                                          instance_id='1234',48                                                          name='foo')49    #50    # Server Actions51    #52    @test.create_mocks({api.nova: ['instance_action_list']})53    def test_serveractions_list(self):54        request = self.mock_rest_request()55        self.mock_instance_action_list.return_value = [56            mock.Mock(**{'to_dict.return_value': {'id': '1'}}),57            mock.Mock(**{'to_dict.return_value': {'id': '2'}}),58        ]59        response = nova.ServerActions().get(request, 'MegaMan')60        self.assertStatusCode(response, 200)61        self.assertEqual(response.json, {'items': [{'id': '1'}, {'id': '2'}]})62        self.mock_instance_action_list.assert_called_once_with(request,63                                                               'MegaMan')64    @test.create_mocks({api.nova: ['server_start']})65    def test_server_start(self):66        self.mock_server_start.return_value = None67        request = self.mock_rest_request(body='{"operation": "start"}')68        response = nova.Server().post(request, 'MegaMan')69        self.assertStatusCode(response, 204)70        self.mock_server_start.assert_called_once_with(request, 'MegaMan')71    @test.create_mocks({api.nova: ['server_stop']})72    def test_server_stop(self):73        self.mock_server_stop.return_value = None74        request = self.mock_rest_request(body='{"operation": "stop"}')75        response = nova.Server().post(request, 'MegaMan')76        self.assertStatusCode(response, 204)77        self.mock_server_stop.assert_called_once_with(request, 'MegaMan')78    @test.create_mocks({api.nova: ['server_pause']})79    def test_server_pause(self):80        self.mock_server_pause.return_value = None81        request = self.mock_rest_request(body='{"operation": "pause"}')82        response = nova.Server().post(request, 'MegaMan')83        self.assertStatusCode(response, 204)84        self.mock_server_pause.assert_called_once_with(request, 'MegaMan')85    @test.create_mocks({api.nova: ['server_unpause']})86    def test_server_unpause(self):87        self.mock_server_unpause.return_value = None88        request = self.mock_rest_request(body='{"operation": "unpause"}')89        response = nova.Server().post(request, 'MegaMan')90        self.assertStatusCode(response, 204)91        self.mock_server_unpause.assert_called_once_with(request, 'MegaMan')92    @test.create_mocks({api.nova: ['server_suspend']})93    def test_server_suspend(self):94        self.mock_server_suspend.return_value = None95        request = self.mock_rest_request(body='{"operation": "suspend"}')96        response = nova.Server().post(request, 'MegaMan')97        self.assertStatusCode(response, 204)98        self.mock_server_suspend.assert_called_once_with(request, 'MegaMan')99    @test.create_mocks({api.nova: ['server_resume']})100    def test_server_resume(self):101        self.mock_server_resume.return_value = None102        request = self.mock_rest_request(body='{"operation": "resume"}')103        response = nova.Server().post(request, 'MegaMan')104        self.assertStatusCode(response, 204)105        self.mock_server_resume.assert_called_once_with(request, 'MegaMan')106    @test.create_mocks({api.nova: ['server_reboot']})107    def test_server_hard_reboot(self):108        self.mock_server_reboot.return_value = None109        request = self.mock_rest_request(body='{"operation": "hard_reboot"}')110        response = nova.Server().post(request, 'MegaMan')111        self.assertStatusCode(response, 204)112        self.mock_server_reboot.assert_called_once_with(request, 'MegaMan',113                                                        False)114    @test.create_mocks({api.nova: ['server_reboot']})115    def test_server_soft_reboot(self):116        self.mock_server_reboot.return_value = None117        request = self.mock_rest_request(body='{"operation": "soft_reboot"}')118        response = nova.Server().post(request, 'MegaMan')119        self.assertStatusCode(response, 204)120        self.mock_server_reboot.assert_called_once_with(request, 'MegaMan',121                                                        True)122    #123    # Security Groups124    #125    @test.create_mocks({api.neutron: ['server_security_groups']})126    def test_securitygroups_list(self):127        request = self.mock_rest_request()128        self.mock_server_security_groups.return_value = [129            mock.Mock(**{'to_dict.return_value': {'id': '1'}}),130            mock.Mock(**{'to_dict.return_value': {'id': '2'}}),131        ]132        response = nova.SecurityGroups().get(request, 'MegaMan')133        self.assertStatusCode(response, 200)134        self.assertEqual(response.json, {'items': [{'id': '1'}, {'id': '2'}]})135        self.mock_server_security_groups.assert_called_once_with(request,136                                                                 'MegaMan')137    #138    # Console Output139    #140    @test.create_mocks({api.nova: ['server_console_output']})141    def test_console_output(self):142        request = self.mock_rest_request(body='{"length": 50}')143        self.mock_server_console_output.return_value = "this\nis\ncool"144        response = nova.ConsoleOutput().post(request, 'MegaMan')145        self.assertStatusCode(response, 200)146        self.assertEqual(response.json, {'lines': ["this", "is", "cool"]})147        self.mock_server_console_output.assert_called_once_with(request,148                                                                'MegaMan',149                                                                tail_length=50)150    #151    # Remote Console Info152    #153    @test.create_mocks({api.nova: ['server_serial_console']})154    def test_console_info(self):155        request = self.mock_rest_request(body='{"console_type": "SERIAL"}')156        retval = mock.Mock(**{"url": "http://here.com"})157        self.mock_server_serial_console.return_value = retval158        response = nova.RemoteConsoleInfo().post(request, 'MegaMan')159        self.assertStatusCode(response, 200)160        self.assertEqual(response.json,161                         {"type": "SERIAL", "url": "http://here.com"})162        self.mock_server_serial_console.assert_called_once_with(request,163                                                                'MegaMan')164    #165    # Volumes166    #167    @test.create_mocks({api.nova: ['instance_volumes_list']})168    def test_volumes_list(self):169        request = self.mock_rest_request()170        self.mock_instance_volumes_list.return_value = [171            mock.Mock(**{'to_dict.return_value': {'id': '1'}}),172            mock.Mock(**{'to_dict.return_value': {'id': '2'}}),173        ]174        response = nova.Volumes().get(request, 'MegaMan')175        self.assertStatusCode(response, 200)176        self.assertEqual(response.json, {'items': [{'id': '1'}, {'id': '2'}]})177        self.mock_instance_volumes_list.assert_called_once_with(request,178                                                                'MegaMan')179    #180    # Keypairs181    #182    @test.create_mocks({api.nova: ['keypair_list']})183    def test_keypair_list(self):184        request = self.mock_rest_request()185        self.mock_keypair_list.return_value = [186            mock.Mock(**{'to_dict.return_value': {'id': 'one'}}),187            mock.Mock(**{'to_dict.return_value': {'id': 'two'}}),188        ]189        response = nova.Keypairs().get(request)190        self.assertStatusCode(response, 200)191        self.assertEqual({"items": [{"id": "one"}, {"id": "two"}]},192                         response.json)193        self.mock_keypair_list.assert_called_once_with(request)194    @test.create_mocks({api.nova: ['keypair_create']})195    def test_keypair_create(self):196        request = self.mock_rest_request(body='''{"name": "Ni!",197                                                  "key_type": "ssh"}''')198        new = self.mock_keypair_create.return_value199        new.to_dict.return_value = {'name': 'Ni!',200                                    'key_type': 'ssh',201                                    'public_key': 'sekrit'}202        new.name = 'Ni!'203        with mock.patch.object(settings, 'DEBUG', True):204            response = nova.Keypairs().post(request)205        self.assertStatusCode(response, 201)206        self.assertEqual({"name": "Ni!",207                          "key_type": "ssh",208                          "public_key": "sekrit"},209                         response.json)210        self.assertEqual('/api/nova/keypairs/Ni%21', response['location'])211        self.mock_keypair_create.assert_called_once_with(request, 'Ni!', 'ssh')212    @test.create_mocks({api.nova: ['keypair_import']})213    def test_keypair_import(self):214        request = self.mock_rest_request(body='''215            {"name": "Ni!", "public_key": "hi", "key_type": "ssh"}216        ''')217        new = self.mock_keypair_import.return_value218        new.to_dict.return_value = {'name': 'Ni!',219                                    'public_key': 'hi',220                                    'key_type': 'ssh'}221        new.name = 'Ni!'222        with mock.patch.object(settings, 'DEBUG', True):223            response = nova.Keypairs().post(request)224        self.assertStatusCode(response, 201)225        self.assertEqual({"name": "Ni!",226                          "public_key": "hi",227                          "key_type": "ssh"},228                         response.json)229        self.assertEqual('/api/nova/keypairs/Ni%21', response['location'])230        self.mock_keypair_import.assert_called_once_with(request,231                                                         'Ni!',232                                                         'hi',233                                                         'ssh')234    @test.create_mocks({api.nova: ['keypair_get']})235    def test_keypair_get(self):236        request = self.mock_rest_request()237        self.mock_keypair_get.return_value.to_dict.return_value = {'name': '1'}238        response = nova.Keypair().get(request, '1')239        self.assertStatusCode(response, 200)240        self.assertEqual({"name": "1"},241                         response.json)242        self.mock_keypair_get.assert_called_once_with(request, "1")243    @test.create_mocks({api.nova: ['keypair_delete']})244    def test_keypair_delete(self):245        self.mock_keypair_delete.return_value = None246        request = self.mock_rest_request()247        nova.Keypair().delete(request, "1")248        self.mock_keypair_delete.assert_called_once_with(request, "1")249    #250    # Availability Zones251    #252    def test_availzone_get_brief(self):253        self._test_availzone_get(False)254    def test_availzone_get_detailed(self):255        self._test_availzone_get(True)256    @test.create_mocks({api.nova: ['availability_zone_list']})257    def _test_availzone_get(self, detail):258        if detail:259            request = self.mock_rest_request(GET={'detailed': 'true'})260        else:261            request = self.mock_rest_request(GET={})262        self.mock_availability_zone_list.return_value = [263            mock.Mock(**{'to_dict.return_value': {'id': 'one'}}),264            mock.Mock(**{'to_dict.return_value': {'id': 'two'}}),265        ]266        response = nova.AvailabilityZones().get(request)267        self.assertStatusCode(response, 200)268        self.assertEqual({"items": [{"id": "one"}, {"id": "two"}]},269                         response.json)270        self.mock_availability_zone_list.assert_called_once_with(request,271                                                                 detail)272    #273    # Limits274    #275    def test_limits_get_not_reserved(self):276        self._test_limits_get(False)277    def test_limits_get_reserved(self):278        self._test_limits_get(True)279    @test.create_mocks({api.nova: ['tenant_absolute_limits']})280    def _test_limits_get(self, reserved):281        if reserved:282            request = self.mock_rest_request(GET={'reserved': 'true'})283        else:284            request = self.mock_rest_request(GET={})285        self.mock_tenant_absolute_limits.return_value = {'id': 'one'}286        response = nova.Limits().get(request)287        self.assertStatusCode(response, 200)288        self.mock_tenant_absolute_limits.assert_called_once_with(request,289                                                                 reserved)290        self.assertEqual({"id": "one"}, response.json)291    #292    # Servers293    #294    def test_server_create_missing(self):295        request = self.mock_rest_request(body='''{"name": "hi"}''')296        response = nova.Servers().post(request)297        self.assertStatusCode(response, 400)298        self.assertEqual("missing required parameter 'source_id'",299                         response.json)300    @test.create_mocks({api.nova: ['server_create']})301    def test_server_create_basic(self):302        request = self.mock_rest_request(body='''{"name": "Ni!",303            "source_id": "image123", "flavor_id": "flavor123",304            "key_name": "sekrit", "user_data": "base64 yes",305            "security_groups": [{"name": "root"}]}306        ''')307        new = self.mock_server_create.return_value308        new.to_dict.return_value = {'id': 'server123'}309        new.id = 'server123'310        response = nova.Servers().post(request)311        self.assertStatusCode(response, 201)312        self.assertEqual({"id": "server123"}, response.json)313        self.assertEqual('/api/nova/servers/server123', response['location'])314        self.mock_server_create.assert_called_once_with(315            request, 'Ni!', 'image123', 'flavor123', 'sekrit', 'base64 yes',316            [{'name': 'root'}]317        )318    @test.create_mocks({api.nova: ['server_create']})319    def test_server_create_with_leading_trailing_space(self):320        request = self.mock_rest_request(body='''{"name": " Ni! ",321                "source_id": "image123", "flavor_id": "flavor123",322                "key_name": "sekrit", "user_data": "base64 yes",323                "security_groups": [{"name": "root"}]}324            ''')325        new = self.mock_server_create.return_value326        new.to_dict.return_value = {'name': ' Ni! '.strip()}327        new.id = str(uuid.uuid4())328        response = nova.Servers().post(request)329        self.assertStatusCode(response, 201)330        self.assertEqual({"name": "Ni!"}, response.json)331        self.mock_server_create.assert_called_once_with(332            request, ' Ni! ', 'image123', 'flavor123', 'sekrit', 'base64 yes',333            [{'name': 'root'}])334    @test.create_mocks({api.nova: ['server_list']})335    def test_server_list(self):336        request = self.mock_rest_request()337        self.mock_server_list.return_value = ([338            mock.Mock(**{'to_dict.return_value': {'id': 'one'}}),339            mock.Mock(**{'to_dict.return_value': {'id': 'two'}}),340        ], False)341        response = nova.Servers().get(request)342        self.assertStatusCode(response, 200)343        self.assertEqual({'items': [{'id': 'one'}, {'id': 'two'}]},344                         response.json)345        self.mock_server_list.assert_called_once_with(request)346    @test.create_mocks({api.nova: ['server_get']})347    def test_server_get_single(self):348        request = self.mock_rest_request()349        self.mock_server_get.return_value.to_dict.return_value = {'name': '1'}350        response = nova.Server().get(request, "1")351        self.assertStatusCode(response, 200)352        self.mock_server_get.assert_called_once_with(request, "1")353    #354    # Server Groups355    #356    @test.create_mocks({api.nova: ['server_group_list']})357    def test_server_group_list(self):358        request = self.mock_rest_request()359        self.mock_server_group_list.return_value = [360            mock.Mock(**{'to_dict.return_value': {'id': '1'}}),361            mock.Mock(**{'to_dict.return_value': {'id': '2'}}),362        ]363        response = nova.ServerGroups().get(request)364        self.assertStatusCode(response, 200)365        self.assertEqual({'items': [{'id': '1'}, {'id': '2'}]},366                         response.json)367        self.mock_server_group_list.assert_called_once_with(request)368    @test.create_mocks({api.nova: ['server_group_create']})369    def test_server_group_create(self):370        req_data = json.dumps({371            'name': 'server_group', 'policies': ['affinity']})372        self.mock_server_group_create.return_value = mock.Mock(**{373            'id': '123',374            'to_dict.return_value': {'id': '123',375                                     'name': 'server_group',376                                     'policies': ['affinity']}377        })378        server_group_data = {'name': 'server_group',379                             'policies': ['affinity']}380        request = self.mock_rest_request(body=req_data)381        response = nova.ServerGroups().post(request)382        self.assertStatusCode(response, 201)383        self.assertEqual('/api/nova/servergroups/123', response['location'])384        self.mock_server_group_create.assert_called_once_with(385            request, **server_group_data)386    @test.create_mocks({api.nova: ['server_group_delete']})387    def test_server_group_delete(self):388        request = self.mock_rest_request()389        self.mock_server_group_delete.return_value = None390        nova.ServerGroup().delete(request, "1")391        self.mock_server_group_delete.assert_called_once_with(request, "1")392    @test.create_mocks({api.nova: ['server_group_get']})393    def test_server_group_get_single(self):394        request = self.mock_rest_request()395        servergroup = self.server_groups.first()396        self.mock_server_group_get.return_value = servergroup397        response = nova.ServerGroup().get(request, "1")398        self.assertStatusCode(response, 200)399        self.assertEqual(servergroup.to_dict(), response.json)400        self.mock_server_group_get.assert_called_once_with(request, "1")401    #402    # Server Metadata403    #404    @test.create_mocks({api.nova: ['server_get']})405    def test_server_get_metadata(self):406        request = self.mock_rest_request()407        meta = {'foo': 'bar'}408        ret_val_server = self.mock_server_get.return_value409        ret_val_server.to_dict.return_value.get.return_value = meta410        response = nova.ServerMetadata().get(request, "1")411        self.assertStatusCode(response, 200)412        self.mock_server_get.assert_called_once_with(request, "1")413    @test.create_mocks({api.nova: ['server_metadata_delete',414                                   'server_metadata_update']})415    def test_server_edit_metadata(self):416        request = self.mock_rest_request(417            body='{"updated": {"a": "1", "b": "2"}, "removed": ["c", "d"]}'418        )419        self.mock_server_metadata_update.return_value = None420        self.mock_server_metadata_delete.return_value = None421        response = nova.ServerMetadata().patch(request, '1')422        self.assertStatusCode(response, 204)423        self.assertEqual(b'', response.content)424        self.mock_server_metadata_update.assert_called_once_with(425            request, '1', {'a': '1', 'b': '2'}426        )427        self.mock_server_metadata_delete.assert_called_once_with(428            request, '1', ['c', 'd']429        )430    #431    # Extensions432    #433    @test.create_mocks({api.nova: ['list_extensions']})434    def test_extension_list(self):435        request = self.mock_rest_request()436        self.mock_list_extensions.return_value = ['foo', 'bar']437        response = nova.Extensions().get(request)438        self.assertStatusCode(response, 200)439        self.assertEqual({"items": [{"name": "foo"}, {"name": "bar"}]},440                         response.json)441        self.mock_list_extensions.assert_called_once_with(request)442    #443    # Flavors444    #445    @test.create_mocks({api.nova: ['flavor_get', 'flavor_access_list']})446    def test_flavor_get_single_with_access_list(self):447        request = self.mock_rest_request(GET={'get_access_list': 'tRuE'})448        self.mock_flavor_get.return_value.to_dict.return_value = {'name': '1'}449        self.mock_flavor_get.return_value.is_public = False450        self.mock_flavor_access_list.return_value = [451            mock.Mock(**{'tenant_id': '11'}),452            mock.Mock(**{'tenant_id': '22'}),453        ]454        response = nova.Flavor().get(request, "1")455        self.assertStatusCode(response, 200)456        self.assertEqual(to_json(response.content.decode('utf-8')),457                         to_json('{"access-list": ["11", "22"], "name": "1"}'))458        self.mock_flavor_get.assert_called_once_with(request, "1",459                                                     get_extras=False)460        self.mock_flavor_access_list.assert_called_once_with(request, "1")461    def test_get_extras_no(self):462        self._test_flavor_get_single(get_extras=False)463    def test_get_extras_yes(self):464        self._test_flavor_get_single(get_extras=True)465    def test_get_extras_default(self):466        self._test_flavor_get_single(get_extras=None)467    @test.create_mocks({api.nova: ['flavor_get']})468    def _test_flavor_get_single(self, get_extras):469        if get_extras:470            request = self.mock_rest_request(GET={'get_extras': 'tRuE'})471        elif get_extras is None:472            request = self.mock_rest_request()473            get_extras = False474        else:475            request = self.mock_rest_request(GET={'get_extras': 'fAlsE'})476        self.mock_flavor_get.return_value.to_dict.return_value = {'name': '1'}477        response = nova.Flavor().get(request, "1")478        self.assertStatusCode(response, 200)479        if get_extras:480            self.assertEqual(response.json, {"extras": {}, "name": "1"})481        else:482            self.assertEqual({"name": "1"}, response.json)483        self.mock_flavor_get.assert_called_once_with(request, "1",484                                                     get_extras=get_extras)485    @test.create_mocks({api.nova: ['flavor_get']})486    def test_flavor_get_single_with_swap_set_to_empty(self):487        request = self.mock_rest_request()488        self.mock_flavor_get.return_value\489            .to_dict.return_value = {'name': '1', 'swap': ''}490        response = nova.Flavor().get(request, "1")491        self.assertStatusCode(response, 200)492        self.assertEqual(to_json(response.content.decode('utf-8')),493                         to_json('{"name": "1", "swap": 0}'))494        self.mock_flavor_get.assert_called_once_with(request, '1',495                                                     get_extras=False)496    @test.create_mocks({api.nova: ['flavor_delete']})497    def test_flavor_delete(self):498        self.mock_flavor_delete.return_value = None499        request = self.mock_rest_request()500        nova.Flavor().delete(request, "1")501        self.mock_flavor_delete.assert_called_once_with(request, "1")502    @test.create_mocks({api.nova: ['flavor_create']})503    def test_flavor_create(self):504        flavor_req_data = '{"name": "flavor", ' \505                          '"ram": 12, ' \506                          '"vcpus": 1, ' \507                          '"disk": 2, ' \508                          '"OS-FLV-EXT-DATA:ephemeral": 3, ' \509                          '"swap": 4, ' \510                          '"id": "123"' \511                          '}'512        self.mock_flavor_create.return_value = mock.Mock(**{513            'id': '123',514            'to_dict.return_value': {'id': '123', 'name': 'flavor'}515        })516        flavor_data = {'name': 'flavor',517                       'memory': 12,518                       'vcpu': 1,519                       'disk': 2,520                       'ephemeral': 3,521                       'swap': 4,522                       'flavorid': '123',523                       'is_public': True}524        request = self.mock_rest_request(body=flavor_req_data)525        response = nova.Flavors().post(request)526        self.assertStatusCode(response, 201)527        self.assertEqual('/api/nova/flavors/123', response['location'])528        self.mock_flavor_create.assert_called_once_with(request, **flavor_data)529    @test.create_mocks({api.nova: ['flavor_create',530                                   'add_tenant_to_flavor']})531    def test_flavor_create_with_access_list(self):532        flavor_req_data = '{"name": "flavor", ' \533                          '"ram": 12, ' \534                          '"vcpus": 1, ' \535                          '"disk": 2, ' \536                          '"OS-FLV-EXT-DATA:ephemeral": 3, ' \537                          '"swap": 4, ' \538                          '"id": "123", ' \539                          '"flavor_access": [{"id":"1", "name":"test"}]' \540                          '}'541        self.mock_flavor_create.return_value = mock.Mock(**{542            'id': '1234',543            'to_dict.return_value': {'id': '1234', 'name': 'flavor'}544        })545        # A list of FlavorAccess object is returned but it is actually unused.546        self.mock_add_tenant_to_flavor.return_value = [547            mock.sentinel.flavor_access1,548        ]549        flavor_data = {'name': 'flavor',550                       'memory': 12,551                       'vcpu': 1,552                       'disk': 2,553                       'ephemeral': 3,554                       'swap': 4,555                       'flavorid': '123',556                       'is_public': False}557        request = self.mock_rest_request(body=flavor_req_data)558        response = nova.Flavors().post(request)559        self.assertStatusCode(response, 201)560        self.assertEqual('/api/nova/flavors/1234', response['location'])561        self.mock_flavor_create.assert_called_once_with(request, **flavor_data)562        self.mock_add_tenant_to_flavor.assert_called_once_with(563            request, '1234', '1')564    @test.create_mocks({api.nova: ['flavor_create',565                                   'flavor_delete',566                                   'flavor_get_extras']})567    def test_flavor_update(self):568        flavor_req_data = '{"name": "flavor", ' \569                          '"ram": 12, ' \570                          '"vcpus": 1, ' \571                          '"disk": 2, ' \572                          '"OS-FLV-EXT-DATA:ephemeral": 3, ' \573                          '"swap": 4' \574                          '}'575        self.mock_flavor_get_extras.return_value = {}576        self.mock_flavor_create.return_value = mock.Mock(**{577            'id': '123',578            'to_dict.return_value': {'id': '123', 'name': 'flavor'}579        })580        self.mock_flavor_delete.return_value = None581        flavor_data = {'name': 'flavor',582                       'memory': 12,583                       'vcpu': 1,584                       'disk': 2,585                       'ephemeral': 3,586                       'swap': 4,587                       'flavorid': '123',588                       'is_public': True}589        request = self.mock_rest_request(body=flavor_req_data)590        response = nova.Flavor().patch(request, '123')591        self.assertStatusCode(response, 204)592        self.mock_flavor_get_extras.assert_called_once_with(593            request, '123', raw=True)594        self.mock_flavor_delete.assert_called_once_with(request, '123')595        self.mock_flavor_create.assert_called_once_with(request, **flavor_data)596    @test.create_mocks({api.nova: ['flavor_create', 'flavor_delete',597                                   'flavor_extra_set', 'flavor_get_extras']})598    def test_flavor_update_with_extras(self):599        flavor_req_data = '{"name": "flavor", ' \600                          '"ram": 12, ' \601                          '"vcpus": 1, ' \602                          '"disk": 2, ' \603                          '"OS-FLV-EXT-DATA:ephemeral": 3, ' \604                          '"swap": 4' \605                          '}'606        extra_dict = mock.Mock()607        self.mock_flavor_get_extras.return_value = extra_dict608        self.mock_flavor_create.return_value = mock.Mock(**{609            'id': '1234',610            'to_dict.return_value': {'id': '1234', 'name': 'flavor'}611        })612        self.mock_flavor_delete.return_value = None613        self.mock_flavor_extra_set.return_value = None614        flavor_data = {'name': 'flavor',615                       'memory': 12,616                       'vcpu': 1,617                       'disk': 2,618                       'ephemeral': 3,619                       'swap': 4,620                       'flavorid': '123',621                       'is_public': True}622        request = self.mock_rest_request(body=flavor_req_data)623        response = nova.Flavor().patch(request, '123')624        self.assertStatusCode(response, 204)625        self.mock_flavor_delete.assert_called_once_with(request, '123')626        self.mock_flavor_create.assert_called_once_with(request, **flavor_data)627        self.mock_flavor_get_extras.assert_called_once_with(request, '123',628                                                            raw=True)629        self.mock_flavor_extra_set.assert_called_once_with(request, '1234',630                                                           extra_dict)631    @test.create_mocks({api.nova: ['flavor_create',632                                   'flavor_delete',633                                   'flavor_get_extras',634                                   'add_tenant_to_flavor']})635    def test_flavor_update_with_access_list(self):636        flavor_req_data = '{"name": "flavor", ' \637                          '"ram": 12, ' \638                          '"vcpus": 1, ' \639                          '"disk": 2, ' \640                          '"OS-FLV-EXT-DATA:ephemeral": 3, ' \641                          '"swap": 4, ' \642                          '"flavor_access": [{"id":"1", "name":"test"}]' \643                          '}'644        self.mock_flavor_get_extras.return_value = {}645        self.mock_flavor_create.return_value = mock.Mock(**{646            'id': '1234',647            'to_dict.return_value': {'id': '1234', 'name': 'flavor'}648        })649        self.mock_flavor_delete.return_value = None650        # A list of FlavorAccess object is returned but it is actually unused.651        self.mock_add_tenant_to_flavor.return_value = [652            mock.sentinel.flavor_access1,653        ]654        flavor_data = {'name': 'flavor',655                       'memory': 12,656                       'vcpu': 1,657                       'disk': 2,658                       'ephemeral': 3,659                       'swap': 4,660                       'flavorid': '123',661                       'is_public': False}662        request = self.mock_rest_request(body=flavor_req_data)663        response = nova.Flavor().patch(request, '123')664        self.assertStatusCode(response, 204)665        self.mock_flavor_get_extras.assert_called_once_with(666            request, '123', raw=True)667        self.mock_flavor_delete.assert_called_once_with(request, '123')668        self.mock_flavor_create.assert_called_once_with(request, **flavor_data)669        self.mock_add_tenant_to_flavor.assert_called_once_with(670            request, '1234', '1')671    @test.create_mocks({api.nova: ['flavor_list']})672    def _test_flavor_list_public(self, is_public=None):673        if is_public:674            request = self.mock_rest_request(GET={'is_public': 'tRuE'})675        elif is_public is None:676            request = self.mock_rest_request(GET={})677        else:678            request = self.mock_rest_request(GET={'is_public': 'fAlsE'})679        self.mock_flavor_list.return_value = [680            FakeFlavor("1"), FakeFlavor("2")681        ]682        response = nova.Flavors().get(request)683        self.assertStatusCode(response, 200)684        self.assertEqual({"items": [{"id": "1"}, {"id": "2"}]},685                         response.json)686        self.mock_flavor_list.assert_called_once_with(request,687                                                      is_public=is_public,688                                                      get_extras=False)689    def test_flavor_list_private(self):690        self._test_flavor_list_public(is_public=False)691    def test_flavor_list_public(self):692        self._test_flavor_list_public(is_public=True)693    def test_flavor_list_public_none(self):694        self._test_flavor_list_public(is_public=None)695    @test.create_mocks({api.nova: ['flavor_list']})696    def _test_flavor_list_extras(self, get_extras=None):697        if get_extras:698            request = self.mock_rest_request(GET={'get_extras': 'tRuE'})699        elif get_extras is None:700            request = self.mock_rest_request(GET={})701            get_extras = False702        else:703            request = self.mock_rest_request(GET={'get_extras': 'fAlsE'})704        self.mock_flavor_list.return_value = [705            FakeFlavor("1"), FakeFlavor("2")706        ]707        response = nova.Flavors().get(request)708        self.assertStatusCode(response, 200)709        if get_extras:710            self.assertEqual({"items": [{"extras": {}, "id": "1"},711                                        {"extras": {}, "id": "2"}]},712                             response.json)713        else:714            self.assertEqual({"items": [{"id": "1"}, {"id": "2"}]},715                             response.json)716        self.mock_flavor_list.assert_called_once_with(request, is_public=None,717                                                      get_extras=get_extras)718    def test_flavor_list_extras_no(self):719        self._test_flavor_list_extras(get_extras=False)720    def test_flavor_list_extras_yes(self):721        self._test_flavor_list_extras(get_extras=True)722    def test_flavor_list_extras_absent(self):723        self._test_flavor_list_extras(get_extras=None)724    @test.create_mocks({api.nova: ['flavor_get_extras']})725    def test_flavor_get_extra_specs(self):726        request = self.mock_rest_request()727        self.mock_flavor_get_extras.return_value.to_dict.return_value = \728            {'foo': '1'}729        response = nova.FlavorExtraSpecs().get(request, "1")730        self.assertStatusCode(response, 200)731        self.mock_flavor_get_extras.assert_called_once_with(request, "1",732                                                            raw=True)733    @test.create_mocks({api.nova: ['flavor_extra_delete',734                                   'flavor_extra_set']})735    def test_flavor_edit_extra_specs(self):736        request = self.mock_rest_request(737            body='{"updated": {"a": "1", "b": "2"}, "removed": ["c", "d"]}'738        )739        self.mock_flavor_extra_delete.return_value = None740        self.mock_flavor_extra_set.return_value = {'a': '1', 'b': '2'}741        response = nova.FlavorExtraSpecs().patch(request, '1')742        self.assertStatusCode(response, 204)743        self.assertEqual(b'', response.content)744        self.mock_flavor_extra_set.assert_called_once_with(745            request, '1', {'a': '1', 'b': '2'}746        )747        self.mock_flavor_extra_delete.assert_called_once_with(748            request, '1', ['c', 'd']749        )750    @test.create_mocks({api.nova: ['aggregate_get']})751    def test_aggregate_get_extra_specs(self):752        request = self.mock_rest_request()753        self.mock_aggregate_get.return_value.metadata = {'a': '1', 'b': '2'}754        response = nova.AggregateExtraSpecs().get(request, "1")755        self.assertStatusCode(response, 200)756        self.assertEqual({"a": "1", "b": "2"}, response.json)757        self.mock_aggregate_get.assert_called_once_with(request, "1")758    @test.create_mocks({api.nova: ['aggregate_set_metadata']})759    def test_aggregate_edit_extra_specs(self):760        self.mock_aggregate_set_metadata.return_value = self.aggregates.first()761        request = self.mock_rest_request(762            body='{"updated": {"a": "1", "b": "2"}, "removed": ["c", "d"]}'763        )764        response = nova.AggregateExtraSpecs().patch(request, '1')765        self.assertStatusCode(response, 204)766        self.assertEqual(b'', response.content)767        self.mock_aggregate_set_metadata.assert_called_once_with(768            request, '1', {'a': '1', 'b': '2', 'c': None, 'd': None}769        )770    #771    # Services772    #773    @test.create_mocks({api.base: ['is_service_enabled'],774                        api.nova: ['service_list',775                                   'extension_supported']})776    def test_services_get(self):777        request = self.mock_rest_request(GET={})778        self.mock_service_list.return_value = [779            mock.Mock(**{'to_dict.return_value': {'id': '1'}}),780            mock.Mock(**{'to_dict.return_value': {'id': '2'}})781        ]782        self.mock_is_service_enabled.return_value = True783        self.mock_extension_supported.return_value = True784        response = nova.Services().get(request)785        self.assertStatusCode(response, 200)786        self.assertEqual('{"items": [{"id": "1"}, {"id": "2"}]}',787                         response.content.decode('utf-8'))788        self.mock_service_list.assert_called_once_with(request)789        self.mock_is_service_enabled.assert_called_once_with(request,790                                                             'compute')791        self.mock_extension_supported.assert_called_once_with(792            'Services', request)793    @mock.patch.object(api.base, 'is_service_enabled')794    def test_services_get_disabled(self, mock_is_service_enabled):795        request = self.mock_rest_request(GET={})796        mock_is_service_enabled.return_value = False797        response = nova.Services().get(request)798        self.assertStatusCode(response, 501)799        mock_is_service_enabled.assert_called_once_with(request, 'compute')800    @test.create_mocks({api.base: ['is_service_enabled'],801                        quotas: ['get_disabled_quotas'],802                        api.nova: ['default_quota_get']})803    def test_quota_sets_defaults_get(self):804        filters = {'user': {'tenant_id': 'tenant'}}805        request = self.mock_rest_request(**{'GET': dict(filters)})806        self.mock_is_service_enabled.return_value = True807        self.mock_get_disabled_quotas.return_value = ['floating_ips']808        self.mock_default_quota_get.return_value = [809            Quota('metadata_items', 100),810            Quota('floating_ips', 1),811            Quota('q2', 101)812        ]813        response = nova.DefaultQuotaSets().get(request)814        self.assertStatusCode(response, 200)815        self.assertEqual(response.json,816                         {"items": [817                             {"limit": 100,818                              "display_name": "Metadata Items",819                              "name": "metadata_items"},820                             {"limit": 101,821                              "display_name": "Q2",822                              "name": "q2"}823                         ]})824        self.mock_is_service_enabled.assert_called_once_with(request,825                                                             'compute')826        self.mock_get_disabled_quotas.assert_called_once_with(request)827        self.mock_default_quota_get.assert_called_once_with(828            request, request.user.tenant_id)829    @test.create_mocks({api.base: ['is_service_enabled']})830    def test_quota_sets_defaults_get_when_service_is_disabled(self):831        filters = {'user': {'tenant_id': 'tenant'}}832        request = self.mock_rest_request(**{'GET': dict(filters)})833        self.mock_is_service_enabled.return_value = False834        response = nova.DefaultQuotaSets().get(request)835        self.assertStatusCode(response, 501)836        self.assertEqual(response.content.decode('utf-8'),837                         '"Service Nova is disabled."')838        self.mock_is_service_enabled.assert_called_once_with(request,839                                                             'compute')840    @test.create_mocks({api.base: ['is_service_enabled'],841                        quotas: ['get_disabled_quotas'],842                        api.nova: ['default_quota_update']})843    def test_quota_sets_defaults_patch(self):844        request = self.mock_rest_request(body='''845            {"key_pairs": "15", "metadata_items": "5000",846            "cores": "10", "instances": "20", "floating_ips": 10,847            "injected_file_content_bytes": "15",848            "injected_file_path_bytes": "5000",849            "injected_files": "5", "ram": "10", "gigabytes": "5"}850        ''')851        self.mock_is_service_enabled.return_value = True852        self.mock_get_disabled_quotas.return_value = ['floating_ips']853        self.mock_default_quota_update.return_value = None854        response = nova.DefaultQuotaSets().patch(request)855        self.assertStatusCode(response, 204)856        self.assertEqual(response.content.decode('utf-8'), '')857        self.mock_is_service_enabled.assert_called_once_with(request,858                                                             'compute')859        self.mock_get_disabled_quotas.assert_called_once_with(request)860        self.mock_default_quota_update.assert_called_once_with(861            request, key_pairs='15',862            metadata_items='5000', cores='10',863            instances='20', injected_file_content_bytes='15',864            injected_file_path_bytes='5000',865            injected_files='5', ram='10')866    @test.create_mocks({api.base: ['is_service_enabled']})867    def test_quota_sets_defaults_patch_when_service_is_disabled(self):868        request = self.mock_rest_request(body='''869            {"key_pairs": "15", "metadata_items": "5000",870            "cores": "10", "instances": "20", "floating_ips": 10,871            "injected_file_content_bytes": "15",872            "injected_file_path_bytes": "5000",873            "injected_files": "5", "ram": "10", "gigabytes": "5"}874        ''')875        self.mock_is_service_enabled.return_value = False876        response = nova.DefaultQuotaSets().patch(request)877        self.assertStatusCode(response, 501)878        self.assertEqual(response.content.decode('utf-8'),879                         '"Service Nova is disabled."')880        self.mock_is_service_enabled.assert_called_once_with(request,881                                                             'compute')882    @test.create_mocks({quotas: ['get_disabled_quotas']})883    def test_editable_quotas_get(self):884        disabled_quotas = {'floating_ips', 'fixed_ips',885                           'security_groups', 'security_group_rules'}886        editable_quotas = {'cores', 'volumes', 'network', 'fixed_ips'}887        self.mock_get_disabled_quotas.return_value = disabled_quotas888        request = self.mock_rest_request()889        with mock.patch.object(quotas, 'QUOTA_FIELDS', editable_quotas):890            response = nova.EditableQuotaSets().get(request)891        self.assertStatusCode(response, 200)892        # NOTE(amotoki): assertItemsCollectionEqual cannot be used below893        # since the item list is generated from a set and the order of items894        # is unpredictable.895        self.assertEqual(set(response.json['items']),896                         {'cores', 'volumes', 'network'})897        self.mock_get_disabled_quotas.assert_called_once_with(request)898    @test.create_mocks({api.base: ['is_service_enabled'],899                        quotas: ['get_disabled_quotas'],900                        api.nova: ['tenant_quota_update']})901    def test_quota_sets_patch(self):902        quota_data = dict(cores='15', instances='5',903                          ram='50000', metadata_items='150',904                          injected_files='5',905                          injected_file_content_bytes='10240',906                          floating_ips='50', fixed_ips='5',907                          security_groups='10',908                          security_group_rules='100')909        request = self.mock_rest_request(body='''910            {"cores": "15", "ram": "50000", "instances": "5",911             "metadata_items": "150", "injected_files": "5",912             "injected_file_content_bytes": "10240", "floating_ips": "50",913             "fixed_ips": "5", "security_groups": "10" ,914             "security_group_rules": "100", "volumes": "10"}915        ''')916        self.mock_get_disabled_quotas.return_value = set()917        self.mock_is_service_enabled.return_value = True918        self.mock_tenant_quota_update.return_value = None919        with mock.patch.object(quotas, 'NOVA_QUOTA_FIELDS',920                               {n for n in quota_data}):921            response = nova.QuotaSets().patch(request, 'spam123')922        self.assertStatusCode(response, 204)923        self.assertEqual(response.content.decode('utf-8'), '')924        self.mock_is_service_enabled.assert_called_once_with(request,925                                                             'compute')926        self.mock_get_disabled_quotas.assert_called_once_with(request)927        self.mock_tenant_quota_update.assert_called_once_with(928            request, 'spam123', **quota_data)929    @test.create_mocks({api.nova: ['tenant_quota_update'],930                        api.base: ['is_service_enabled'],931                        quotas: ['get_disabled_quotas']})932    def test_quota_sets_patch_when_service_is_disabled(self):933        quota_data = dict(cores='15', instances='5',934                          ram='50000', metadata_items='150',935                          injected_files='5',936                          injected_file_content_bytes='10240',937                          floating_ips='50', fixed_ips='5',938                          security_groups='10',939                          security_group_rules='100')940        request = self.mock_rest_request(body='''941            {"cores": "15", "ram": "50000", "instances": "5",942             "metadata_items": "150", "injected_files": "5",943             "injected_file_content_bytes": "10240", "floating_ips": "50",944             "fixed_ips": "5", "security_groups": "10" ,945             "security_group_rules": "100", "volumes": "10"}946        ''')947        self.mock_get_disabled_quotas.return_value = {}948        self.mock_is_service_enabled.return_value = False949        self.mock_tenant_quota_update.return_value = None950        with mock.patch.object(quotas, 'NOVA_QUOTA_FIELDS',951                               {n for n in quota_data}):952            response = nova.QuotaSets().patch(request, 'spam123')953        self.assertStatusCode(response, 501)954        self.assertEqual(response.content.decode('utf-8'),955                         '"Service Nova is disabled."')956        self.mock_get_disabled_quotas.assert_called_once_with(request)957        self.mock_is_service_enabled.assert_called_once_with(request,958                                                             'compute')959        self.mock_tenant_quota_update.assert_not_called()960    @test.create_mocks({api.nova: ['is_feature_available']})961    def test_version_get(self):962        request = self.mock_rest_request()963        self.mock_is_feature_available.return_value = True964        response = nova.Features().get(request, 'fake')965        self.assertStatusCode(response, 200)966        self.assertEqual(response.content.decode('utf-8'), 'true')967        self.mock_is_feature_available.assert_called_once_with(request,...

test_keystone.py

Source:test_keystone.py

1# Copyright 2014, Rackspace, US, Inc.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain 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,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14from django.conf import settings15import mock16from oslo_serialization import jsonutils17from openstack_dashboard import api18from openstack_dashboard.api.rest import keystone19from openstack_dashboard.test import helpers as test20class KeystoneRestTestCase(test.TestCase):21    #22    # Version23    #24    @test.create_mocks({api.keystone: ['get_version']})25    def test_version_get(self):26        request = self.mock_rest_request()27        self.mock_get_version.return_value = '3'28        response = keystone.Version().get(request)29        self.assertStatusCode(response, 200)30        self.assertEqual(response.json, {"version": "3"})31        self.mock_get_version.assert_called_once_with()32    #33    # Users34    #35    @test.create_mocks({api.keystone: ['user_get']})36    def test_user_get(self):37        request = self.mock_rest_request()38        self.mock_user_get.return_value.to_dict.return_value = {'name': 'Ni!'}39        response = keystone.User().get(request, 'the_id')40        self.assertStatusCode(response, 200)41        self.assertEqual(response.json, {"name": "Ni!"})42        self.mock_user_get.assert_called_once_with(43            request, 'the_id', admin=False)44    @test.create_mocks({api.keystone: ['user_get']})45    def test_user_get_current(self):46        request = self.mock_rest_request(**{'user.id': 'current_id'})47        self.mock_user_get.return_value.to_dict.return_value = {'name': 'Ni!'}48        response = keystone.User().get(request, 'current')49        self.assertStatusCode(response, 200)50        self.assertEqual(response.json, {"name": "Ni!"})51        self.mock_user_get.assert_called_once_with(52            request, 'current_id', admin=False)53    @test.create_mocks({api.keystone: ['user_list']})54    def test_user_get_list(self):55        request = self.mock_rest_request(**{56            'session.get': mock.Mock(return_value='the_domain'),57            'GET': {},58        })59        self.mock_user_list.return_value = [60            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),61            mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})62        ]63        response = keystone.Users().get(request)64        self.assertStatusCode(response, 200)65        self.assertEqual(response.json,66                         {"items": [{"name": "Ni!"}, {"name": "Ptang!"}]})67        self.mock_user_list.assert_called_once_with(request, project=None,68                                                    domain='the_domain',69                                                    group=None,70                                                    filters=None)71    @test.create_mocks({api.keystone: ['user_list']})72    def test_user_get_list_with_filters(self):73        filters = {'enabled': True}74        request = self.mock_rest_request(**{75            'session.get': mock.Mock(return_value='the_domain'),76            'GET': dict(**filters),77        })78        self.mock_user_list.return_value = [79            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),80            mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})81        ]82        response = keystone.Users().get(request)83        self.assertStatusCode(response, 200)84        self.assertEqual(response.json,85                         {"items": [{"name": "Ni!"}, {"name": "Ptang!"}]})86        self.mock_user_list.assert_called_once_with(request, project=None,87                                                    domain='the_domain',88                                                    group=None,89                                                    filters=filters)90    def test_user_create_full(self):91        self._test_user_create(92            '{"name": "bob", '93            '"password": "sekrit", "project": "123", '94            '"email": "spam@company.example", '95            '"description": "hello, puff"}',96            {97                'name': 'bob',98                'password': 'sekrit',99                'email': 'spam@company.example',100                'project': '123',101                'domain': 'the_domain',102                'enabled': True,103                'description': 'hello, puff'104            }105        )106    def test_user_create_existing_role(self):107        self._test_user_create(108            '{"name": "bob", '109            '"password": "sekrit", "project": "123", '110            '"email": "spam@company.example"}',111            {112                'name': 'bob',113                'password': 'sekrit',114                'email': 'spam@company.example',115                'project': '123',116                'domain': 'the_domain',117                'enabled': True,118                'description': None119            }120        )121    def test_user_create_no_project(self):122        self._test_user_create(123            '{"name": "bob", '124            '"password": "sekrit", "project": "", '125            '"email": "spam@company.example"}',126            {127                'name': 'bob',128                'password': 'sekrit',129                'email': 'spam@company.example',130                'project': None,131                'domain': 'the_domain',132                'enabled': True,133                'description': None134            }135        )136    def test_user_create_partial(self):137        self._test_user_create(138            '{"name": "bob", "project": ""}',139            {140                'name': 'bob',141                'password': None,142                'email': None,143                'project': None,144                'domain': 'the_domain',145                'enabled': True,146                'description': None147            }148        )149    @test.create_mocks({api.keystone: ['get_default_domain',150                                       'user_create']})151    def _test_user_create(self, supplied_body, add_user_call):152        request = self.mock_rest_request(body=supplied_body)153        self.mock_get_default_domain.return_value = \154            mock.Mock(**{'id': 'the_domain'})155        self.mock_user_create.return_value = mock.Mock(**{156            'id': 'user123',157            'to_dict.return_value': {'id': 'user123', 'name': 'bob'}158        })159        response = keystone.Users().post(request)160        self.assertStatusCode(response, 201)161        self.assertEqual(response['location'],162                         '/api/keystone/users/user123')163        self.assertEqual(response.json,164                         {"id": "user123", "name": "bob"})165        self.mock_user_create.assert_called_once_with(request, **add_user_call)166        self.mock_get_default_domain.assert_called_once_with(request)167    @test.create_mocks({api.keystone: ['user_delete']})168    def test_user_delete_many(self):169        request = self.mock_rest_request(body='''170            ["id1", "id2", "id3"]171        ''')172        self.mock_user_delete.return_value = None173        response = keystone.Users().delete(request)174        self.assertStatusCode(response, 204)175        self.assertEqual(response.content, b'')176        self.mock_user_delete.assert_has_calls([177            mock.call(request, 'id1'),178            mock.call(request, 'id2'),179            mock.call(request, 'id3'),180        ])181    @test.create_mocks({api.keystone: ['user_delete']})182    def test_user_delete(self):183        request = self.mock_rest_request()184        self.mock_user_delete.return_value = None185        response = keystone.User().delete(request, 'the_id')186        self.assertStatusCode(response, 204)187        self.assertEqual(response.content, b'')188        self.mock_user_delete.assert_called_once_with(request, 'the_id')189    @test.create_mocks({api.keystone: ['user_get',190                                       'user_update_password']})191    def test_user_patch_password(self):192        request = self.mock_rest_request(body='''193            {"password": "sekrit"}194        ''')195        user = keystone.User()196        self.mock_user_get.return_value = mock.sentinel.user197        self.mock_user_update_password.return_value = None198        response = user.patch(request, 'user123')199        self.assertStatusCode(response, 204)200        self.assertEqual(response.content, b'')201        self.mock_user_get.assert_called_once_with(request, 'user123')202        self.mock_user_update_password.assert_called_once_with(203            request, mock.sentinel.user, 'sekrit')204    @test.create_mocks({api.keystone: ['user_get',205                                       'user_update_enabled']})206    def test_user_patch_enabled(self):207        request = self.mock_rest_request(body='''208            {"enabled": false}209        ''')210        user = keystone.User()211        self.mock_user_get.return_value = mock.sentinel.user212        self.mock_user_update_enabled.return_value = None213        response = user.patch(request, 'user123')214        self.assertStatusCode(response, 204)215        self.assertEqual(response.content, b'')216        self.mock_user_get.assert_called_once_with(request, 'user123')217        self.mock_user_update_enabled.assert_called_once_with(218            request, mock.sentinel.user, False)219    @test.create_mocks({api.keystone: ['user_get',220                                       'user_update']})221    def test_user_patch_project(self):222        request = self.mock_rest_request(body='''223            {"project": "other123"}224        ''')225        user = keystone.User()226        self.mock_user_get.return_value = mock.sentinel.user227        self.mock_user_update.return_value = self.users.first()228        response = user.patch(request, 'user123')229        self.assertStatusCode(response, 204)230        self.assertEqual(response.content, b'')231        self.mock_user_get.assert_called_once_with(request, 'user123')232        self.mock_user_update.assert_called_once_with(233            request, mock.sentinel.user, project='other123')234    @test.create_mocks({api.keystone: ['user_get',235                                       'user_update']})236    def test_user_patch_multiple(self):237        request = self.mock_rest_request(body='''238            {"project": "other123", "name": "something"}239        ''')240        user = keystone.User()241        self.mock_user_get.return_value = mock.sentinel.user242        self.mock_user_update.return_value = self.users.first()243        response = user.patch(request, 'user123')244        self.assertStatusCode(response, 204)245        self.assertEqual(response.content, b'')246        self.mock_user_get.assert_called_once_with(request, 'user123')247        self.mock_user_update.assert_called_once_with(248            request, mock.sentinel.user, project='other123', name='something')249    #250    # Roles251    #252    @test.create_mocks({api.keystone: ['role_get']})253    def test_role_get(self):254        request = self.mock_rest_request()255        self.mock_role_get.return_value.to_dict.return_value = {'name': 'Ni!'}256        response = keystone.Role().get(request, 'the_id')257        self.assertStatusCode(response, 200)258        self.assertEqual(response.json, {"name": "Ni!"})259        self.mock_role_get.assert_called_once_with(request, 'the_id')260    @test.create_mocks({api.keystone: ['get_default_role']})261    def test_role_get_default(self):262        request = self.mock_rest_request()263        ret_val_role = self.mock_get_default_role.return_value264        ret_val_role.to_dict.return_value = {'name': 'Ni!'}265        response = keystone.Role().get(request, 'default')266        self.assertStatusCode(response, 200)267        self.assertEqual(response.json, {"name": "Ni!"})268        self.mock_get_default_role.assert_called_once_with(request)269    @test.create_mocks({api.keystone: ['role_list']})270    def test_role_get_list(self):271        request = self.mock_rest_request(**{'GET': {}})272        self.mock_role_list.return_value = [273            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),274            mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})275        ]276        response = keystone.Roles().get(request)277        self.assertStatusCode(response, 200)278        self.assertEqual(response.json,279                         {"items": [{"name": "Ni!"}, {"name": "Ptang!"}]})280        self.mock_role_list.assert_called_once_with(request)281    @test.create_mocks({api.keystone: ['roles_for_user']})282    def test_role_get_for_user(self):283        request = self.mock_rest_request(**{'GET': {'user_id': 'user123',284                                         'project_id': 'project123'}})285        self.mock_roles_for_user.return_value = [286            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),287            mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})288        ]289        response = keystone.Roles().get(request)290        self.assertStatusCode(response, 200)291        self.assertEqual(response.json,292                         {"items": [{"name": "Ni!"}, {"name": "Ptang!"}]})293        self.mock_roles_for_user.assert_called_once_with(request, 'user123',294                                                         'project123')295    @test.create_mocks({api.keystone: ['role_create']})296    def test_role_create(self):297        request = self.mock_rest_request(body='''298            {"name": "bob"}299        ''')300        self.mock_role_create.return_value.id = 'role123'301        self.mock_role_create.return_value.to_dict.return_value = {302            'id': 'role123', 'name': 'bob'303        }304        response = keystone.Roles().post(request)305        self.assertStatusCode(response, 201)306        self.assertEqual(response['location'],307                         '/api/keystone/roles/role123')308        self.assertEqual(response.json, {"id": "role123", "name": "bob"})309        self.mock_role_create.assert_called_once_with(request, 'bob')310    @test.create_mocks({api.keystone: ['add_tenant_user_role']})311    def test_role_grant(self):312        self.mock_add_tenant_user_role.return_value = None313        request = self.mock_rest_request(body='''314            {"action": "grant", "data": {"user_id": "user123",315            "role_id": "role123", "project_id": "project123"}}316        ''')317        response = keystone.ProjectRole().put(request, "project1", "role2",318                                              "user3")319        self.assertStatusCode(response, 204)320        self.assertEqual(response.content, b'')321        self.mock_add_tenant_user_role.assert_called_once_with(322            request, 'project1', 'user3', 'role2')323    @test.create_mocks({api.keystone: ['role_delete']})324    def test_role_delete_many(self):325        self.mock_role_delete.return_value = None326        request = self.mock_rest_request(body='''327            ["id1", "id2", "id3"]328        ''')329        response = keystone.Roles().delete(request)330        self.assertStatusCode(response, 204)331        self.assertEqual(response.content, b'')332        self.mock_role_delete.assert_has_calls([333            mock.call(request, 'id1'),334            mock.call(request, 'id2'),335            mock.call(request, 'id3'),336        ])337    @test.create_mocks({api.keystone: ['role_delete']})338    def test_role_delete(self):339        self.mock_role_delete.return_value = None340        request = self.mock_rest_request()341        response = keystone.Role().delete(request, 'the_id')342        self.assertStatusCode(response, 204)343        self.assertEqual(response.content, b'')344        self.mock_role_delete.assert_called_once_with(request, 'the_id')345    @test.create_mocks({api.keystone: ['role_update']})346    def test_role_patch(self):347        self.mock_role_update.return_value = self.roles.first()348        request = self.mock_rest_request(body='{"name": "spam"}')349        response = keystone.Role().patch(request, 'the_id')350        self.assertStatusCode(response, 204)351        self.assertEqual(response.content, b'')352        self.mock_role_update.assert_called_once_with(request,353                                                      'the_id',354                                                      'spam')355    #356    # Domains357    #358    @test.create_mocks({api.keystone: ['get_default_domain']})359    def test_default_domain_get(self):360        request = self.mock_rest_request()361        domain = api.base.APIDictWrapper({'id': 'the_id', 'name': 'the_name'})362        self.mock_get_default_domain.return_value = domain363        response = keystone.DefaultDomain().get(request)364        self.assertStatusCode(response, 200)365        self.assertEqual(response.json, domain.to_dict())366        self.mock_get_default_domain.assert_called_once_with(request)367    @test.create_mocks({api.keystone: ['domain_get']})368    def test_domain_get(self):369        request = self.mock_rest_request()370        ret_val_domain = self.mock_domain_get.return_value371        ret_val_domain.to_dict.return_value = {'name': 'Ni!'}372        response = keystone.Domain().get(request, 'the_id')373        self.assertStatusCode(response, 200)374        self.assertEqual(response.json, {"name": "Ni!"})375        self.mock_domain_get.assert_called_once_with(request, 'the_id')376    @test.create_mocks({api.keystone: ['get_default_domain']})377    def test_domain_get_default(self):378        request = self.mock_rest_request()379        self.mock_get_default_domain.return_value.to_dict.return_value = {380            'name': 'Ni!'381        }382        response = keystone.Domain().get(request, 'default')383        self.assertStatusCode(response, 200)384        self.assertEqual(response.json, {"name": "Ni!"})385        self.mock_get_default_domain.assert_called_once_with(request)386    @test.create_mocks({api.keystone: ['domain_list']})387    def test_domain_get_list(self):388        request = self.mock_rest_request()389        self.mock_domain_list.return_value = [390            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),391            mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})392        ]393        response = keystone.Domains().get(request)394        self.assertStatusCode(response, 200)395        self.assertEqual(response.json,396                         {"items": [{"name": "Ni!"}, {"name": "Ptang!"}]})397        self.mock_domain_list.assert_called_once_with(request)398    def test_domain_create_full(self):399        self._test_domain_create(400            '{"name": "bob", '401            '"description": "sekrit", "enabled": false}',402            {403                'description': 'sekrit',404                'enabled': False405            }406        )407    def test_domain_create_partial(self):408        self._test_domain_create(409            '{"name": "bob"}',410            {411                'description': None,412                'enabled': True413            }414        )415    @test.create_mocks({api.keystone: ['domain_create']})416    def _test_domain_create(self, supplied_body, expected_call):417        request = self.mock_rest_request(body=supplied_body)418        ret_val_domain = self.mock_domain_create.return_value419        ret_val_domain.id = 'domain123'420        ret_val_domain.to_dict.return_value = {421            'id': 'domain123', 'name': 'bob'422        }423        response = keystone.Domains().post(request)424        self.assertStatusCode(response, 201)425        self.assertEqual(response['location'],426                         '/api/keystone/domains/domain123')427        self.assertEqual(response.json, {"id": "domain123", "name": "bob"})428        self.mock_domain_create.assert_called_once_with(request, 'bob',429                                                        **expected_call)430    @test.create_mocks({api.keystone: ['domain_delete']})431    def test_domain_delete_many(self):432        self.mock_domain_delete.return_value = None433        request = self.mock_rest_request(body='''434            ["id1", "id2", "id3"]435        ''')436        response = keystone.Domains().delete(request)437        self.assertStatusCode(response, 204)438        self.assertEqual(response.content, b'')439        self.mock_domain_delete.assert_has_calls([440            mock.call(request, 'id1'),441            mock.call(request, 'id2'),442            mock.call(request, 'id3'),443        ])444    @test.create_mocks({api.keystone: ['domain_delete']})445    def test_domain_delete(self):446        self.mock_domain_delete.return_value = None447        request = self.mock_rest_request()448        response = keystone.Domain().delete(request, 'the_id')449        self.assertStatusCode(response, 204)450        self.assertEqual(response.content, b'')451        self.mock_domain_delete.assert_called_once_with(request, 'the_id')452    @test.create_mocks({api.keystone: ['domain_update']})453    def test_domain_patch(self):454        self.mock_domain_update.return_value = self.domains.first()455        request = self.mock_rest_request(body='{"name": "spam"}')456        response = keystone.Domain().patch(request, 'the_id')457        self.assertStatusCode(response, 204)458        self.assertEqual(response.content, b'')459        self.mock_domain_update.assert_called_once_with(request,460                                                        'the_id',461                                                        name='spam',462                                                        description=None,463                                                        enabled=None)464    #465    # Projects466    #467    @test.create_mocks({api.keystone: ['tenant_get']})468    def test_project_get(self):469        request = self.mock_rest_request()470        ret_val_tenant = self.mock_tenant_get.return_value471        ret_val_tenant.to_dict.return_value = {'name': 'Ni!'}472        response = keystone.Project().get(request, 'the_id')473        self.assertStatusCode(response, 200)474        self.assertEqual(response.json, {"name": "Ni!"})475        self.mock_tenant_get.assert_called_once_with(476            request, 'the_id', admin=False)477    def test_project_get_list(self):478        self._test_project_get_list(479            {},480            {481                'paginate': False,482                'marker': None,483                'domain': None,484                'user': None,485                'admin': True,486                'filters': None487            }488        )489    def test_project_get_list_with_params_true(self):490        self._test_project_get_list(491            {492                'paginate': 'true',493                'admin': 'true'494            },495            {496                'paginate': True,497                'marker': None,498                'domain': None,499                'user': None,500                'admin': True,501                'filters': None502            }503        )504    def test_project_get_list_with_params_false(self):505        self._test_project_get_list(506            {507                'paginate': 'false',508                'admin': 'false'509            },510            {511                'paginate': False,512                'marker': None,513                'domain': None,514                'user': None,515                'admin': False,516                'filters': None517            }518        )519    @test.create_mocks({api.keystone: ['tenant_list']})520    def _test_project_get_list(self, params, expected_call):521        request = self.mock_rest_request(**{'GET': dict(**params)})522        self.mock_tenant_list.return_value = ([523            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),524            mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})525        ], False)526        with mock.patch.object(settings, 'DEBUG', True):527            response = keystone.Projects().get(request)528        self.assertStatusCode(response, 200)529        self.assertEqual(response.json,530                         {"has_more": False,531                          "items": [{"name": "Ni!"}, {"name": "Ptang!"}]})532        self.mock_tenant_list.assert_called_once_with(request, **expected_call)533    @test.create_mocks({api.keystone: ['tenant_list']})534    def test_project_get_list_with_filters(self):535        filters = {'name': 'Ni!'}536        request = self.mock_rest_request(**{'GET': dict(**filters)})537        self.mock_tenant_list.return_value = ([538            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),539            mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}})540        ], False)541        with mock.patch.object(settings, 'DEBUG', True):542            response = keystone.Projects().get(request)543        self.assertStatusCode(response, 200)544        self.assertEqual(response.json,545                         {"has_more": False,546                          "items": [{"name": "Ni!"}, {"name": "Ni!"}]})547        self.mock_tenant_list.assert_called_once_with(request, paginate=False,548                                                      marker=None, domain=None,549                                                      user=None, admin=True,550                                                      filters=filters)551    def test_project_create_full(self):552        self._test_project_create(553            '{"name": "bob", '554            '"domain_id": "domain123", "description": "sekrit", '555            '"enabled": false}',556            {557                'name': 'bob',558                'description': 'sekrit',559                'domain': 'domain123',560                'enabled': False561            }562        )563    def test_project_create_partial(self):564        self._test_project_create(565            '{"name": "bob"}',566            {567                'name': 'bob',568                'description': None,569                'domain': None,570                'enabled': True571            }572        )573    @test.create_mocks({api.keystone: ['tenant_create']})574    def _test_project_create(self, supplied_body, expected_args):575        request = self.mock_rest_request(body=supplied_body)576        self.mock_tenant_create.return_value.id = 'project123'577        self.mock_tenant_create.return_value.to_dict.return_value = {578            'id': 'project123', 'name': 'bob'579        }580        response = keystone.Projects().post(request)581        self.assertStatusCode(response, 201)582        self.assertEqual(response['location'],583                         '/api/keystone/projects/project123')584        self.assertEqual(response.json,585                         {"id": "project123", "name": "bob"})586        self.mock_tenant_create.assert_called_once_with(request,587                                                        **expected_args)588    @test.create_mocks({api.keystone: ['tenant_delete']})589    def test_project_delete_many(self):590        self.mock_tenant_delete.return_value = None591        request = self.mock_rest_request(body='''592            ["id1", "id2", "id3"]593        ''')594        response = keystone.Projects().delete(request)595        self.assertStatusCode(response, 204)596        self.assertEqual(response.content, b'')597        self.mock_tenant_delete.assert_has_calls([598            mock.call(request, 'id1'),599            mock.call(request, 'id2'),600            mock.call(request, 'id3'),601        ])602    @test.create_mocks({api.keystone: ['tenant_delete']})603    def test_project_delete(self):604        self.mock_tenant_delete.return_value = None605        request = self.mock_rest_request()606        response = keystone.Project().delete(request, 'the_id')607        self.assertStatusCode(response, 204)608        self.assertEqual(response.content, b'')609        self.mock_tenant_delete.assert_called_once_with(request, 'the_id')610    @test.create_mocks({api.keystone: ['tenant_update']})611    def test_project_patch(self):612        # nothing in the Horizon code documents what additional parameters are613        # allowed, so we'll just assume GIGO614        self.mock_tenant_update.return_value = self.tenants.first()615        request = self.mock_rest_request(body='''616            {"name": "spam", "domain_id": "domain123", "foo": "bar"}617        ''')618        response = keystone.Project().patch(request, 'spam123')619        self.assertStatusCode(response, 204)620        self.assertEqual(response.content, b'')621        self.mock_tenant_update.assert_called_once_with(request,622                                                        'spam123',623                                                        name='spam', foo='bar',624                                                        description=None,625                                                        domain='domain123',626                                                        enabled=None)627    #628    # Service Catalog629    #630    def test_service_catalog_get(self):631        request = self.mock_rest_request()632        request.user = mock.MagicMock(**{'service_catalog': [633            {'endpoints': [634                {'url': 'http://cool_url/image',635                 'interface': 'admin',636                 'region': 'RegionOne',637                 'region_id': 'RegionOne',638                 'id': 'test'},639                {'url': 'http://cool_url/image',640                 'interface': 'public',641                 'region': 'RegionOne',642                 'region_id': 'RegionOne',643                 'id': 'test'},644                {'url': 'http://cool_url/image',645                 'interface': 'internal',646                 'region': 'RegionOne',647                 'region_id': 'RegionOne',648                 'id': 'test'}],649                'type': 'image',650                'id': '2b5bc2e59b094f898a43f5e8ce446240',651                'name': 'glance'},652            {'endpoints': [653                {'url': 'http://cool_url/volume/v2/test',654                 'interface': 'public',655                 'region': 'RegionOne',656                 'region_id': 'RegionOne',657                 'id': '29a629afb80547ea9baa4266e97b4cb5'},658                {'url': 'http://cool_url/volume/v2/test',659                 'interface': 'admin',660                 'region': 'RegionOne',661                 'region_id': 'RegionOne',662                 'id': '29a629afb80547ea9baa4266e97b4cb5'}],663                'type': 'volumev2',664                'id': '55ef272cfa714e54b8f2046c157b027d',665                'name': 'cinderv2'},666            {'endpoints': [667                {'url': 'http://cool_url/compute/v2/check',668                 'interface': 'internal',669                 'region': 'RegionOne',670                 'region_id': 'RegionOne',671                 'id': 'e8c440e025d94355ab82c78cc2062129'}],672                'type': 'compute_legacy',673                'id': 'b7f1d3f4119643508d5ca2325eb8af87',674                'name': 'nova_legacy'}]})675        response = keystone.ServiceCatalog().get(request)676        self.assertStatusCode(response, 200)677        content = [{'endpoints': [678                    {'url': 'http://cool_url/image',679                     'interface': 'public',680                     'region': 'RegionOne',681                     'region_id': 'RegionOne',682                     'id': 'test'}],683                    'type': 'image',684                    'id': '2b5bc2e59b094f898a43f5e8ce446240',685                    'name': 'glance'},686                   {'endpoints': [687                    {'url': 'http://cool_url/volume/v2/test',688                     'interface': 'public',689                     'region': 'RegionOne',690                     'region_id': 'RegionOne',691                     'id': '29a629afb80547ea9baa4266e97b4cb5'}],692                    'type': 'volumev2',693                    'id': '55ef272cfa714e54b8f2046c157b027d',694                    'name': 'cinderv2'}]695        self.assertEqual(content, jsonutils.loads(response.content))696    #697    # User Session698    #699    def test_user_session_get(self):700        request = self.mock_rest_request()701        request.user = mock.Mock(702            services_region='some region',703            super_secret_thing='not here',704            token=type('', (object,), {'id': 'token here'}),705            is_authenticated=lambda: True,706            spec=['services_region', 'super_secret_thing']707        )708        response = keystone.UserSession().get(request)709        self.assertStatusCode(response, 200)710        content = jsonutils.loads(response.content)711        self.assertEqual(content['services_region'], 'some region')712        self.assertEqual(content['token'], 'token here')713        self.assertNotIn('super_secret_thing', content)714    #715    # Groups716    #717    @test.create_mocks({api.keystone: ['group_list']})718    def test_group_get_list(self):719        request = self.mock_rest_request(**{720            'session.get': mock.Mock(return_value='the_domain'),721            'GET': {},722        })723        self.mock_group_list.return_value = [724            mock.Mock(**{'to_dict.return_value': {'name': 'uno!'}}),725            mock.Mock(**{'to_dict.return_value': {'name': 'dos!'}})726        ]727        response = keystone.Groups().get(request)728        self.assertStatusCode(response, 200)729        self.assertEqual(response.json,730                         {"items": [{"name": "uno!"}, {"name": "dos!"}]})731        self.mock_group_list.assert_called_once_with(request,732                                                     domain='the_domain')733    @test.create_mocks({api.keystone: ['group_create']})734    def test_group_create(self):735        request = self.mock_rest_request(**{736            'session.get': mock.Mock(return_value='the_domain'),737            'GET': {},738            'body': '{"name": "bug!", "description": "bugaboo!!"}',739        })740        self.mock_group_create.return_value.id = 'group789'741        self.mock_group_create.return_value.to_dict.return_value = {742            'id': 'group789', 'name': 'bug!', 'description': 'bugaboo!!'743        }744        response = keystone.Groups().post(request)745        self.assertStatusCode(response, 201)746        self.assertEqual(response['location'],747                         '/api/keystone/groups/group789')748        self.assertEqual(response.json,749                         {"id": "group789",750                          "name": "bug!",751                          "description": "bugaboo!!"})752        self.mock_group_create.assert_called_once_with(request, 'the_domain',753                                                       'bug!', 'bugaboo!!')754    @test.create_mocks({api.keystone: ['group_create']})755    def test_group_create_without_description(self):756        request = self.mock_rest_request(**{757            'session.get': mock.Mock(return_value='the_domain'),758            'GET': {},759            'body': '{"name": "bug!"}',760        })761        self.mock_group_create.return_value.id = 'group789'762        self.mock_group_create.return_value.to_dict.return_value = {763            'id': 'group789', 'name': 'bug!'764        }765        response = keystone.Groups().post(request)766        self.assertStatusCode(response, 201)767        self.assertEqual(response['location'],768                         '/api/keystone/groups/group789')769        self.assertEqual(response.json,770                         {"id": "group789",771                          "name": "bug!"})772        self.mock_group_create.assert_called_once_with(request, 'the_domain',773                                                       'bug!', None)774    @test.create_mocks({api.keystone: ['group_get']})775    def test_group_get(self):776        request = self.mock_rest_request()777        self.mock_group_get.return_value.to_dict.return_value = {778            'name': 'bug!', 'description': 'bugaboo!!'}779        response = keystone.Group().get(request, 'the_id')780        self.assertStatusCode(response, 200)781        self.assertEqual(response.json, {"name": "bug!",782                         "description": "bugaboo!!"})783        self.mock_group_get.assert_called_once_with(request, 'the_id')784    @test.create_mocks({api.keystone: ['group_delete']})785    def test_group_delete(self):786        self.mock_group_delete.return_value = None787        request = self.mock_rest_request()788        response = keystone.Group().delete(request, 'the_id')789        self.assertStatusCode(response, 204)790        self.assertEqual(response.content, b'')791        self.mock_group_delete.assert_called_once_with(request, 'the_id')792    @test.create_mocks({api.keystone: ['group_update']})793    def test_group_patch(self):794        self.mock_group_update.return_value = self.groups.first()795        request = self.mock_rest_request(796            body='{"name": "spam_i_am", "description": "Sir Spam"}')797        response = keystone.Group().patch(request, 'the_id')798        self.assertStatusCode(response, 204)799        self.assertEqual(response.content, b'')800        self.mock_group_update.assert_called_once_with(request,801                                                       'the_id',802                                                       'spam_i_am',803                                                       'Sir Spam')804    @test.create_mocks({api.keystone: ['group_delete']})805    def test_group_delete_many(self):806        self.mock_group_delete.return_value = None807        request = self.mock_rest_request(body='''808            ["id1", "id2", "id3"]809        ''')810        response = keystone.Groups().delete(request)811        self.assertStatusCode(response, 204)812        self.assertEqual(response.content, b'')813        self.mock_group_delete.assert_has_calls([814            mock.call(request, 'id1'),815            mock.call(request, 'id2'),816            mock.call(request, 'id3'),817        ])818    #819    # Services820    #821    @test.create_mocks({api.keystone: ['Service']})822    def test_services_get(self):823        request = self.mock_rest_request()824        mock_service = {825            "name": "srv_name",826            "type": "srv_type",827            "host": "srv_host"828        }829        request.user = mock.Mock(830            service_catalog=[mock_service],831            services_region='some region'832        )833        self.mock_Service.return_value.to_dict.return_value = mock_service834        response = keystone.Services().get(request)835        self.assertStatusCode(response, 200)...

test_sense_pi.py

Source:test_sense_pi.py

1from django.test import SimpleTestCase2from unittest.mock import patch, MagicMock3from testfixtures import TempDirectory4import os5import sys6sys.modules[7    "sense_hat"8] = MagicMock()  # mock these modules so that they don't have to be installed9sys.modules["sense_emu"] = MagicMock()10from hardware.SensorPi.sense_pi import SensePi  # noqa : E40211from hardware.Utils.logger import Logger  # noqa : E40212@patch("hardware.SensorPi.sense_pi.SenseHat")13class SensePiTests(SimpleTestCase):14    def setUp(self):15        self.temp_dir = TempDirectory()16    def tearDown(self):17        self.temp_dir.cleanup()18    def test_init_no_logs_no_ids(self, mock_sense):19        mock_show_message = MagicMock()20        mock_clear = MagicMock()21        mock_sense.return_value.show_message = mock_show_message22        mock_sense.return_value.clear = mock_clear23        expected_sensor_ids = {24            "temperature": 1,25            "pressure": 2,26            "humidity": 3,27            "acceleration": 4,28            "orientation": 5,29        }30        expected_color = (87, 46, 140)31        with patch.dict(32            os.environ,33            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},34        ):35            sense = SensePi()36            mock_show_message.assert_called_with(37                "MERCURY", text_colour=expected_color, scroll_speed=0.0438            )39            mock_clear.assert_called()40            self.assertTrue(sense.logging is not None)41            self.assertTrue(sense.logging.name == "SENSE_HAT_LOG_FILE")42            self.assertIsInstance(sense.logging, Logger)43            self.assertDictEqual(sense.sensor_ids, expected_sensor_ids)44    def test_init_logs_ids(self, mock_sense):45        mock_show_message = MagicMock()46        mock_clear = MagicMock()47        mock_sense.return_value.show_message = mock_show_message48        mock_sense.return_value.clear = mock_clear49        expected_sensor_ids = {50            "temperature": 5,51            "pressure": 4,52            "humidity": 3,53            "acceleration": 2,54            "orientation": 1,55        }56        expected_color = (87, 46, 140)57        with patch.dict(58            os.environ, {"SENSE_LOG": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path}59        ):60            sense = SensePi(log_file_name="SENSE_LOG", sensor_ids=expected_sensor_ids)61            mock_show_message.assert_called_with(62                "MERCURY", text_colour=expected_color, scroll_speed=0.0463            )64            mock_clear.assert_called()65            self.assertTrue(sense.logging is not None)66            self.assertTrue(sense.logging.name == "SENSE_LOG")67            self.assertIsInstance(sense.logging, Logger)68            self.assertDictEqual(sense.sensor_ids, expected_sensor_ids)69    def test_init_no_logs_ids(self, mock_sense):70        mock_show_message = MagicMock()71        mock_clear = MagicMock()72        mock_sense.return_value.show_message = mock_show_message73        mock_sense.return_value.clear = mock_clear74        expected_sensor_ids = {75            "temperature": 5,76            "pressure": 4,77            "humidity": 3,78            "acceleration": 2,79            "orientation": 1,80        }81        expected_color = (87, 46, 140)82        with patch.dict(83            os.environ,84            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},85        ):86            sense = SensePi(sensor_ids=expected_sensor_ids)87            mock_show_message.assert_called_with(88                "MERCURY", text_colour=expected_color, scroll_speed=0.0489            )90            mock_clear.assert_called()91            self.assertTrue(sense.logging is not None)92            self.assertTrue(sense.logging.name == "SENSE_HAT_LOG_FILE")93            self.assertIsInstance(sense.logging, Logger)94            self.assertDictEqual(sense.sensor_ids, expected_sensor_ids)95    def test_init_logs_no_ids(self, mock_sense):96        mock_show_message = MagicMock()97        mock_clear = MagicMock()98        mock_sense.return_value.show_message = mock_show_message99        mock_sense.return_value.clear = mock_clear100        expected_sensor_ids = {101            "temperature": 1,102            "pressure": 2,103            "humidity": 3,104            "acceleration": 4,105            "orientation": 5,106        }107        expected_color = (87, 46, 140)108        with patch.dict(109            os.environ, {"SENSE_LOG": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path}110        ):111            sense = SensePi(log_file_name="SENSE_LOG")112            mock_show_message.assert_called_with(113                "MERCURY", text_colour=expected_color, scroll_speed=0.04114            )115            mock_clear.assert_called()116            self.assertTrue(sense.logging is not None)117            self.assertTrue(sense.logging.name == "SENSE_LOG")118            self.assertIsInstance(sense.logging, Logger)119            self.assertDictEqual(sense.sensor_ids, expected_sensor_ids)120    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")121    def test_factory_temp(self, mock_date, mock_sense):122        mock_show_message = MagicMock()123        mock_clear = MagicMock()124        mock_sense.return_value.show_message = mock_show_message125        mock_sense.return_value.clear = mock_clear126        mock_sense.return_value.get_temperature.return_value = "100"127        mock_sense.return_value.get_pressure.return_value = "50"128        mock_sense.return_value.get_humidity.return_value = "20"129        mock_sense.return_value.get_accelerometer_raw.return_value = "20"130        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)131        date_str = "example_date"132        mock_date.return_value = date_str133        with patch.dict(134            os.environ,135            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},136        ):137            sense = SensePi()138            data = sense.factory("TEMPERATURE")139            expected_data = {140                "sensor_id": 1,141                "values": {"temperature": "100"},142                "date": date_str,143            }144            self.assertDictEqual(expected_data, data)145    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")146    def test_factory_pressure(self, mock_date, mock_sense):147        mock_show_message = MagicMock()148        mock_clear = MagicMock()149        mock_sense.return_value.show_message = mock_show_message150        mock_sense.return_value.clear = mock_clear151        mock_sense.return_value.get_temperature.return_value = "100"152        mock_sense.return_value.get_pressure.return_value = "50"153        mock_sense.return_value.get_humidity.return_value = "20"154        mock_sense.return_value.get_accelerometer_raw.return_value = "20"155        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)156        date_str = "example_date"157        mock_date.return_value = date_str158        with patch.dict(159            os.environ,160            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},161        ):162            sense = SensePi()163            data = sense.factory("PRESSURE")164            expected_data = {165                "sensor_id": 2,166                "values": {"pressure": "50"},167                "date": date_str,168            }169            self.assertDictEqual(expected_data, data)170    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")171    def test_factory_humidity(self, mock_date, mock_sense):172        mock_show_message = MagicMock()173        mock_clear = MagicMock()174        mock_sense.return_value.show_message = mock_show_message175        mock_sense.return_value.clear = mock_clear176        mock_sense.return_value.get_temperature.return_value = "100"177        mock_sense.return_value.get_pressure.return_value = "50"178        mock_sense.return_value.get_humidity.return_value = "20"179        mock_sense.return_value.get_accelerometer_raw.return_value = "20"180        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)181        date_str = "example_date"182        mock_date.return_value = date_str183        with patch.dict(184            os.environ,185            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},186        ):187            sense = SensePi()188            data = sense.factory("HUMIDITY")189            expected_data = {190                "sensor_id": 3,191                "values": {"humidity": "20"},192                "date": date_str,193            }194            self.assertDictEqual(expected_data, data)195    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")196    def test_factory_accel(self, mock_date, mock_sense):197        mock_show_message = MagicMock()198        mock_clear = MagicMock()199        mock_sense.return_value.show_message = mock_show_message200        mock_sense.return_value.clear = mock_clear201        mock_sense.return_value.get_temperature.return_value = "100"202        mock_sense.return_value.get_pressure.return_value = "50"203        mock_sense.return_value.get_humidity.return_value = "20"204        mock_sense.return_value.get_accelerometer_raw.return_value = "20"205        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)206        date_str = "example_date"207        mock_date.return_value = date_str208        with patch.dict(209            os.environ,210            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},211        ):212            sense = SensePi()213            data = sense.factory("ACCELERATION")214            expected_data = {215                "sensor_id": 4,216                "values": {"acceleration": "20"},217                "date": date_str,218            }219            self.assertDictEqual(expected_data, data)220    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")221    def test_factory_orientation(self, mock_date, mock_sense):222        mock_show_message = MagicMock()223        mock_clear = MagicMock()224        mock_sense.return_value.show_message = mock_show_message225        mock_sense.return_value.clear = mock_clear226        mock_sense.return_value.get_temperature.return_value = "100"227        mock_sense.return_value.get_pressure.return_value = "50"228        mock_sense.return_value.get_humidity.return_value = "20"229        mock_sense.return_value.get_accelerometer_raw.return_value = "20"230        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)231        date_str = "example_date"232        mock_date.return_value = date_str233        with patch.dict(234            os.environ,235            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},236        ):237            sense = SensePi()238            data = sense.factory("ORIENTATION")239            expected_data = {240                "sensor_id": 5,241                "values": {"orientation": (1, 1, 1)},242                "date": date_str,243            }244            self.assertDictEqual(expected_data, data)245    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")246    def test_factory_invalid_key(self, mock_date, mock_sense):247        mock_show_message = MagicMock()248        mock_clear = MagicMock()249        mock_sense.return_value.show_message = mock_show_message250        mock_sense.return_value.clear = mock_clear251        date_str = "example_date"252        mock_date.return_value = date_str253        with patch.dict(254            os.environ,255            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},256        ):257            sense = SensePi()258            data = sense.factory("SOME_KEY")259            expected_data = {}260            self.assertDictEqual(expected_data, data)261    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")262    def test_factory_all(self, mock_date, mock_sense):263        mock_show_message = MagicMock()264        mock_clear = MagicMock()265        mock_sense.return_value.show_message = mock_show_message266        mock_sense.return_value.clear = mock_clear267        mock_sense.return_value.get_temperature.return_value = "100"268        mock_sense.return_value.get_pressure.return_value = "50"269        mock_sense.return_value.get_humidity.return_value = "20"270        mock_sense.return_value.get_accelerometer_raw.return_value = "20"271        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)272        date_str = "example_date"273        mock_date.return_value = date_str274        with patch.dict(275            os.environ,276            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},277        ):278            sense = SensePi()279            data = sense.factory("ALL")280            expected_data = {281                "values": {282                    "temperature": "100",283                    "pressure": "50",284                    "humidity": "20",285                    "acceleration": "20",286                    "orientation": (1, 1, 1),287                },288                "date": date_str,289            }290            self.assertDictEqual(expected_data, data)291    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")292    def test_sense_get_all(self, mock_date, mock_sense):293        mock_show_message = MagicMock()294        mock_clear = MagicMock()295        mock_sense.return_value.show_message = mock_show_message296        mock_sense.return_value.clear = mock_clear297        mock_sense.return_value.get_temperature.return_value = "100"298        mock_sense.return_value.get_pressure.return_value = "50"299        mock_sense.return_value.get_humidity.return_value = "20"300        mock_sense.return_value.get_accelerometer_raw.return_value = "20"301        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)302        date_str = "example_date"303        mock_date.return_value = date_str304        with patch.dict(305            os.environ,306            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},307        ):308            sense = SensePi()309            data = sense.get_all()310            expected_data = {311                "values": {312                    "temperature": "100",313                    "pressure": "50",314                    "humidity": "20",315                    "acceleration": "20",316                    "orientation": (1, 1, 1),317                },318                "date": date_str,319            }320            self.assertDictEqual(expected_data, data)321    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")322    def test_sense_get_temp(self, mock_date, mock_sense):323        mock_show_message = MagicMock()324        mock_clear = MagicMock()325        mock_sense.return_value.show_message = mock_show_message326        mock_sense.return_value.clear = mock_clear327        mock_sense.return_value.get_temperature.return_value = "100"328        mock_sense.return_value.get_pressure.return_value = "50"329        mock_sense.return_value.get_humidity.return_value = "20"330        mock_sense.return_value.get_accelerometer_raw.return_value = "20"331        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)332        date_str = "example_date"333        mock_date.return_value = date_str334        with patch.dict(335            os.environ,336            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},337        ):338            sense = SensePi()339            data = sense.get_temperature()340            expected_data = {341                "sensor_id": 1,342                "values": {"temperature": "100"},343                "date": date_str,344            }345            self.assertDictEqual(expected_data, data)346    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")347    def test_sense_get_pressure(self, mock_date, mock_sense):348        mock_show_message = MagicMock()349        mock_clear = MagicMock()350        mock_sense.return_value.show_message = mock_show_message351        mock_sense.return_value.clear = mock_clear352        mock_sense.return_value.get_temperature.return_value = "100"353        mock_sense.return_value.get_pressure.return_value = "50"354        mock_sense.return_value.get_humidity.return_value = "20"355        mock_sense.return_value.get_accelerometer_raw.return_value = "20"356        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)357        date_str = "example_date"358        mock_date.return_value = date_str359        with patch.dict(360            os.environ,361            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},362        ):363            sense = SensePi()364            data = sense.get_pressure()365            expected_data = {366                "sensor_id": 2,367                "values": {"pressure": "50"},368                "date": date_str,369            }370            self.assertDictEqual(expected_data, data)371    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")372    def test_sense_get_humidity(self, mock_date, mock_sense):373        mock_show_message = MagicMock()374        mock_clear = MagicMock()375        mock_sense.return_value.show_message = mock_show_message376        mock_sense.return_value.clear = mock_clear377        mock_sense.return_value.get_temperature.return_value = "100"378        mock_sense.return_value.get_pressure.return_value = "50"379        mock_sense.return_value.get_humidity.return_value = "20"380        mock_sense.return_value.get_accelerometer_raw.return_value = "20"381        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)382        date_str = "example_date"383        mock_date.return_value = date_str384        with patch.dict(385            os.environ,386            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},387        ):388            sense = SensePi()389            data = sense.get_humidity()390            expected_data = {391                "sensor_id": 3,392                "values": {"humidity": "20"},393                "date": date_str,394            }395            self.assertDictEqual(expected_data, data)396    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")397    def test_sense_get_accel(self, mock_date, mock_sense):398        mock_show_message = MagicMock()399        mock_clear = MagicMock()400        mock_sense.return_value.show_message = mock_show_message401        mock_sense.return_value.clear = mock_clear402        mock_sense.return_value.get_temperature.return_value = "100"403        mock_sense.return_value.get_pressure.return_value = "50"404        mock_sense.return_value.get_humidity.return_value = "20"405        mock_sense.return_value.get_accelerometer_raw.return_value = "20"406        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)407        date_str = "example_date"408        mock_date.return_value = date_str409        with patch.dict(410            os.environ,411            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},412        ):413            sense = SensePi()414            data = sense.get_acceleration()415            expected_data = {416                "sensor_id": 4,417                "values": {"acceleration": "20"},418                "date": date_str,419            }420            self.assertDictEqual(expected_data, data)421    @patch("hardware.SensorPi.sense_pi.date_str_with_current_timezone")422    def test_sense_get_orientation(self, mock_date, mock_sense):423        mock_show_message = MagicMock()424        mock_clear = MagicMock()425        mock_sense.return_value.show_message = mock_show_message426        mock_sense.return_value.clear = mock_clear427        mock_sense.return_value.get_temperature.return_value = "100"428        mock_sense.return_value.get_pressure.return_value = "50"429        mock_sense.return_value.get_humidity.return_value = "20"430        mock_sense.return_value.get_accelerometer_raw.return_value = "20"431        mock_sense.return_value.get_orientation.return_value = (1, 1, 1)432        date_str = "example_date"433        mock_date.return_value = date_str434        with patch.dict(435            os.environ,436            {"SENSE_HAT_LOG_FILE": "logger.txt", "LOG_DIRECTORY": self.temp_dir.path},437        ):438            sense = SensePi()439            data = sense.get_orientation()440            expected_data = {441                "sensor_id": 5,442                "values": {"orientation": (1, 1, 1)},443                "date": date_str,444            }...

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.