KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use once method in chromy

Best JavaScript code snippet using chromy

test_servers.py

Source:test_servers.py

1# Copyright 2013: Mirantis Inc.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15from unittest import mock16import ddt17from rally import exceptions as rally_exceptions18from rally_openstack.task.scenarios.nova import servers19from tests.unit import fakes20from tests.unit import test21NOVA_SERVERS_MODULE = "rally_openstack.task.scenarios.nova.servers"22NOVA_SERVERS = NOVA_SERVERS_MODULE + ".NovaServers"23@ddt.ddt24class NovaServersTestCase(test.ScenarioTestCase):25    @ddt.data(("rescue_unrescue", ["_rescue_server", "_unrescue_server"], 1),26              ("stop_start", ["_stop_server", "_start_server"], 2),27              ("pause_unpause", ["_pause_server", "_unpause_server"], 3),28              ("suspend_resume", ["_suspend_server", "_resume_server"], 4),29              ("lock_unlock", ["_lock_server", "_unlock_server"], 5),30              ("shelve_unshelve", ["_shelve_server", "_unshelve_server"], 6))31    @ddt.unpack32    def test_action_pair(self, action_pair, methods, nof_calls):33        actions = [{action_pair: nof_calls}]34        fake_server = mock.MagicMock()35        scenario = servers.BootAndBounceServer(self.context)36        scenario._boot_server = mock.MagicMock(return_value=fake_server)37        scenario._delete_server = mock.MagicMock()38        scenario.generate_random_name = mock.MagicMock(return_value="name")39        for method in methods:40            setattr(scenario, method, mock.MagicMock())41        scenario.run("img", 1, actions=actions)42        scenario._boot_server.assert_called_once_with("img", 1)43        server_calls = []44        for i in range(nof_calls):45            server_calls.append(mock.call(fake_server))46        for method in methods:47            mocked_method = getattr(scenario, method)48            self.assertEqual(nof_calls, mocked_method.call_count,49                             "%s not called %d times" % (method, nof_calls))50            mocked_method.assert_has_calls(server_calls)51        scenario._delete_server.assert_called_once_with(fake_server,52                                                        force=False)53    def test_multiple_bounce_actions(self):54        actions = [{"hard_reboot": 5}, {"stop_start": 8},55                   {"rescue_unrescue": 3}, {"pause_unpause": 2},56                   {"suspend_resume": 4}, {"lock_unlock": 6},57                   {"shelve_unshelve": 7}]58        fake_server = mock.MagicMock()59        scenario = servers.BootAndBounceServer(self.context)60        scenario._boot_server = mock.MagicMock(return_value=fake_server)61        scenario._delete_server = mock.MagicMock()62        scenario._reboot_server = mock.MagicMock()63        scenario._stop_and_start_server = mock.MagicMock()64        scenario._rescue_and_unrescue_server = mock.MagicMock()65        scenario._pause_and_unpause_server = mock.MagicMock()66        scenario._suspend_and_resume_server = mock.MagicMock()67        scenario._lock_and_unlock_server = mock.MagicMock()68        scenario._shelve_and_unshelve_server = mock.MagicMock()69        scenario.generate_random_name = mock.MagicMock(return_value="name")70        scenario.run("img", 1, actions=actions)71        scenario._boot_server.assert_called_once_with("img", 1)72        server_calls = []73        for i in range(5):74            server_calls.append(mock.call(fake_server))75        self.assertEqual(5, scenario._reboot_server.call_count,76                         "Reboot not called 5 times")77        scenario._reboot_server.assert_has_calls(server_calls)78        server_calls = []79        for i in range(8):80            server_calls.append(mock.call(fake_server))81        self.assertEqual(8, scenario._stop_and_start_server.call_count,82                         "Stop/Start not called 8 times")83        scenario._stop_and_start_server.assert_has_calls(server_calls)84        server_calls = []85        for i in range(3):86            server_calls.append(mock.call(fake_server))87        self.assertEqual(3, scenario._rescue_and_unrescue_server.call_count,88                         "Rescue/Unrescue not called 3 times")89        scenario._rescue_and_unrescue_server.assert_has_calls(server_calls)90        server_calls = []91        for i in range(2):92            server_calls.append(mock.call(fake_server))93        self.assertEqual(2, scenario._pause_and_unpause_server.call_count,94                         "Pause/Unpause not called 2 times")95        scenario._pause_and_unpause_server.assert_has_calls(server_calls)96        server_calls = []97        for i in range(4):98            server_calls.append(mock.call(fake_server))99        self.assertEqual(4, scenario._suspend_and_resume_server.call_count,100                         "Suspend/Resume not called 4 times")101        scenario._suspend_and_resume_server.assert_has_calls(server_calls)102        server_calls = []103        for i in range(6):104            server_calls.append(mock.call(fake_server))105        self.assertEqual(6, scenario._lock_and_unlock_server.call_count,106                         "Lock/Unlock not called 6 times")107        scenario._lock_and_unlock_server.assert_has_calls(server_calls)108        server_calls = []109        for i in range(7):110            server_calls.append(mock.call(fake_server))111        self.assertEqual(7, scenario._shelve_and_unshelve_server.call_count,112                         "Shelve/Unshelve not called 7 times")113        scenario._shelve_and_unshelve_server.assert_has_calls(server_calls)114        scenario._delete_server.assert_called_once_with(fake_server,115                                                        force=False)116    def test_boot_lock_unlock_and_delete(self):117        server = fakes.FakeServer()118        image = fakes.FakeImage()119        flavor = fakes.FakeFlavor()120        scenario = servers.BootLockUnlockAndDelete(self.context)121        scenario._boot_server = mock.Mock(return_value=server)122        scenario._lock_server = mock.Mock(side_effect=lambda s: s.lock())123        scenario._unlock_server = mock.Mock(side_effect=lambda s: s.unlock())124        scenario._delete_server = mock.Mock(125            side_effect=lambda s, **kwargs:126                self.assertFalse(getattr(s, "OS-EXT-STS:locked", False)))127        scenario.run(image, flavor, fakearg="fakearg")128        scenario._boot_server.assert_called_once_with(image, flavor,129                                                      fakearg="fakearg")130        scenario._lock_server.assert_called_once_with(server)131        scenario._unlock_server.assert_called_once_with(server)132        scenario._delete_server.assert_called_once_with(server, force=False)133    @ddt.data("hard_reboot", "soft_reboot", "stop_start",134              "rescue_unrescue", "pause_unpause", "suspend_resume",135              "lock_unlock", "shelve_unshelve")136    def test_validate_actions(self, action):137        scenario = servers.BootAndBounceServer(self.context)138        self.assertRaises(rally_exceptions.InvalidConfigException,139                          scenario.run,140                          1, 1, actions=[{action: "no"}])141        self.assertRaises(rally_exceptions.InvalidConfigException,142                          scenario.run,143                          1, 1, actions=[{action: -1}])144        self.assertRaises(rally_exceptions.InvalidConfigException,145                          scenario.run,146                          1, 1, actions=[{action: 0}])147    def test_validate_actions_additional(self):148        scenario = servers.BootAndBounceServer(self.context)149        self.assertRaises(rally_exceptions.InvalidConfigException,150                          scenario.run,151                          1, 1, actions=[{"not_existing_action": "no"}])152        # NOTE: next should fail because actions parameter is a just a153        # dictionary, not an array of dictionaries154        self.assertRaises(rally_exceptions.InvalidConfigException,155                          scenario.run,156                          1, 1, actions={"hard_reboot": 1})157    def _verify_reboot(self, soft=True):158        actions = [{"soft_reboot" if soft else "hard_reboot": 5}]159        fake_server = mock.MagicMock()160        scenario = servers.BootAndBounceServer(self.context)161        scenario._reboot_server = mock.MagicMock()162        scenario._soft_reboot_server = mock.MagicMock()163        scenario._boot_server = mock.MagicMock(return_value=fake_server)164        scenario._delete_server = mock.MagicMock()165        scenario.generate_random_name = mock.MagicMock(return_value="name")166        scenario.run("img", 1, actions=actions)167        scenario._boot_server.assert_called_once_with("img", 1)168        server_calls = []169        for i in range(5):170            server_calls.append(mock.call(fake_server))171        if soft:172            self.assertEqual(5, scenario._soft_reboot_server.call_count,173                             "Reboot not called 5 times")174            scenario._soft_reboot_server.assert_has_calls(server_calls)175        else:176            self.assertEqual(5, scenario._reboot_server.call_count,177                             "Reboot not called 5 times")178            scenario._reboot_server.assert_has_calls(server_calls)179        scenario._delete_server.assert_called_once_with(fake_server,180                                                        force=False)181    def test_boot_soft_reboot(self):182        self._verify_reboot(soft=True)183    def test_boot_hard_reboot(self):184        self._verify_reboot(soft=False)185    def test_boot_and_delete_server(self):186        fake_server = object()187        scenario = servers.BootAndDeleteServer(self.context)188        scenario.generate_random_name = mock.MagicMock(return_value="name")189        scenario._boot_server = mock.MagicMock(return_value=fake_server)190        scenario._delete_server = mock.MagicMock()191        scenario.sleep_between = mock.MagicMock()192        scenario.run("img", 0, 10, 20, fakearg="fakearg")193        scenario._boot_server.assert_called_once_with("img", 0,194                                                      fakearg="fakearg")195        scenario.sleep_between.assert_called_once_with(10, 20)196        scenario._delete_server.assert_called_once_with(fake_server,197                                                        force=False)198    def test_boot_and_delete_multiple_servers(self):199        scenario = servers.BootAndDeleteMultipleServers(self.context)200        scenario._boot_servers = mock.Mock()201        scenario._delete_servers = mock.Mock()202        scenario.sleep_between = mock.Mock()203        scenario.run("img", "flavor", count=15, min_sleep=10,204                     max_sleep=20, fakearg="fakearg")205        scenario._boot_servers.assert_called_once_with("img", "flavor", 1,206                                                       instances_amount=15,207                                                       fakearg="fakearg")208        scenario.sleep_between.assert_called_once_with(10, 20)209        scenario._delete_servers.assert_called_once_with(210            scenario._boot_servers.return_value, force=False)211    def test_boot_and_list_server(self):212        scenario = servers.BootAndListServer(self.context)213#        scenario.generate_random_name = mock.MagicMock(return_value="name")214        img_name = "img"215        flavor_uuid = 0216        details = True217        fake_server_name = mock.MagicMock()218        scenario._boot_server = mock.MagicMock()219        scenario._list_servers = mock.MagicMock()220        scenario._list_servers.return_value = [mock.MagicMock(),221                                               fake_server_name,222                                               mock.MagicMock()]223        # Positive case224        scenario._boot_server.return_value = fake_server_name225        scenario.run(img_name, flavor_uuid, fakearg="fakearg")226        scenario._boot_server.assert_called_once_with(img_name, flavor_uuid,227                                                      fakearg="fakearg")228        scenario._list_servers.assert_called_once_with(details)229        # Negative case1: server isn't created230        scenario._boot_server.return_value = None231        self.assertRaises(rally_exceptions.RallyAssertionError,232                          scenario.run,233                          img_name, flavor_uuid, fakearg="fakearg")234        scenario._boot_server.assert_called_with(img_name, flavor_uuid,235                                                 fakearg="fakearg")236        # Negative case2: server not in the list of available servers237        scenario._boot_server.return_value = mock.MagicMock()238        self.assertRaises(rally_exceptions.RallyAssertionError,239                          scenario.run,240                          img_name, flavor_uuid, fakearg="fakearg")241        scenario._boot_server.assert_called_with(img_name, flavor_uuid,242                                                 fakearg="fakearg")243        scenario._list_servers.assert_called_with(details)244    def test_suspend_and_resume_server(self):245        fake_server = object()246        scenario = servers.SuspendAndResumeServer(self.context)247        scenario.generate_random_name = mock.MagicMock(return_value="name")248        scenario._boot_server = mock.MagicMock(return_value=fake_server)249        scenario._suspend_server = mock.MagicMock()250        scenario._resume_server = mock.MagicMock()251        scenario._delete_server = mock.MagicMock()252        scenario.run("img", 0, fakearg="fakearg")253        scenario._boot_server.assert_called_once_with("img", 0,254                                                      fakearg="fakearg")255        scenario._suspend_server.assert_called_once_with(fake_server)256        scenario._resume_server.assert_called_once_with(fake_server)257        scenario._delete_server.assert_called_once_with(fake_server,258                                                        force=False)259    def test_pause_and_unpause_server(self):260        fake_server = object()261        scenario = servers.PauseAndUnpauseServer(self.context)262        scenario.generate_random_name = mock.MagicMock(return_value="name")263        scenario._boot_server = mock.MagicMock(return_value=fake_server)264        scenario._pause_server = mock.MagicMock()265        scenario._unpause_server = mock.MagicMock()266        scenario._delete_server = mock.MagicMock()267        scenario.run("img", 0, fakearg="fakearg")268        scenario._boot_server.assert_called_once_with("img", 0,269                                                      fakearg="fakearg")270        scenario._pause_server.assert_called_once_with(fake_server)271        scenario._unpause_server.assert_called_once_with(fake_server)272        scenario._delete_server.assert_called_once_with(fake_server,273                                                        force=False)274    def test_shelve_and_unshelve_server(self):275        fake_server = mock.MagicMock()276        scenario = servers.ShelveAndUnshelveServer(self.context)277        scenario._boot_server = mock.MagicMock(return_value=fake_server)278        scenario._shelve_server = mock.MagicMock()279        scenario._unshelve_server = mock.MagicMock()280        scenario._delete_server = mock.MagicMock()281        scenario.run("img", 0, fakearg="fakearg")282        scenario._boot_server.assert_called_once_with("img", 0,283                                                      fakearg="fakearg")284        scenario._shelve_server.assert_called_once_with(fake_server)285        scenario._unshelve_server.assert_called_once_with(fake_server)286        scenario._delete_server.assert_called_once_with(fake_server,287                                                        force=False)288    def test_list_servers(self):289        scenario = servers.ListServers(self.context)290        scenario._list_servers = mock.MagicMock()291        scenario.run(True)292        scenario._list_servers.assert_called_once_with(True)293    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")294    def test_boot_server_from_volume(self, mock_block_storage):295        fake_server = object()296        scenario = servers.BootServerFromVolume(297            self.context, clients=mock.Mock())298        scenario._boot_server = mock.MagicMock(return_value=fake_server)299        fake_volume = fakes.FakeVolumeManager().create()300        fake_volume.id = "volume_id"301        cinder = mock_block_storage.return_value302        cinder.create_volume.return_value = fake_volume303        scenario.run("img", 0, 5, volume_type=None,304                     auto_assign_nic=False, fakearg="f")305        cinder.create_volume.assert_called_once_with(5, imageRef="img",306                                                     volume_type=None)307        scenario._boot_server.assert_called_once_with(308            None, 0, auto_assign_nic=False,309            block_device_mapping={"vda": "volume_id:::0"},310            fakearg="f")311    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")312    def test_boot_server_from_volume_and_delete(self, mock_block_storage):313        fake_server = object()314        scenario = servers.BootServerFromVolumeAndDelete(315            self.context, clients=mock.Mock())316        scenario._boot_server = mock.MagicMock(return_value=fake_server)317        scenario.sleep_between = mock.MagicMock()318        scenario._delete_server = mock.MagicMock()319        fake_volume = fakes.FakeVolumeManager().create()320        fake_volume.id = "volume_id"321        cinder = mock_block_storage.return_value322        cinder.create_volume.return_value = fake_volume323        scenario.run("img", 0, 5, None, 10, 20, fakearg="f")324        cinder.create_volume.assert_called_once_with(5, imageRef="img",325                                                     volume_type=None)326        scenario._boot_server.assert_called_once_with(327            None, 0,328            block_device_mapping={"vda": "volume_id:::0"},329            fakearg="f")330        scenario.sleep_between.assert_called_once_with(10, 20)331        scenario._delete_server.assert_called_once_with(fake_server,332                                                        force=False)333    def _prepare_boot(self, nic=None, assert_nic=False):334        fake_server = mock.MagicMock()335        scenario = servers.BootServer(self.context)336        scenario._boot_server = mock.MagicMock(return_value=fake_server)337        scenario.generate_random_name = mock.MagicMock(return_value="name")338        kwargs = {"fakearg": "f"}339        expected_kwargs = {"fakearg": "f"}340        assert_nic = nic or assert_nic341        if nic:342            kwargs["nics"] = nic343        if assert_nic:344            self.clients("nova").networks.create("net-1")345            expected_kwargs["nics"] = nic or [{"net-id": "net-2"}]346        return scenario, kwargs, expected_kwargs347    def _verify_boot_server(self, nic=None, assert_nic=False):348        scenario, kwargs, expected_kwargs = self._prepare_boot(349            nic=nic, assert_nic=assert_nic)350        scenario.run("img", 0, **kwargs)351        scenario._boot_server.assert_called_once_with(352            "img", 0, auto_assign_nic=False, **expected_kwargs)353    def test_boot_server_no_nics(self):354        self._verify_boot_server(nic=None, assert_nic=False)355    def test_boot_server_with_nic(self):356        self._verify_boot_server(nic=[{"net-id": "net-1"}], assert_nic=True)357    def test_snapshot_server(self):358        fake_server = object()359        fake_image = fakes.FakeImageManager()._create()360        fake_image.id = "image_id"361        scenario = servers.SnapshotServer(self.context)362        scenario.generate_random_name = mock.MagicMock(return_value="name")363        scenario._boot_server = mock.MagicMock(return_value=fake_server)364        scenario._create_image = mock.MagicMock(return_value=fake_image)365        scenario._delete_server = mock.MagicMock()366        scenario._delete_image = mock.MagicMock()367        scenario.run("i", 0, fakearg=2)368        scenario._boot_server.assert_has_calls([369            mock.call("i", 0, fakearg=2),370            mock.call("image_id", 0, fakearg=2)])371        scenario._create_image.assert_called_once_with(fake_server)372        scenario._delete_server.assert_has_calls([373            mock.call(fake_server, force=False),374            mock.call(fake_server, force=False)])375        scenario._delete_image.assert_called_once_with(fake_image)376    def _test_resize(self, confirm=False):377        fake_server = object()378        fake_image = fakes.FakeImageManager()._create()379        fake_image.id = "image_id"380        flavor = mock.MagicMock()381        to_flavor = mock.MagicMock()382        scenario = servers.ResizeServer(self.context)383        scenario.generate_random_name = mock.MagicMock(return_value="name")384        scenario._boot_server = mock.MagicMock(return_value=fake_server)385        scenario._resize_confirm = mock.MagicMock()386        scenario._resize_revert = mock.MagicMock()387        scenario._resize = mock.MagicMock()388        scenario._delete_server = mock.MagicMock()389        kwargs = {"confirm": confirm}390        scenario.run(fake_image, flavor, to_flavor, **kwargs)391        scenario._resize.assert_called_once_with(fake_server, to_flavor)392        if confirm:393            scenario._resize_confirm.assert_called_once_with(fake_server)394        else:395            scenario._resize_revert.assert_called_once_with(fake_server)396    def test_resize_with_confirm(self):397        self._test_resize(confirm=True)398    def test_resize_with_revert(self):399        self._test_resize(confirm=False)400    @ddt.data({"confirm": True},401              {"confirm": False})402    @ddt.unpack403    def test_resize_shoutoff_server(self, confirm=False):404        fake_server = object()405        flavor = mock.MagicMock()406        to_flavor = mock.MagicMock()407        scenario = servers.ResizeShutoffServer(self.context)408        scenario.generate_random_name = mock.MagicMock(return_value="name")409        scenario._boot_server = mock.MagicMock(return_value=fake_server)410        scenario._stop_server = mock.MagicMock()411        scenario._resize_confirm = mock.MagicMock()412        scenario._resize_revert = mock.MagicMock()413        scenario._resize = mock.MagicMock()414        scenario._delete_server = mock.MagicMock()415        scenario.run("img", flavor, to_flavor, confirm=confirm)416        scenario._boot_server.assert_called_once_with("img", flavor)417        scenario._stop_server.assert_called_once_with(fake_server)418        scenario._resize.assert_called_once_with(fake_server, to_flavor)419        if confirm:420            scenario._resize_confirm.assert_called_once_with(fake_server,421                                                             "SHUTOFF")422        else:423            scenario._resize_revert.assert_called_once_with(fake_server,424                                                            "SHUTOFF")425        scenario._delete_server.assert_called_once_with(fake_server,426                                                        force=False)427    @ddt.data({"confirm": True, "do_delete": True},428              {"confirm": False, "do_delete": True})429    @ddt.unpack430    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")431    def test_boot_server_attach_created_volume_and_resize(432            self, mock_block_storage, confirm=False, do_delete=False):433        fake_volume = mock.MagicMock()434        fake_server = mock.MagicMock()435        flavor = mock.MagicMock()436        to_flavor = mock.MagicMock()437        fake_attachment = mock.MagicMock()438        cinder = mock_block_storage.return_value439        cinder.create_volume.return_value = fake_volume440        scenario = servers.BootServerAttachCreatedVolumeAndResize(441            self.context, clients=mock.Mock())442        scenario.generate_random_name = mock.MagicMock(return_value="name")443        scenario._boot_server = mock.MagicMock(return_value=fake_server)444        scenario._attach_volume = mock.MagicMock(return_value=fake_attachment)445        scenario._resize_confirm = mock.MagicMock()446        scenario._resize_revert = mock.MagicMock()447        scenario._resize = mock.MagicMock()448        scenario._detach_volume = mock.MagicMock()449        scenario._delete_server = mock.MagicMock()450        scenario.sleep_between = mock.MagicMock()451        volume_size = 10452        scenario.run("img", flavor, to_flavor, volume_size, min_sleep=10,453                     max_sleep=20, confirm=confirm, do_delete=do_delete)454        scenario._boot_server.assert_called_once_with("img", flavor)455        cinder.create_volume.assert_called_once_with(volume_size)456        scenario._attach_volume.assert_called_once_with(fake_server,457                                                        fake_volume)458        scenario._detach_volume.assert_called_once_with(fake_server,459                                                        fake_volume)460        scenario.sleep_between.assert_called_once_with(10, 20)461        scenario._resize.assert_called_once_with(fake_server, to_flavor)462        if confirm:463            scenario._resize_confirm.assert_called_once_with(fake_server)464        else:465            scenario._resize_revert.assert_called_once_with(fake_server)466        if do_delete:467            scenario._detach_volume.assert_called_once_with(fake_server,468                                                            fake_volume)469            cinder.delete_volume.assert_called_once_with(fake_volume)470            scenario._delete_server.assert_called_once_with(fake_server,471                                                            force=False)472    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")473    def test_list_attachments(self, mock_block_storage):474        mock_volume_service = mock_block_storage.return_value475        fake_volume = mock.MagicMock()476        fake_server = mock.MagicMock()477        flavor = mock.MagicMock()478        fake_attachment = mock.MagicMock()479        list_attachments = [mock.MagicMock(),480                            fake_attachment,481                            mock.MagicMock()]482        context = self.context483        context.update({484            "admin": {485                "id": "fake_user_id",486                "credential": mock.MagicMock()487            },488            "user": {"id": "fake_user_id",489                     "credential": mock.MagicMock()},490            "tenant": {"id": "fake", "name": "fake",491                       "volumes": [{"id": "uuid", "size": 1}],492                       "servers": [1]}})493        scenario = servers.BootServerAttachVolumeAndListAttachments(494            context)495        scenario._boot_server = mock.MagicMock(return_value=fake_server)496        scenario._attach_volume = mock.MagicMock()497        scenario._list_attachments = mock.MagicMock()498        mock_volume_service.create_volume.return_value = fake_volume499        scenario._list_attachments.return_value = list_attachments500        img_name = "img"501        volume_size = 10502        volume_num = 1503        scenario._attach_volume.return_value = fake_attachment504        scenario.run(img_name, flavor, volume_size, volume_num)505        scenario._boot_server.assert_called_once_with(img_name, flavor)506        mock_volume_service.create_volume.assert_called_once_with(volume_size)507        scenario._attach_volume.assert_called_once_with(fake_server,508                                                        fake_volume)509        scenario._list_attachments.assert_called_once_with(fake_server.id)510    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")511    def test_list_attachments_fails(self, mock_block_storage):512        mock_volume_service = mock_block_storage.return_value513        fake_volume = mock.MagicMock()514        fake_server = mock.MagicMock()515        flavor = mock.MagicMock()516        fake_attachment = mock.MagicMock()517        list_attachments = [mock.MagicMock(),518                            mock.MagicMock(),519                            mock.MagicMock()]520        context = self.context521        context.update({522            "admin": {523                "id": "fake_user_id",524                "credential": mock.MagicMock()525            },526            "user": {"id": "fake_user_id",527                     "credential": mock.MagicMock()},528            "tenant": {"id": "fake", "name": "fake",529                       "volumes": [{"id": "uuid", "size": 1}],530                       "servers": [1]}})531        scenario = servers.BootServerAttachVolumeAndListAttachments(532            context)533        scenario._boot_server = mock.MagicMock(return_value=fake_server)534        mock_volume_service.create_volume.return_value = fake_volume535        scenario._attach_volume = mock.MagicMock()536        scenario._list_attachments = mock.MagicMock()537        scenario._attach_volume.return_value = fake_attachment538        scenario._list_attachments.return_value = list_attachments539        img_name = "img"540        volume_size = 10541        # Negative case: attachment not included into list of542        # available attachments543        self.assertRaises(rally_exceptions.RallyAssertionError,544                          scenario.run,545                          img_name, flavor, volume_size)546        scenario._boot_server.assert_called_with(img_name, flavor)547        mock_volume_service.create_volume.assert_called_with(volume_size)548        scenario._attach_volume.assert_called_with(fake_server,549                                                   fake_volume)550        scenario._list_attachments.assert_called_with(fake_server.id)551    @ddt.data({"confirm": True, "do_delete": True},552              {"confirm": False, "do_delete": True})553    @ddt.unpack554    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")555    def test_boot_server_from_volume_and_resize(556            self, mock_block_storage, confirm=False, do_delete=False):557        fake_server = object()558        flavor = mock.MagicMock()559        to_flavor = mock.MagicMock()560        scenario = servers.BootServerFromVolumeAndResize(self.context,561                                                         clients=mock.Mock())562        scenario._boot_server = mock.MagicMock(return_value=fake_server)563        scenario.generate_random_name = mock.MagicMock(return_value="name")564        scenario._resize_confirm = mock.MagicMock()565        scenario._resize_revert = mock.MagicMock()566        scenario._resize = mock.MagicMock()567        scenario.sleep_between = mock.MagicMock()568        scenario._delete_server = mock.MagicMock()569        fake_volume = fakes.FakeVolumeManager().create()570        fake_volume.id = "volume_id"571        cinder = mock_block_storage.return_value572        cinder.create_volume.return_value = fake_volume573        volume_size = 10574        scenario.run("img", flavor, to_flavor, volume_size, min_sleep=10,575                     max_sleep=20, confirm=confirm, do_delete=do_delete)576        cinder.create_volume.assert_called_once_with(10, imageRef="img")577        scenario._boot_server.assert_called_once_with(578            None, flavor,579            block_device_mapping={"vda": "volume_id:::0"})580        scenario.sleep_between.assert_called_once_with(10, 20)581        scenario._resize.assert_called_once_with(fake_server, to_flavor)582        if confirm:583            scenario._resize_confirm.assert_called_once_with(fake_server)584        else:585            scenario._resize_revert.assert_called_once_with(fake_server)586        if do_delete:587            scenario._delete_server.assert_called_once_with(fake_server,588                                                            force=False)589    def test_boot_and_live_migrate_server(self):590        fake_server = mock.MagicMock()591        scenario = servers.BootAndLiveMigrateServer(self.context)592        scenario.generate_random_name = mock.MagicMock(return_value="name")593        scenario._boot_server = mock.MagicMock(return_value=fake_server)594        scenario.sleep_between = mock.MagicMock()595        scenario._live_migrate = mock.MagicMock()596        scenario._delete_server = mock.MagicMock()597        scenario.run("img", 0, min_sleep=10, max_sleep=20, fakearg="fakearg")598        scenario._boot_server.assert_called_once_with("img", 0,599                                                      fakearg="fakearg")600        scenario.sleep_between.assert_called_once_with(10, 20)601        scenario._live_migrate.assert_called_once_with(fake_server,602                                                       False, False)603        scenario._delete_server.assert_called_once_with(fake_server)604    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")605    def test_boot_server_from_volume_and_live_migrate(self,606                                                      mock_block_storage):607        fake_server = mock.MagicMock()608        scenario = servers.BootServerFromVolumeAndLiveMigrate(609            self.context, clients=mock.Mock())610        scenario.generate_random_name = mock.MagicMock(return_value="name")611        scenario._boot_server = mock.MagicMock(return_value=fake_server)612        scenario.sleep_between = mock.MagicMock()613        scenario._live_migrate = mock.MagicMock()614        scenario._delete_server = mock.MagicMock()615        fake_volume = fakes.FakeVolumeManager().create()616        fake_volume.id = "volume_id"617        cinder = mock_block_storage.return_value618        cinder.create_volume.return_value = fake_volume619        scenario.run("img", 0, 5, volume_type=None,620                     min_sleep=10, max_sleep=20, fakearg="f")621        cinder.create_volume.assert_called_once_with(5, imageRef="img",622                                                     volume_type=None)623        scenario._boot_server.assert_called_once_with(624            None, 0,625            block_device_mapping={"vda": "volume_id:::0"},626            fakearg="f")627        scenario.sleep_between.assert_called_once_with(10, 20)628        scenario._live_migrate.assert_called_once_with(fake_server,629                                                       False, False)630        scenario._delete_server.assert_called_once_with(fake_server,631                                                        force=False)632    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")633    def test_boot_server_attach_created_volume_and_live_migrate(634            self, mock_block_storage):635        fake_volume = mock.MagicMock()636        fake_server = mock.MagicMock()637        fake_attachment = mock.MagicMock()638        clients = mock.Mock()639        cinder = mock_block_storage.return_value640        cinder.create_volume.return_value = fake_volume641        scenario = servers.BootServerAttachCreatedVolumeAndLiveMigrate(642            self.context, clients=clients)643        scenario._attach_volume = mock.MagicMock(return_value=fake_attachment)644        scenario._detach_volume = mock.MagicMock()645        scenario.sleep_between = mock.MagicMock()646        scenario._live_migrate = mock.MagicMock()647        scenario._boot_server = mock.MagicMock(return_value=fake_server)648        scenario._delete_server = mock.MagicMock()649        image = "img"650        flavor = "flavor"651        size = 5652        boot_kwargs = {"some_var": "asd"}653        scenario.run(image, flavor, size, min_sleep=10, max_sleep=20,654                     boot_server_kwargs=boot_kwargs)655        scenario._boot_server.assert_called_once_with(image, flavor,656                                                      **boot_kwargs)657        cinder.create_volume.assert_called_once_with(size)658        scenario._attach_volume.assert_called_once_with(fake_server,659                                                        fake_volume)660        scenario._detach_volume.assert_called_once_with(fake_server,661                                                        fake_volume)662        scenario.sleep_between.assert_called_once_with(10, 20)663        scenario._live_migrate.assert_called_once_with(fake_server,664                                                       False, False)665        cinder.delete_volume.assert_called_once_with(fake_volume)666        scenario._delete_server.assert_called_once_with(fake_server)667    def _test_boot_and_migrate_server(self, confirm=False):668        fake_server = mock.MagicMock()669        scenario = servers.BootAndMigrateServer(self.context)670        scenario.generate_random_name = mock.MagicMock(return_value="name")671        scenario._boot_server = mock.MagicMock(return_value=fake_server)672        scenario._migrate = mock.MagicMock()673        scenario._resize_confirm = mock.MagicMock()674        scenario._resize_revert = mock.MagicMock()675        scenario._delete_server = mock.MagicMock()676        kwargs = {"confirm": confirm}677        scenario.run("img", 0, fakearg="fakearg", **kwargs)678        scenario._boot_server.assert_called_once_with("img", 0,679                                                      fakearg="fakearg",680                                                      confirm=confirm)681        scenario._migrate.assert_called_once_with(fake_server)682        if confirm:683            scenario._resize_confirm.assert_called_once_with(fake_server,684                                                             status="ACTIVE")685        else:686            scenario._resize_revert.assert_called_once_with(fake_server,687                                                            status="ACTIVE")688        scenario._delete_server.assert_called_once_with(fake_server)689    def test_boot_and_migrate_server_with_confirm(self):690        self._test_boot_and_migrate_server(confirm=True)691    def test_boot_and_migrate_server_with_revert(self):692        self._test_boot_and_migrate_server(confirm=False)693    def test_boot_and_rebuild_server(self):694        scenario = servers.BootAndRebuildServer(self.context)695        scenario._boot_server = mock.Mock()696        scenario._rebuild_server = mock.Mock()697        scenario._delete_server = mock.Mock()698        from_image = "img1"699        to_image = "img2"700        flavor = "flavor"701        scenario.run(from_image, to_image, flavor, fakearg="fakearg")702        scenario._boot_server.assert_called_once_with(from_image, flavor,703                                                      fakearg="fakearg")704        server = scenario._boot_server.return_value705        scenario._rebuild_server.assert_called_once_with(server, to_image)706        scenario._delete_server.assert_called_once_with(server)707    def test_boot_and_show_server(self):708        server = fakes.FakeServer()709        image = fakes.FakeImage()710        flavor = fakes.FakeFlavor()711        scenario = servers.BootAndShowServer(self.context)712        scenario._boot_server = mock.MagicMock(return_value=server)713        scenario._show_server = mock.MagicMock()714        scenario.run(image, flavor, fakearg="fakearg")715        scenario._boot_server.assert_called_once_with(image, flavor,716                                                      fakearg="fakearg")717        scenario._show_server.assert_called_once_with(server)718    def test_boot_server_and_list_interfaces(self):719        server = fakes.FakeServer()720        image = fakes.FakeImage()721        flavor = fakes.FakeFlavor()722        scenario = servers.BootServerAndListInterfaces(self.context)723        scenario._boot_server = mock.MagicMock(return_value=server)724        scenario._list_interfaces = mock.MagicMock()725        scenario.run(image, flavor, fakearg="fakearg")726        scenario._boot_server.assert_called_once_with(image, flavor,727                                                      fakearg="fakearg")728        scenario._list_interfaces.assert_called_once_with(server)729    @ddt.data({"length": None},730              {"length": 10})731    @ddt.unpack732    def test_boot_and_get_console_server(self, length):733        server = fakes.FakeServer()734        image = fakes.FakeImage()735        flavor = fakes.FakeFlavor()736        kwargs = {"fakearg": "fakearg"}737        scenario = servers.BootAndGetConsoleOutput(self.context)738        scenario._boot_server = mock.MagicMock(return_value=server)739        scenario._get_server_console_output = mock.MagicMock()740        scenario.run(image, flavor, length, **kwargs)741        scenario._boot_server.assert_called_once_with(image, flavor,742                                                      **kwargs)743        scenario._get_server_console_output.assert_called_once_with(server,744                                                                    length)745    def test_boot_and_get_console_url(self):746        server = fakes.FakeServer()747        image = fakes.FakeImage()748        flavor = fakes.FakeFlavor()749        kwargs = {"fakearg": "fakearg"}750        scenario = servers.BootAndGetConsoleUrl(self.context)751        scenario._boot_server = mock.MagicMock(return_value=server)752        scenario._get_console_url_server = mock.MagicMock()753        scenario.run(image, flavor, console_type="novnc", **kwargs)754        scenario._boot_server.assert_called_once_with(image, flavor,755                                                      **kwargs)756        scenario._get_console_url_server.assert_called_once_with(757            server, "novnc")758    def test_boot_and_associate_floating_ip(self):759        clients = mock.MagicMock(credential=mock.MagicMock(api_info={}))760        neutronclient = clients.neutron.return_value761        floatingip = "floatingip"762        neutronclient.create_floatingip.return_value = {763            "floatingip": floatingip}764        scenario = servers.BootAndAssociateFloatingIp(self.context,765                                                      clients=clients)766        server = mock.Mock()767        scenario._boot_server = mock.Mock(return_value=server)768        scenario._associate_floating_ip = mock.Mock()769        image = "img"770        flavor = "flavor"771        scenario.run(image, flavor, fakearg="fakearg")772        scenario._boot_server.assert_called_once_with(image, flavor,773                                                      fakearg="fakearg")774        neutronclient.create_floatingip.assert_called_once_with(775            {"floatingip": mock.ANY}776        )777        scenario._associate_floating_ip.assert_called_once_with(778            server, floatingip)779        # check ext_network780        neutronclient.list_networks.return_value = {781            "networks": [782                {"id": "id1", "name": "net1", "router:external": True},783                {"id": "id2", "name": "net2", "router:external": True},784                {"id": "id3", "name": "net3", "router:external": True},785            ]786        }787        neutronclient.create_floatingip.reset_mock()788        # case 1: new argument is used789        scenario.run(image, flavor, floating_network="net3")790        neutronclient.create_floatingip.assert_called_once_with(791            {"floatingip": {"description": mock.ANY,792                            "floating_network_id": "id3"}}793        )794        # case 2: new argument is transmitted with an old one795        neutronclient.create_floatingip.reset_mock()796        scenario.run(image, flavor, floating_network="net3",797                     create_floating_ip_args={"ext_network": "net2"})798        neutronclient.create_floatingip.assert_called_once_with(799            {"floatingip": {"description": mock.ANY,800                            "floating_network_id": "id3"}}801        )802        # case 3: new argument is transmitted with an semi-old one803        neutronclient.create_floatingip.reset_mock()804        scenario.run(image, flavor, floating_network="net3",805                     create_floating_ip_args={"floating_network": "net1"})806        neutronclient.create_floatingip.assert_called_once_with(807            {"floatingip": {"description": mock.ANY,808                            "floating_network_id": "id3"}}809        )810        # case 4: only old argument is transmitted811        neutronclient.create_floatingip.reset_mock()812        scenario.run(image, flavor,813                     create_floating_ip_args={"ext_network": "net2"})814        neutronclient.create_floatingip.assert_called_once_with(815            {"floatingip": {"description": mock.ANY,816                            "floating_network_id": "id2"}}817        )818        # case 5: only semi-old argument is transmitted819        neutronclient.create_floatingip.reset_mock()820        scenario.run(image, flavor,821                     create_floating_ip_args={"floating_network": "net1"})822        neutronclient.create_floatingip.assert_called_once_with(823            {"floatingip": {"description": mock.ANY,824                            "floating_network_id": "id1"}}825        )826    def test_boot_server_associate_and_dissociate_floating_ip(self):827        clients = mock.MagicMock(credential=mock.MagicMock(api_info={}))828        neutronclient = clients.neutron.return_value829        floatingip = "floatingip"830        neutronclient.create_floatingip.return_value = {831            "floatingip": floatingip}832        scenario = servers.BootServerAssociateAndDissociateFloatingIP(833            self.context, clients=clients)834        server = mock.Mock()835        scenario._boot_server = mock.Mock(return_value=server)836        scenario._associate_floating_ip = mock.Mock()837        scenario._dissociate_floating_ip = mock.Mock()838        image = "img"839        flavor = "flavor"840        scenario.run(image, flavor, fakearg="fakearg")841        scenario._boot_server.assert_called_once_with(image, flavor,842                                                      fakearg="fakearg")843        neutronclient.create_floatingip.assert_called_once_with(844            {"floatingip": mock.ANY}845        )846        scenario._associate_floating_ip.assert_called_once_with(847            server, floatingip)848        scenario._dissociate_floating_ip.assert_called_once_with(849            server, floatingip)850        # check ext_network851        neutronclient.list_networks.return_value = {852            "networks": [853                {"id": "id1", "name": "net1", "router:external": True},854                {"id": "id2", "name": "net2", "router:external": True},855                {"id": "id3", "name": "net3", "router:external": True},856            ]857        }858        neutronclient.create_floatingip.reset_mock()859        # case 1: new argument is used860        scenario.run(image, flavor, floating_network="net3")861        neutronclient.create_floatingip.assert_called_once_with(862            {"floatingip": {"description": mock.ANY,863                            "floating_network_id": "id3"}}864        )865        # case 2: new argument is transmitted with an old one866        neutronclient.create_floatingip.reset_mock()867        scenario.run(image, flavor, floating_network="net3",868                     create_floating_ip_args={"ext_network": "net2"})869        neutronclient.create_floatingip.assert_called_once_with(870            {"floatingip": {"description": mock.ANY,871                            "floating_network_id": "id3"}}872        )873        # case 3: new argument is transmitted with an semi-old one874        neutronclient.create_floatingip.reset_mock()875        scenario.run(image, flavor, floating_network="net3",876                     create_floating_ip_args={"floating_network": "net1"})877        neutronclient.create_floatingip.assert_called_once_with(878            {"floatingip": {"description": mock.ANY,879                            "floating_network_id": "id3"}}880        )881        # case 4: only old argument is transmitted882        neutronclient.create_floatingip.reset_mock()883        scenario.run(image, flavor,884                     create_floating_ip_args={"ext_network": "net2"})885        neutronclient.create_floatingip.assert_called_once_with(886            {"floatingip": {"description": mock.ANY,887                            "floating_network_id": "id2"}}888        )889        # case 5: only semi-old argument is transmitted890        neutronclient.create_floatingip.reset_mock()891        scenario.run(image, flavor,892                     create_floating_ip_args={"floating_network": "net1"})893        neutronclient.create_floatingip.assert_called_once_with(894            {"floatingip": {"description": mock.ANY,895                            "floating_network_id": "id1"}}896        )897    def test_boot_and_update_server(self):898        scenario = servers.BootAndUpdateServer(self.context)899        scenario._boot_server = mock.Mock()900        scenario._update_server = mock.Mock()901        scenario.run("img", "flavor", "desp", fakearg="fakearg")902        scenario._boot_server.assert_called_once_with("img", "flavor",903                                                      fakearg="fakearg")904        scenario._update_server.assert_called_once_with(905            scenario._boot_server.return_value, "desp")906    def test_boot_server_and_attach_interface(self):907        network_create_args = {"router:external": True}908        subnet_create_args = {"allocation_pools": []}909        subnet_cidr_start = "10.1.0.0/16"910        boot_server_args = {}911        net = mock.MagicMock()912        subnet = mock.MagicMock()913        server = mock.MagicMock()914        scenario = servers.BootServerAndAttachInterface(self.context)915        scenario._get_or_create_network = mock.Mock(return_value=net)916        scenario._create_subnet = mock.Mock(return_value=subnet)917        scenario._boot_server = mock.Mock(return_value=server)918        scenario._attach_interface = mock.Mock()919        scenario.run("image", "flavor",920                     network_create_args=network_create_args,921                     subnet_create_args=subnet_create_args,922                     subnet_cidr_start=subnet_cidr_start,923                     boot_server_args=boot_server_args)924        scenario._get_or_create_network.assert_called_once_with(925            network_create_args)926        scenario._create_subnet.assert_called_once_with(net,927                                                        subnet_create_args,928                                                        subnet_cidr_start)929        scenario._boot_server.assert_called_once_with("image", "flavor",930                                                      **boot_server_args)931        scenario._attach_interface.assert_called_once_with(932            server, net_id=net["network"]["id"])933    @mock.patch("rally_openstack.common.services.storage.block.BlockStorage")934    def test_boot_server_from_volume_snapshot(self, mock_block_storage):935        fake_volume = mock.MagicMock(id="volume_id")936        fake_snapshot = mock.MagicMock(id="snapshot_id")937        cinder = mock_block_storage.return_value938        cinder.create_volume.return_value = fake_volume939        cinder.create_snapshot.return_value = fake_snapshot940        scenario = servers.BootServerFromVolumeSnapshot(self.context,941                                                        clients=mock.Mock())942        scenario._boot_server = mock.MagicMock()943        scenario.run("img", "flavor", 1, volume_type=None,944                     auto_assign_nic=False, fakearg="f")945        cinder.create_volume.assert_called_once_with(1, imageRef="img",946                                                     volume_type=None)947        cinder.create_snapshot.assert_called_once_with("volume_id",948                                                       force=False)949        scenario._boot_server.assert_called_once_with(950            None, "flavor", auto_assign_nic=False,951            block_device_mapping={"vda": "snapshot_id:snap::1"},...

test_volumes.py

Source:test_volumes.py

1# Copyright 2013 Huawei Technologies Co.,LTD.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15from unittest import mock16import ddt17from rally_openstack.task.scenarios.cinder import volumes18from tests.unit import test19CINDER_VOLUMES = ("rally_openstack.task.scenarios.cinder.volumes")20@ddt.ddt21class CinderServersTestCase(test.ScenarioTestCase):22    def _get_context(self):23        context = test.get_test_context()24        context.update({25            "admin": {26                "id": "fake_user_id",27                "credential": mock.MagicMock()28            },29            "user": {"id": "fake_user_id",30                     "credential": mock.MagicMock()},31            "tenant": {"id": "fake", "name": "fake",32                       "volumes": [{"id": "uuid", "size": 1}],33                       "servers": [1]}})34        return context35    def setUp(self):36        super(CinderServersTestCase, self).setUp()37        patch = mock.patch(38            "rally_openstack.common.services.storage.block.BlockStorage")39        self.addCleanup(patch.stop)40        self.mock_cinder = patch.start()41    def test_create_and_list_volume(self):42        mock_service = self.mock_cinder.return_value43        scenario = volumes.CreateAndListVolume(self._get_context())44        scenario.run(1, True, fakearg="f")45        mock_service.create_volume.assert_called_once_with(1, fakearg="f")46        mock_service.list_volumes.assert_called_once_with(True)47    def test_create_and_get_volume(self):48        mock_service = self.mock_cinder.return_value49        scenario = volumes.CreateAndGetVolume(self._get_context())50        scenario.run(1, fakearg="f")51        mock_service.create_volume.assert_called_once_with(1, fakearg="f")52        mock_service.get_volume.assert_called_once_with(53            mock_service.create_volume.return_value.id)54    def test_list_volumes(self):55        mock_service = self.mock_cinder.return_value56        scenario = volumes.ListVolumes(self._get_context())57        scenario.run(True)58        mock_service.list_volumes.assert_called_once_with(59            True, limit=None, marker=None, search_opts=None, sort=None)60    def test_list_types(self):61        mock_service = self.mock_cinder.return_value62        scenario = volumes.ListTypes(self._get_context())63        scenario.run(None, is_public=None)64        mock_service.list_types.assert_called_once_with(None,65                                                        is_public=None)66    def test_list_transfers(self):67        mock_service = self.mock_cinder.return_value68        scenario = volumes.ListTransfers(self._get_context())69        scenario._list_transfers = mock.MagicMock()70        scenario.run(True, search_opts=None)71        mock_service.list_transfers.assert_called_once_with(72            True, search_opts=None)73    @ddt.data({"update_args": {"description": "desp"},74               "expected": {"description": "desp"}},75              {"update_args": {"update_name": True, "description": "desp"},76               "expected": {"name": "new_name", "description": "desp"}})77    @ddt.unpack78    def test_create_and_update_volume(self, update_args, expected):79        mock_service = self.mock_cinder.return_value80        scenario = volumes.CreateAndUpdateVolume(self._get_context())81        scenario.generate_random_name = mock.MagicMock()82        scenario.generate_random_name.return_value = "new_name"83        scenario.run(1, update_volume_kwargs=update_args)84        mock_service.create_volume.assert_called_once_with(1)85        mock_service.update_volume.assert_called_once_with(86            mock_service.create_volume.return_value, **expected)87        if update_args.get("update_name", False):88            scenario.generate_random_name.assert_called_once_with()89    def test_create_volume_and_update_readonly_flag(self):90        mock_service = self.mock_cinder.return_value91        scenario = volumes.CreateVolumeAndUpdateReadonlyFlag(92            self._get_context())93        scenario.run(1, image=None, read_only=True, fakearg="f")94        mock_service.create_volume.assert_called_once_with(1, fakearg="f")95        mock_service.update_readonly_flag.assert_called_once_with(96            mock_service.create_volume.return_value.id, read_only=True)97    def test_create_and_delete_volume(self):98        mock_service = self.mock_cinder.return_value99        scenario = volumes.CreateAndDeleteVolume(self._get_context())100        scenario.sleep_between = mock.MagicMock()101        scenario.run(size=1, min_sleep=10, max_sleep=20, fakearg="f")102        mock_service.create_volume.assert_called_once_with(1, fakearg="f")103        scenario.sleep_between.assert_called_once_with(10, 20)104        mock_service.delete_volume.assert_called_once_with(105            mock_service.create_volume.return_value)106    def test_create_volume(self):107        mock_service = self.mock_cinder.return_value108        scenario = volumes.CreateVolume(self._get_context())109        scenario.run(1, fakearg="f")110        mock_service.create_volume.assert_called_once_with(1, fakearg="f")111    def test_create_volume_and_modify_metadata(self):112        mock_service = self.mock_cinder.return_value113        scenario = volumes.ModifyVolumeMetadata(self._get_context())114        scenario.run(sets=5, set_size=4, deletes=3, delete_size=2)115        mock_service.set_metadata.assert_called_once_with(116            "uuid", set_size=4, sets=5)117        mock_service.delete_metadata.assert_called_once_with(118            "uuid",119            keys=mock_service.set_metadata.return_value,120            deletes=3, delete_size=2)121    def test_create_and_extend_volume(self):122        mock_service = self.mock_cinder.return_value123        scenario = volumes.CreateAndExtendVolume(self._get_context())124        scenario.sleep_between = mock.MagicMock()125        scenario.run(1, 2, 10, 20, fakearg="f")126        mock_service.create_volume.assert_called_once_with(1, fakearg="f")127        mock_service.extend_volume.assert_called_once_with(128            mock_service.create_volume.return_value, new_size=2)129        scenario.sleep_between.assert_called_once_with(10, 20)130        mock_service.delete_volume.assert_called_once_with(131            mock_service.create_volume.return_value)132    def test_create_from_image_and_delete_volume(self):133        mock_service = self.mock_cinder.return_value134        scenario = volumes.CreateAndDeleteVolume(self._get_context())135        scenario.run(1, image="fake_image")136        mock_service.create_volume.assert_called_once_with(137            1, imageRef="fake_image")138        mock_service.delete_volume.assert_called_once_with(139            mock_service.create_volume.return_value)140    def test_create_volume_from_image(self):141        mock_service = self.mock_cinder.return_value142        scenario = volumes.CreateVolume(self._get_context())143        scenario.run(1, image="fake_image")144        mock_service.create_volume.assert_called_once_with(145            1, imageRef="fake_image")146    def test_create_volume_from_image_and_list(self):147        mock_service = self.mock_cinder.return_value148        scenario = volumes.CreateAndListVolume(self._get_context())149        scenario.run(1, True, "fake_image")150        mock_service.create_volume.assert_called_once_with(151            1, imageRef="fake_image")152        mock_service.list_volumes.assert_called_once_with(True)153    def test_create_from_volume_and_delete_volume(self):154        mock_service = self.mock_cinder.return_value155        vol_size = 1156        scenario = volumes.CreateFromVolumeAndDeleteVolume(self._get_context())157        scenario.run(vol_size)158        mock_service.create_volume.assert_called_once_with(159            1, source_volid="uuid")160        mock_service.delete_volume.assert_called_once_with(161            mock_service.create_volume.return_value)162    @mock.patch("%s.CreateAndDeleteSnapshot.sleep_between" % CINDER_VOLUMES)163    def test_create_and_delete_snapshot(self, mock_sleep_between):164        mock_service = self.mock_cinder.return_value165        scenario = volumes.CreateAndDeleteSnapshot(self._get_context())166        scenario.run(False, 10, 20, fakearg="f")167        mock_service.create_snapshot.assert_called_once_with("uuid",168                                                             force=False,169                                                             fakearg="f")170        mock_sleep_between.assert_called_once_with(10, 20)171        mock_service.delete_snapshot.assert_called_once_with(172            mock_service.create_snapshot.return_value)173    def test_create_and_list_snapshots(self):174        mock_service = self.mock_cinder.return_value175        scenario = volumes.CreateAndListSnapshots(self._get_context())176        scenario.run(False, True, fakearg="f")177        mock_service.create_snapshot.assert_called_once_with("uuid",178                                                             force=False,179                                                             fakearg="f")180        mock_service.list_snapshots.assert_called_once_with(True)181    def test_create_and_attach_volume(self):182        fake_server = mock.MagicMock()183        mock_service = self.mock_cinder.return_value184        scenario = volumes.CreateAndAttachVolume(self._get_context())185        scenario._boot_server = mock.MagicMock(return_value=fake_server)186        scenario._delete_server = mock.MagicMock()187        scenario._attach_volume = mock.MagicMock()188        scenario._detach_volume = mock.MagicMock()189        volume_args = {"some_key": "some_val"}190        vm_args = {"some_key": "some_val"}191        scenario.run(10, "img", "0",192                     create_volume_params=volume_args,193                     create_vm_params=vm_args)194        mock_service.create_volume.assert_called_once_with(195            10, **volume_args)196        scenario._attach_volume.assert_called_once_with(197            fake_server, mock_service.create_volume.return_value)198        scenario._detach_volume.assert_called_once_with(199            fake_server, mock_service.create_volume.return_value)200        mock_service.delete_volume.assert_called_once_with(201            mock_service.create_volume.return_value)202        scenario._delete_server.assert_called_once_with(fake_server)203    @mock.patch("rally_openstack.common.services.image.image.Image")204    def test_create_and_upload_volume_to_image(self, mock_image):205        mock_volume_service = self.mock_cinder.return_value206        mock_image_service = mock_image.return_value207        scenario = volumes.CreateAndUploadVolumeToImage(self._get_context())208        scenario.run(2, image="img", container_format="fake",209                     disk_format="disk", do_delete=False, fakeargs="fakeargs")210        mock_volume_service.create_volume.assert_called_once_with(211            2, imageRef="img", fakeargs="fakeargs")212        mock_volume_service.upload_volume_to_image.assert_called_once_with(213            mock_volume_service.create_volume.return_value,214            container_format="fake", disk_format="disk", force=False)215        mock_volume_service.create_volume.reset_mock()216        mock_volume_service.upload_volume_to_image.reset_mock()217        scenario.run(1, image=None, do_delete=True, fakeargs="fakeargs")218        mock_volume_service.create_volume.assert_called_once_with(219            1, fakeargs="fakeargs")220        mock_volume_service.upload_volume_to_image.assert_called_once_with(221            mock_volume_service.create_volume.return_value,222            container_format="bare", disk_format="raw", force=False)223        mock_volume_service.delete_volume.assert_called_once_with(224            mock_volume_service.create_volume.return_value)225        mock_image_service.delete_image.assert_called_once_with(226            mock_volume_service.upload_volume_to_image.return_value.id)227    def test_create_snapshot_and_attach_volume(self):228        mock_service = self.mock_cinder.return_value229        scenario = volumes.CreateSnapshotAndAttachVolume(self._get_context())230        scenario._boot_server = mock.MagicMock()231        scenario._attach_volume = mock.MagicMock()232        scenario._detach_volume = mock.MagicMock()233        scenario.run("img", "flavor")234        self.assertTrue(mock_service.create_volume.called)235        volume = mock_service.create_volume.return_value236        snapshot = mock_service.create_snapshot.return_value237        mock_service.create_snapshot.assert_called_once_with(volume.id,238                                                             force=False)239        mock_service.delete_snapshot.assert_called_once_with(snapshot)240        scenario._attach_volume.assert_called_once_with(241            scenario._boot_server.return_value, volume)242        scenario._detach_volume.assert_called_once_with(243            scenario._boot_server.return_value, volume)244        mock_service.delete_volume.assert_called_once_with(volume)245    @mock.patch("random.choice")246    def test_create_snapshot_and_attach_volume_use_volume_type_with_name(247            self, mock_choice):248        mock_service = self.mock_cinder.return_value249        scenario = volumes.CreateSnapshotAndAttachVolume(self._get_context())250        scenario._boot_server = mock.MagicMock()251        scenario._attach_volume = mock.MagicMock()252        scenario._detach_volume = mock.MagicMock()253        scenario.run("img", "flavor", volume_type="type")254        fake_volume = mock_service.create_volume.return_value255        fake_server = scenario._boot_server.return_value256        fake_snapshot = mock_service.create_snapshot.return_value257        mock_service.create_volume.assert_called_once_with(258            {"min": 1, "max": 5}, volume_type="type")259        mock_service.create_snapshot.assert_called_once_with(fake_volume.id,260                                                             force=False)261        mock_service.delete_snapshot.assert_called_once_with(fake_snapshot)262        scenario._attach_volume.assert_called_once_with(fake_server,263                                                        fake_volume)264        scenario._detach_volume.assert_called_once_with(fake_server,265                                                        fake_volume)266        mock_service.delete_volume.assert_called_once_with(fake_volume)267    @mock.patch("random.randint")268    def test_create_nested_snapshots_and_attach_volume(self, mock_randint):269        mock_service = self.mock_cinder.return_value270        mock_randint.return_value = 2271        volume_kwargs = {"volume_type": "type1"}272        snapshot_kwargs = {"name": "snapshot1", "description": "snaphot one"}273        scenario = volumes.CreateNestedSnapshotsAndAttachVolume(274            context=self._get_context())275        scenario._boot_server = mock.MagicMock()276        scenario._attach_volume = mock.MagicMock()277        scenario._detach_volume = mock.MagicMock()278        scenario.run("img", "flavor", create_volume_kwargs=volume_kwargs,279                     create_snapshot_kwargs=snapshot_kwargs)280        mock_service.create_volume.assert_called_once_with(281            mock_randint.return_value, **volume_kwargs)282        mock_service.create_snapshot.assert_called_once_with(283            mock_service.create_volume.return_value.id, force=False,284            **snapshot_kwargs)285        scenario._attach_volume(scenario._boot_server.return_value,286                                mock_service.create_volume.return_value)287        mock_service.delete_volume.assert_called_once_with(288            mock_service.create_volume.return_value)289        mock_service.delete_snapshot.assert_called_once_with(290            mock_service.create_snapshot.return_value)291        scenario._detach_volume.assert_called_once_with(292            scenario._boot_server.return_value,293            mock_service.create_volume.return_value)294    @mock.patch("random.randint")295    def test_create_nested_snapshots_and_attach_volume_2(self, mock_randint):296        mock_service = self.mock_cinder.return_value297        mock_randint.return_value = 2298        nested_level = 3299        volume_size = mock_randint.return_value300        fake_volumes = [mock.Mock(size=volume_size)301                        for i in range(nested_level)]302        fake_snapshots = [mock.Mock()303                          for i in range(nested_level)]304        mock_service.create_volume.side_effect = fake_volumes305        mock_service.create_snapshot.side_effect = fake_snapshots306        scenario = volumes.CreateNestedSnapshotsAndAttachVolume(307            context=self._get_context())308        scenario._boot_server = mock.MagicMock()309        scenario._attach_volume = mock.MagicMock()310        scenario._detach_volume = mock.MagicMock()311        scenario.run("img", "flavor", nested_level=nested_level)312        expected_volumes = [mock.call(volume_size)]313        expected_snapshots = [mock.call(fake_volumes[0].id, force=False)]314        expected_attachs = [mock.call(scenario._boot_server.return_value,315                                      fake_volumes[0])]316        for i in range(nested_level - 1):317            expected_volumes.append(318                mock.call(volume_size, snapshot_id=fake_snapshots[i].id))319            expected_snapshots.append(320                mock.call(fake_volumes[i + 1].id, force=False))321            expected_attachs.append(322                mock.call(scenario._boot_server.return_value,323                          fake_volumes[i + 1]))324        mock_service.create_volume.assert_has_calls(expected_volumes)325        mock_service.create_snapshot.assert_has_calls(expected_snapshots)326        scenario._attach_volume.assert_has_calls(expected_attachs)327        fake_volumes.reverse()328        fake_snapshots.reverse()329        mock_service.delete_volume.assert_has_calls(330            [mock.call(volume) for volume in fake_volumes])331        mock_service.delete_snapshot.assert_has_calls(332            [mock.call(snapshot) for snapshot in fake_snapshots])333        scenario._detach_volume.assert_has_calls(334            [mock.call(scenario._boot_server.return_value,335                       fake_volumes[i])336             for i in range(len(fake_volumes))])337    def test_create_volume_backup(self):338        mock_service = self.mock_cinder.return_value339        scenario = volumes.CreateVolumeBackup(self._get_context())340        volume_kwargs = {"some_var": "zaq"}341        scenario.run(1, do_delete=True, create_volume_kwargs=volume_kwargs)342        mock_service.create_volume.assert_called_once_with(1, **volume_kwargs)343        mock_service.create_backup.assert_called_once_with(344            mock_service.create_volume.return_value.id)345        mock_service.delete_volume.assert_called_once_with(346            mock_service.create_volume.return_value)347        mock_service.delete_backup.assert_called_once_with(348            mock_service.create_backup.return_value)349    def test_create_volume_backup_no_delete(self):350        mock_service = self.mock_cinder.return_value351        scenario = volumes.CreateVolumeBackup(self._get_context())352        volume_kwargs = {"some_var": "zaq"}353        scenario.run(1, do_delete=False, create_volume_kwargs=volume_kwargs)354        mock_service.create_volume.assert_called_once_with(1, **volume_kwargs)355        mock_service.create_backup.assert_called_once_with(356            mock_service.create_volume.return_value.id)357        self.assertFalse(mock_service.delete_volume.called)358        self.assertFalse(mock_service.delete_backup.called)359    def test_create_and_restore_volume_backup(self):360        mock_service = self.mock_cinder.return_value361        volume_kwargs = {"some_var": "zaq"}362        scenario = volumes.CreateAndRestoreVolumeBackup(self._get_context())363        scenario.run(1, do_delete=True, create_volume_kwargs=volume_kwargs)364        fake_volume = mock_service.create_volume.return_value365        fake_backup = mock_service.create_backup.return_value366        mock_service.create_volume.assert_called_once_with(1, **volume_kwargs)367        mock_service.create_backup.assert_called_once_with(fake_volume.id)368        mock_service.restore_backup.assert_called_once_with(fake_backup.id)369        mock_service.delete_volume.assert_called_once_with(fake_volume)370        mock_service.delete_backup.assert_called_once_with(fake_backup)371    def test_create_and_restore_volume_backup_no_delete(self):372        mock_service = self.mock_cinder.return_value373        volume_kwargs = {"some_var": "zaq"}374        scenario = volumes.CreateAndRestoreVolumeBackup(self._get_context())375        scenario.run(1, do_delete=False, create_volume_kwargs=volume_kwargs)376        fake_volume = mock_service.create_volume.return_value377        fake_backup = mock_service.create_backup.return_value378        mock_service.create_volume.assert_called_once_with(1, **volume_kwargs)379        mock_service.create_backup.assert_called_once_with(fake_volume.id)380        mock_service.restore_backup.assert_called_once_with(fake_backup.id)381        self.assertFalse(mock_service.delete_volume.called)382        self.assertFalse(mock_service.delete_backup.called)383    def test_create_and_list_volume_backups(self):384        mock_service = self.mock_cinder.return_value385        volume_kwargs = {"some_var": "zaq"}386        scenario = volumes.CreateAndListVolumeBackups(self._get_context())387        scenario.run(1, detailed=True, do_delete=True,388                     create_volume_kwargs=volume_kwargs)389        fake_volume = mock_service.create_volume.return_value390        fake_backup = mock_service.create_backup.return_value391        mock_service.create_volume.assert_called_once_with(1, **volume_kwargs)392        mock_service.create_backup.assert_called_once_with(fake_volume.id)393        mock_service.list_backups.assert_called_once_with(True)394        mock_service.delete_volume.assert_called_once_with(fake_volume)395        mock_service.delete_backup.assert_called_once_with(fake_backup)396    def test_create_and_list_volume_backups_no_delete(self):397        mock_service = self.mock_cinder.return_value398        volume_kwargs = {"some_var": "zaq"}399        scenario = volumes.CreateAndListVolumeBackups(self._get_context())400        scenario.run(1, detailed=True, do_delete=False,401                     create_volume_kwargs=volume_kwargs)402        fake_volume = mock_service.create_volume.return_value403        mock_service.create_volume.assert_called_once_with(1, **volume_kwargs)404        mock_service.create_backup.assert_called_once_with(fake_volume.id)405        mock_service.list_backups.assert_called_once_with(True)406        self.assertFalse(mock_service.delete_volume.called)407        self.assertFalse(mock_service.delete_backup.called)408    @ddt.data({},409              {"nested_level": 2},410              {"image": "img"})411    @ddt.unpack412    def test_create_volume_and_clone(self, nested_level=1,413                                     image=None):414        create_volumes_count = nested_level + 1415        fake_volumes = [mock.Mock(size=1)416                        for i in range(create_volumes_count)]417        mock_service = self.mock_cinder.return_value418        mock_service.create_volume.side_effect = fake_volumes419        scenario = volumes.CreateVolumeAndClone(self._get_context())420        scenario.run(1, image=image, nested_level=nested_level,421                     fakearg="fake")422        expected = [mock.call(1, imageRef=image, fakearg="fake")423                    if image else mock.call(1, fakearg="fake")]424        for i in range(nested_level):425            expected.append(mock.call(fake_volumes[i].size,426                                      source_volid=fake_volumes[i].id,427                                      fakearg="fake")428                            )429            self._test_atomic_action_timer(scenario.atomic_actions(),430                                           "cinder.clone_volume",431                                           count=nested_level)432        mock_service.create_volume.assert_has_calls(expected)433    def test_create_volume_from_snapshot(self):434        mock_service = self.mock_cinder.return_value435        create_snapshot_args = {"force": False}436        scenario = volumes.CreateVolumeFromSnapshot(self._get_context())437        scenario.run(fakearg="f")438        fake_snapshot = mock_service.create_snapshot.return_value439        fake_volume = mock_service.create_volume.return_value440        mock_service.create_snapshot.assert_called_once_with("uuid")441        mock_service.create_volume.assert_called_once_with(442            1, snapshot_id=fake_snapshot.id, fakearg="f")443        mock_service.delete_snapshot.assert_called_once_with(fake_snapshot)444        mock_service.delete_volume.assert_called_once_with(fake_volume)445        mock_service.create_snapshot.reset_mock()446        mock_service.create_volume.reset_mock()447        mock_service.delete_snapshot.reset_mock()448        mock_service.delete_volume.reset_mock()449        scenario.run(do_delete=False,450                     create_snapshot_kwargs=create_snapshot_args,451                     fakearg="f")452        mock_service.create_snapshot.assert_called_once_with(453            "uuid", **create_snapshot_args)454        mock_service.create_volume.assert_called_once_with(455            1, snapshot_id=fake_snapshot.id, fakearg="f")456        self.assertFalse(mock_service.delete_snapshot.called)457        self.assertFalse(mock_service.delete_volume.called)458    @ddt.data({},459              {"image": "img"})460    @ddt.unpack461    def test_create_and_accept_transfer(self, image=None):462        mock_service = self.mock_cinder.return_value463        scenario = volumes.CreateAndAcceptTransfer(self._get_context())464        scenario.run(1, image=image, fakearg="fake")465        expected = [mock.call(1, imageRef=image, fakearg="fake")466                    if image else mock.call(1, fakearg="fake")]467        mock_service.create_volume.assert_has_calls(expected)468        mock_service.transfer_create.assert_called_once_with(469            mock_service.create_volume.return_value.id)470        mock_service.transfer_accept.assert_called_once_with(471            mock_service.transfer_create.return_value.id,...

test_basic.py

Source:test_basic.py

1# Copyright 2013: Mirantis Inc.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15from unittest import mock16import ddt17from rally import exceptions18from rally_openstack.task.scenarios.keystone import basic19from tests.unit import test20@ddt.ddt21class KeystoneBasicTestCase(test.ScenarioTestCase):22    def get_test_context(self):23        context = super(KeystoneBasicTestCase, self).get_test_context()24        context.update({25            "admin": {26                "id": "fake_user_id",27                "credential": mock.MagicMock()28            },29            "user": {30                "id": "fake_user_id",31                "credential": mock.MagicMock()32            },33            "tenant": {"id": "fake_tenant_id",34                       "name": "fake_tenant_name"}35        })36        return context37    def setUp(self):38        super(KeystoneBasicTestCase, self).setUp()39        patch = mock.patch(40            "rally_openstack.common.services.identity.identity.Identity")41        self.addCleanup(patch.stop)42        self.mock_identity = patch.start()43    def test_create_user(self):44        scenario = basic.CreateUser(self.context)45        scenario.run(password="tttt", project_id="id")46        self.mock_identity.return_value.create_user.assert_called_once_with(47            password="tttt", project_id="id")48    def test_create_delete_user(self):49        identity_service = self.mock_identity.return_value50        fake_email = "abcd"51        fake_user = identity_service.create_user.return_value52        scenario = basic.CreateDeleteUser(self.context)53        scenario.run(email=fake_email, enabled=True)54        identity_service.create_user.assert_called_once_with(55            email=fake_email, enabled=True)56        identity_service.delete_user.assert_called_once_with(fake_user.id)57    def test_create_user_set_enabled_and_delete(self):58        identity_service = self.mock_identity.return_value59        scenario = basic.CreateUserSetEnabledAndDelete(self.context)60        fake_email = "abcd"61        fake_user = identity_service.create_user.return_value62        scenario.run(enabled=True, email=fake_email)63        identity_service.create_user.assert_called_once_with(64            email=fake_email, enabled=True)65        identity_service.update_user.assert_called_once_with(66            fake_user.id, enabled=False)67        identity_service.delete_user.assert_called_once_with(fake_user.id)68    def test_user_authenticate_and_validate_token(self):69        identity_service = self.mock_identity.return_value70        scenario = basic.AuthenticateUserAndValidateToken(self.context)71        fake_token = identity_service.fetch_token.return_value72        scenario.run()73        identity_service.fetch_token.assert_called_once_with()74        identity_service.validate_token.assert_called_once_with(fake_token)75    def test_create_tenant(self):76        scenario = basic.CreateTenant(self.context)77        scenario.run(enabled=True)78        self.mock_identity.return_value.create_project.assert_called_once_with(79            enabled=True)80    def test_create_tenant_with_users(self):81        identity_service = self.mock_identity.return_value82        fake_project = identity_service.create_project.return_value83        number_of_users = 184        scenario = basic.CreateTenantWithUsers(self.context)85        scenario.run(users_per_tenant=number_of_users, enabled=True)86        identity_service.create_project.assert_called_once_with(enabled=True)87        identity_service.create_users.assert_called_once_with(88            fake_project.id, number_of_users=number_of_users)89    def test_create_and_list_users(self):90        scenario = basic.CreateAndListUsers(self.context)91        passwd = "tttt"92        project_id = "id"93        scenario.run(password=passwd, project_id=project_id)94        self.mock_identity.return_value.create_user.assert_called_once_with(95            password=passwd, project_id=project_id)96        self.mock_identity.return_value.list_users.assert_called_once_with()97    def test_create_and_list_tenants(self):98        identity_service = self.mock_identity.return_value99        scenario = basic.CreateAndListTenants(self.context)100        scenario.run(enabled=True)101        identity_service.create_project.assert_called_once_with(enabled=True)102        identity_service.list_projects.assert_called_once_with()103    def test_assign_and_remove_user_role(self):104        fake_tenant = self.context["tenant"]["id"]105        fake_user = self.context["user"]["id"]106        fake_role = mock.MagicMock()107        self.mock_identity.return_value.create_role.return_value = fake_role108        scenario = basic.AddAndRemoveUserRole(self.context)109        scenario.run()110        self.mock_identity.return_value.create_role.assert_called_once_with()111        self.mock_identity.return_value.add_role.assert_called_once_with(112            role_id=fake_role.id, user_id=fake_user, project_id=fake_tenant)113        self.mock_identity.return_value.revoke_role.assert_called_once_with(114            fake_role.id, user_id=fake_user, project_id=fake_tenant)115    def test_create_and_delete_role(self):116        fake_role = mock.MagicMock()117        self.mock_identity.return_value.create_role.return_value = fake_role118        scenario = basic.CreateAndDeleteRole(self.context)119        scenario.run()120        self.mock_identity.return_value.create_role.assert_called_once_with()121        self.mock_identity.return_value.delete_role.assert_called_once_with(122            fake_role.id)123    def test_create_and_get_role(self):124        fake_role = mock.MagicMock()125        self.mock_identity.return_value.create_role.return_value = fake_role126        scenario = basic.CreateAndGetRole(self.context)127        scenario.run()128        self.mock_identity.return_value.create_role.assert_called_once_with()129        self.mock_identity.return_value.get_role.assert_called_once_with(130            fake_role.id)131    def test_create_and_list_user_roles(self):132        scenario = basic.CreateAddAndListUserRoles(self.context)133        fake_tenant = self.context["tenant"]["id"]134        fake_user = self.context["user"]["id"]135        fake_role = mock.MagicMock()136        self.mock_identity.return_value.create_role.return_value = fake_role137        scenario.run()138        self.mock_identity.return_value.create_role.assert_called_once_with()139        self.mock_identity.return_value.add_role.assert_called_once_with(140            user_id=fake_user, role_id=fake_role.id, project_id=fake_tenant)141        self.mock_identity.return_value.list_roles.assert_called_once_with(142            user_id=fake_user, project_id=fake_tenant)143    def test_create_and_list_roles(self):144        # Positive case145        scenario = basic.CreateAddListRoles(self.context)146        create_kwargs = {"fakewargs": "name"}147        list_kwargs = {"fakewargs": "f"}148        self.mock_identity.return_value.create_role = mock.Mock(149            return_value="role1")150        self.mock_identity.return_value.list_roles = mock.Mock(151            return_value=("role1", "role2"))152        scenario.run(create_role_kwargs=create_kwargs,153                     list_role_kwargs=list_kwargs)154        self.mock_identity.return_value.create_role.assert_called_once_with(155            **create_kwargs)156        self.mock_identity.return_value.list_roles.assert_called_once_with(157            **list_kwargs)158        # Negative case 1: role isn't created159        self.mock_identity.return_value.create_role.return_value = None160        self.assertRaises(exceptions.RallyAssertionError,161                          scenario.run, create_role_kwargs=create_kwargs,162                          list_role_kwargs=list_kwargs)163        self.mock_identity.return_value.create_role.assert_called_with(164            **create_kwargs)165        # Negative case 2: role was created but included into list166        self.mock_identity.return_value.create_role.return_value = "role3"167        self.assertRaises(exceptions.RallyAssertionError,168                          scenario.run, create_role_kwargs=create_kwargs,169                          list_role_kwargs=list_kwargs)170        self.mock_identity.return_value.create_role.assert_called_with(171            **create_kwargs)172        self.mock_identity.return_value.list_roles.assert_called_with(173            **list_kwargs)174    @ddt.data(None, "keystone", "fooservice")175    def test_get_entities(self, service_name):176        identity_service = self.mock_identity.return_value177        fake_project = identity_service.create_project.return_value178        fake_user = identity_service.create_user.return_value179        fake_role = identity_service.create_role.return_value180        fake_service = identity_service.create_service.return_value181        scenario = basic.GetEntities(self.context)182        scenario.run(service_name)183        identity_service.create_project.assert_called_once_with()184        identity_service.create_user.assert_called_once_with(185            project_id=fake_project.id)186        identity_service.create_role.assert_called_once_with()187        identity_service.get_project.assert_called_once_with(fake_project.id)188        identity_service.get_user.assert_called_once_with(fake_user.id)189        identity_service.get_role.assert_called_once_with(fake_role.id)190        if service_name is None:191            identity_service.create_service.assert_called_once_with()192            self.assertFalse(identity_service.get_service_by_name.called)193            identity_service.get_service.assert_called_once_with(194                fake_service.id)195        else:196            identity_service.get_service_by_name.assert_called_once_with(197                service_name)198            self.assertFalse(identity_service.create_service.called)199            identity_service.get_service.assert_called_once_with(200                identity_service.get_service_by_name.return_value.id)201    def test_create_and_delete_service(self):202        identity_service = self.mock_identity.return_value203        scenario = basic.CreateAndDeleteService(self.context)204        service_type = "test_service_type"205        description = "test_description"206        fake_service = identity_service.create_service.return_value207        scenario.run(service_type=service_type, description=description)208        identity_service.create_service.assert_called_once_with(209            service_type=service_type, description=description)210        identity_service.delete_service.assert_called_once_with(211            fake_service.id)212    def test_create_update_and_delete_tenant(self):213        identity_service = self.mock_identity.return_value214        scenario = basic.CreateUpdateAndDeleteTenant(self.context)215        gen_name = mock.MagicMock()216        basic.CreateUpdateAndDeleteTenant.generate_random_name = gen_name217        fake_project = identity_service.create_project.return_value218        scenario.run()219        identity_service.create_project.assert_called_once_with()220        identity_service.update_project.assert_called_once_with(221            fake_project.id, description=gen_name.return_value,222            name=gen_name.return_value)223        identity_service.delete_project(fake_project.id)224    def test_create_user_update_password(self):225        identity_service = self.mock_identity.return_value226        scenario = basic.CreateUserUpdatePassword(self.context)227        fake_password = "pswd"228        fake_user = identity_service.create_user.return_value229        scenario.generate_random_name = mock.MagicMock(230            return_value=fake_password)231        scenario.run()232        scenario.generate_random_name.assert_called_once_with()233        identity_service.create_user.assert_called_once_with()234        identity_service.update_user.assert_called_once_with(235            fake_user.id, password=fake_password)236    def test_create_and_update_user(self):237        identity_service = self.mock_identity.return_value238        scenario = basic.CreateAndUpdateUser(self.context)239        scenario.admin_clients("keystone").users.get = mock.MagicMock()240        fake_user = identity_service.create_user.return_value241        create_args = {"fakearg1": "f"}242        update_args = {"fakearg1": "fakearg"}243        setattr(self.admin_clients("keystone").users.get.return_value,244                "fakearg1", "fakearg")245        scenario.run(create_user_kwargs=create_args,246                     update_user_kwargs=update_args)247        identity_service.create_user.assert_called_once_with(**create_args)248        identity_service.update_user.assert_called_once_with(249            fake_user.id, **update_args)250    def test_create_and_list_services(self):251        identity_service = self.mock_identity.return_value252        scenario = basic.CreateAndListServices(self.context)253        service_type = "test_service_type"254        description = "test_description"255        scenario.run(service_type=service_type, description=description)256        identity_service.create_service.assert_called_once_with(257            service_type=service_type, description=description)258        identity_service.list_services.assert_called_once_with()259    def test_create_and_list_ec2credentials(self):260        identity_service = self.mock_identity.return_value261        scenario = basic.CreateAndListEc2Credentials(self.context)262        scenario.run()263        identity_service.create_ec2credentials.assert_called_once_with(264            self.context["user"]["id"],265            project_id=self.context["tenant"]["id"])266        identity_service.list_ec2credentials.assert_called_with(267            self.context["user"]["id"])268    def test_create_and_delete_ec2credential(self):269        identity_service = self.mock_identity.return_value270        fake_creds = identity_service.create_ec2credentials.return_value271        scenario = basic.CreateAndDeleteEc2Credential(self.context)272        scenario.run()273        identity_service.create_ec2credentials.assert_called_once_with(274            self.context["user"]["id"],275            project_id=self.context["tenant"]["id"])276        identity_service.delete_ec2credential.assert_called_once_with(277            self.context["user"]["id"], access=fake_creds.access)278    def test_add_and_remove_user_role(self):279        context = self.context280        tenant_id = context["tenant"]["id"]281        user_id = context["user"]["id"]282        fake_role = mock.MagicMock()283        self.mock_identity.return_value.create_role.return_value = fake_role284        scenario = basic.AddAndRemoveUserRole(context)285        scenario.run()286        self.mock_identity.return_value.create_role.assert_called_once_with()287        self.mock_identity.return_value.add_role.assert_called_once_with(288            role_id=fake_role.id, user_id=user_id, project_id=tenant_id)289        self.mock_identity.return_value.revoke_role.assert_called_once_with(...

test_init.py

Source:test_init.py

...31    with patch_project, patch_at as mock_t, patch_tn as mock_tn, patch_l as mock_l:32        main(args_in=["init"])33    mock_tn.assert_called_once_with()34    mock_l.assert_called_once_with()35    mock_t.assert_called_once()36    mock_project.load_settings.assert_called_once_with()37    mock_project.init.assert_called_once_with(38        type_name,39        language,40        {41            "version": False,42            "subparser_name": None,43            "verbose": 0,44            "force": False,45            "type_name": None,46            "artifact_type": None,47        },48    )49    mock_project.generate.assert_called_once_with()50def test_init_module_method_interactive():51    type_name = object()52    language = object()53    mock_project, patch_project = get_mock_project()54    patch_tn = patch(55        "rpdk.core.module.init_module.input_typename", return_value=type_name56    )57    patch_l = patch(58        "rpdk.core.resource.init_resource.input_language", return_value=language59    )60    patch_at = patch("rpdk.core.init.init_artifact_type", return_value="MODULE")61    with patch_project, patch_tn as mock_tn, patch_l as mock_l, patch_at as mock_t:62        main(args_in=["init"])63    mock_tn.assert_called_once_with()64    mock_l.assert_not_called()65    mock_t.assert_called_once()66    mock_project.load_settings.assert_called_once_with()67    mock_project.init_module.assert_called_once_with(type_name)68    mock_project.generate.assert_not_called()69def test_init_resource_method_noninteractive():70    add_dummy_language_plugin()71    artifact_type = "RESOURCE"72    args = get_args("dummy", "Test::Test::Test", artifact_type)73    mock_project, patch_project = get_mock_project()74    patch_get_parser = patch(75        "rpdk.core.init.get_parsers", return_value={"dummy": dummy_parser}76    )77    # flake8: noqa: B95078    # pylint: disable=line-too-long79    with patch_project, patch_get_parser as mock_parser:80        main(81            args_in=[82                "init",83                "--type-name",84                args.type_name,85                "--artifact-type",86                args.artifact_type,87                args.language,88                "--dummy",89            ]90        )91    mock_parser.assert_called_once()92    mock_project.load_settings.assert_called_once_with()93    mock_project.init.assert_called_once_with(94        args.type_name,95        args.language,96        {97            "version": False,98            "subparser_name": args.language,99            "verbose": 0,100            "force": False,101            "type_name": args.type_name,102            "language": args.language,103            "dummy": True,104            "artifact_type": artifact_type,105        },106    )107    mock_project.generate.assert_called_once_with()108def test_init_resource_method_noninteractive_invalid_type_name():109    add_dummy_language_plugin()110    type_name = object()111    artifact_type = "RESOURCE"112    args = get_args("dummy", "invalid_type_name", "RESOURCE")113    mock_project, patch_project = get_mock_project()114    patch_tn = patch(115        "rpdk.core.resource.init_resource.input_typename", return_value=type_name116    )117    patch_t = patch("rpdk.core.init.init_artifact_type", return_value=artifact_type)118    patch_get_parser = patch(119        "rpdk.core.init.get_parsers", return_value={"dummy": dummy_parser}120    )121    # flake8: noqa: B950122    # pylint: disable=line-too-long123    with patch_project, patch_t, patch_tn as mock_tn, patch_get_parser as mock_parser:124        main(125            args_in=[126                "init",127                "-t",128                args.type_name,129                "-a",130                args.artifact_type,131                args.language,132                "--dummy",133            ]134        )135    mock_tn.assert_called_once_with()136    mock_parser.assert_called_once()137    mock_project.load_settings.assert_called_once_with()138    mock_project.init.assert_called_once_with(139        type_name,140        args.language,141        {142            "version": False,143            "subparser_name": args.language,144            "verbose": 0,145            "force": False,146            "type_name": args.type_name,147            "artifact_type": artifact_type,148            "dummy": True,149            "language": args.language,150        },151    )152    mock_project.generate.assert_called_once_with()153def test_input_with_validation_valid_first_try(capsys):154    sentinel1 = object()155    sentinel2 = object()156    validator = Mock(return_value=sentinel1)157    with patch("rpdk.core.init.input", return_value=sentinel2) as mock_input:158        ret = input_with_validation(PROMPT, validator)159    mock_input.assert_called_once_with()160    validator.assert_called_once_with(sentinel2)161    assert ret is sentinel1162    out, err = capsys.readouterr()163    assert not err164    assert PROMPT in out165def test_input_with_validation_valid_second_try(capsys):166    def mock_validator(value):167        if value == ERROR:168            raise WizardValidationError(ERROR)169        return value170    sentinel = object()171    with patch("rpdk.core.init.input", side_effect=(ERROR, sentinel)) as mock_input:172        ret = input_with_validation(PROMPT, mock_validator)173    assert mock_input.call_count == 2174    assert ret is sentinel175    out, err = capsys.readouterr()176    assert not err177    assert ERROR in out178def test_validate_type_name_valid():179    assert validate_type_name("AWS::Color::Red") == "AWS::Color::Red"180def test_validate_type_name_invalid():181    with pytest.raises(WizardValidationError):182        validate_type_name("AWS-Color-Red")183@pytest.mark.parametrize("value", ("y", "yes", "Y", "YES", "yEs", "YeS"))184def test_validate_yes_yes(value):185    assert validate_yes(value)186@pytest.mark.parametrize("value", ("n", "N", "no", "NO", "yesn't"))187def test_validate_yes_no(value):188    assert not validate_yes(value)189def test_validate_plugin_choice_not_an_int():190    validator = ValidatePluginChoice(["test"])191    with pytest.raises(WizardValidationError) as excinfo:192        validator("a")193    assert "integer" in str(excinfo.value)194def test_validate_plugin_choice_less_than_zero():195    validator = ValidatePluginChoice(["test"])196    with pytest.raises(WizardValidationError) as excinfo:197        validator("-1")198    assert "select" in str(excinfo.value)199def test_validate_plugin_choice_greater_than_choice():200    choices = range(3)201    validator = ValidatePluginChoice(choices)202    with pytest.raises(WizardValidationError) as excinfo:203        validator(str(len(choices) + 1))  # index is 1 based for input204    assert "select" in str(excinfo.value)205def test_validate_plugin_choice_valid():206    choices = ["1", PROMPT, "2"]207    validator = ValidatePluginChoice(choices)208    assert validator("2") == PROMPT209def test_init_module_method_noninteractive():210    add_dummy_language_plugin()211    artifact_type = "MODULE"212    args = get_args("dummy", "Test::Test::Test::MODULE", artifact_type)213    mock_project, patch_project = get_mock_project()214    patch_get_parser = patch(215        "rpdk.core.init.get_parsers", return_value={"dummy": dummy_parser}216    )217    # flake8: noqa: B950218    # pylint: disable=line-too-long219    with patch_project, patch_get_parser as mock_parser:220        main(221            args_in=[222                "init",223                "--type-name",224                args.type_name,225                "--artifact-type",226                args.artifact_type,227                args.language,228                "--dummy",229            ]230        )231    mock_parser.assert_called_once()232    mock_project.load_settings.assert_called_once_with()233    mock_project.init_module.assert_called_once_with(args.type_name)234    mock_project.generate.assert_not_called()235def test_check_for_existing_project_good_path():236    project = Mock(spec=Project)237    project.load_settings.side_effect = FileNotFoundError238    type(project).overwrite_enabled = mock_overwrite = PropertyMock()239    check_for_existing_project(project)240    project.load_settings.assert_called_once_with()241    mock_overwrite.assert_not_called()242def test_check_for_existing_project_bad_path_overwrite():243    project = Mock(spec=Project)244    project.overwrite_enabled = True245    project.settings_path = ""  # not sure why this doesn't get specced?246    project.type_name = ""247    with patch("rpdk.core.init.input_with_validation", autospec=True) as mock_input:248        check_for_existing_project(project)249    mock_input.assert_not_called()250def test_check_for_existing_project_bad_path_ask_yes():251    project = Mock(spec=Project)252    project.overwrite_enabled = False253    project.settings_path = ""254    project.type_name = ""255    patch_input = patch(256        "rpdk.core.init.input_with_validation", autospec=True, return_value="m"257    )258    with patch_input as mock_input:259        check_for_existing_project(project)260    mock_input.assert_called_once_with(ANY, validate_yes, ANY)261    assert project.overwrite_enabled262def test_check_for_existing_project_bad_path_ask_no():263    project = Mock(spec=Project)264    project.overwrite_enabled = False265    project.settings_path = ""266    project.type_name = ""267    patch_input = patch(268        "rpdk.core.init.input_with_validation", autospec=True, return_value=False269    )270    with patch_input as mock_input:271        with pytest.raises(WizardAbortError):272            check_for_existing_project(project)273    mock_input.assert_called_once_with(ANY, validate_yes, ANY)274    assert not project.overwrite_enabled275def test_ignore_abort_ok():276    sentinel = object()277    function = Mock()278    wrapped = ignore_abort(function)279    wrapped(sentinel)280    function.assert_called_once_with(sentinel)281def test_ignore_abort_keyboard_interrupt():282    sentinel = object()283    function = Mock(side_effect=KeyboardInterrupt)284    wrapped = ignore_abort(function)285    with pytest.raises(SystemExit):286        wrapped(sentinel)287    function.assert_called_once_with(sentinel)288def test_ignore_abort_abort():289    sentinel = object()290    function = Mock(side_effect=WizardAbortError)291    wrapped = ignore_abort(function)292    with pytest.raises(SystemExit):293        wrapped(sentinel)294    function.assert_called_once_with(sentinel)295def test_input_typename_resource():296    type_name = "AWS::Color::Red"297    patch_input = patch("rpdk.core.utils.init_utils.input", return_value=type_name)298    with patch_input as mock_input:299        assert input_typename_resource() == type_name300    mock_input.assert_called_once()301def test_input_typename_module():302    type_name = "AWS::Color::Red::MODULE"303    patch_input = patch("rpdk.core.utils.init_utils.input", return_value=type_name)304    with patch_input as mock_input:305        assert input_typename_module() == type_name306    mock_input.assert_called_once()307def test_input_language_no_plugins():308    validator = ValidatePluginChoice([])309    with patch("rpdk.core.init.validate_plugin_choice", validator):310        with pytest.raises(WizardAbortError):311            input_language()312def test_input_language_one_plugin():313    validator = ValidatePluginChoice([PROMPT])314    with patch("rpdk.core.init.validate_plugin_choice", validator):315        assert input_language() == PROMPT316def test_input_language_several_plugins():317    validator = ValidatePluginChoice(["1", PROMPT, "2"])318    patch_validator = patch("rpdk.core.init.validate_plugin_choice", validator)319    patch_input = patch("rpdk.core.init.input", return_value="2")320    with patch_validator, patch_input as mock_input:321        assert input_language() == PROMPT...

fixtures.py

Source:fixtures.py

1# testing/fixtures.py2# Copyright (C) 2005-2019 the SQLAlchemy authors and contributors3# <see AUTHORS file>4#5# This module is part of SQLAlchemy and is released under6# the MIT License: http://www.opensource.org/licenses/mit-license.php7import sys8import sqlalchemy as sa9from . import assertions10from . import config11from . import schema12from .engines import drop_all_tables13from .entities import BasicEntity14from .entities import ComparableEntity15from .util import adict16from .. import event17from .. import util18from ..ext.declarative import declarative_base19from ..ext.declarative import DeclarativeMeta20# whether or not we use unittest changes things dramatically,21# as far as how py.test collection works.22class TestBase(object):23    # A sequence of database names to always run, regardless of the24    # constraints below.25    __whitelist__ = ()26    # A sequence of requirement names matching testing.requires decorators27    __requires__ = ()28    # A sequence of dialect names to exclude from the test class.29    __unsupported_on__ = ()30    # If present, test class is only runnable for the *single* specified31    # dialect.  If you need multiple, use __unsupported_on__ and invert.32    __only_on__ = None33    # A sequence of no-arg callables. If any are True, the entire testcase is34    # skipped.35    __skip_if__ = None36    def assert_(self, val, msg=None):37        assert val, msg38    # apparently a handful of tests are doing this....OK39    def setup(self):40        if hasattr(self, "setUp"):41            self.setUp()42    def teardown(self):43        if hasattr(self, "tearDown"):44            self.tearDown()45class TablesTest(TestBase):46    # 'once', None47    run_setup_bind = "once"48    # 'once', 'each', None49    run_define_tables = "once"50    # 'once', 'each', None51    run_create_tables = "once"52    # 'once', 'each', None53    run_inserts = "each"54    # 'each', None55    run_deletes = "each"56    # 'once', None57    run_dispose_bind = None58    bind = None59    metadata = None60    tables = None61    other = None62    @classmethod63    def setup_class(cls):64        cls._init_class()65        cls._setup_once_tables()66        cls._setup_once_inserts()67    @classmethod68    def _init_class(cls):69        if cls.run_define_tables == "each":70            if cls.run_create_tables == "once":71                cls.run_create_tables = "each"72            assert cls.run_inserts in ("each", None)73        cls.other = adict()74        cls.tables = adict()75        cls.bind = cls.setup_bind()76        cls.metadata = sa.MetaData()77        cls.metadata.bind = cls.bind78    @classmethod79    def _setup_once_inserts(cls):80        if cls.run_inserts == "once":81            cls._load_fixtures()82            cls.insert_data()83    @classmethod84    def _setup_once_tables(cls):85        if cls.run_define_tables == "once":86            cls.define_tables(cls.metadata)87            if cls.run_create_tables == "once":88                cls.metadata.create_all(cls.bind)89            cls.tables.update(cls.metadata.tables)90    def _setup_each_tables(self):91        if self.run_define_tables == "each":92            self.tables.clear()93            if self.run_create_tables == "each":94                drop_all_tables(self.metadata, self.bind)95            self.metadata.clear()96            self.define_tables(self.metadata)97            if self.run_create_tables == "each":98                self.metadata.create_all(self.bind)99            self.tables.update(self.metadata.tables)100        elif self.run_create_tables == "each":101            drop_all_tables(self.metadata, self.bind)102            self.metadata.create_all(self.bind)103    def _setup_each_inserts(self):104        if self.run_inserts == "each":105            self._load_fixtures()106            self.insert_data()107    def _teardown_each_tables(self):108        # no need to run deletes if tables are recreated on setup109        if self.run_define_tables != "each" and self.run_deletes == "each":110            with self.bind.connect() as conn:111                for table in reversed(self.metadata.sorted_tables):112                    try:113                        conn.execute(table.delete())114                    except sa.exc.DBAPIError as ex:115                        util.print_(116                            ("Error emptying table %s: %r" % (table, ex)),117                            file=sys.stderr,118                        )119    def setup(self):120        self._setup_each_tables()121        self._setup_each_inserts()122    def teardown(self):123        self._teardown_each_tables()124    @classmethod125    def _teardown_once_metadata_bind(cls):126        if cls.run_create_tables:127            drop_all_tables(cls.metadata, cls.bind)128        if cls.run_dispose_bind == "once":129            cls.dispose_bind(cls.bind)130        cls.metadata.bind = None131        if cls.run_setup_bind is not None:132            cls.bind = None133    @classmethod134    def teardown_class(cls):135        cls._teardown_once_metadata_bind()136    @classmethod137    def setup_bind(cls):138        return config.db139    @classmethod140    def dispose_bind(cls, bind):141        if hasattr(bind, "dispose"):142            bind.dispose()143        elif hasattr(bind, "close"):144            bind.close()145    @classmethod146    def define_tables(cls, metadata):147        pass148    @classmethod149    def fixtures(cls):150        return {}151    @classmethod152    def insert_data(cls):153        pass154    def sql_count_(self, count, fn):155        self.assert_sql_count(self.bind, fn, count)156    def sql_eq_(self, callable_, statements):157        self.assert_sql(self.bind, callable_, statements)158    @classmethod159    def _load_fixtures(cls):160        """Insert rows as represented by the fixtures() method."""161        headers, rows = {}, {}162        for table, data in cls.fixtures().items():163            if len(data) < 2:164                continue165            if isinstance(table, util.string_types):166                table = cls.tables[table]167            headers[table] = data[0]168            rows[table] = data[1:]169        for table in cls.metadata.sorted_tables:170            if table not in headers:171                continue172            cls.bind.execute(173                table.insert(),174                [175                    dict(zip(headers[table], column_values))176                    for column_values in rows[table]177                ],178            )179class RemovesEvents(object):180    @util.memoized_property181    def _event_fns(self):182        return set()183    def event_listen(self, target, name, fn, **kw):184        self._event_fns.add((target, name, fn))185        event.listen(target, name, fn, **kw)186    def teardown(self):187        for key in self._event_fns:188            event.remove(*key)189        super_ = super(RemovesEvents, self)190        if hasattr(super_, "teardown"):191            super_.teardown()192class _ORMTest(object):193    @classmethod194    def teardown_class(cls):195        sa.orm.session.Session.close_all()196        sa.orm.clear_mappers()197class ORMTest(_ORMTest, TestBase):198    pass199class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults):200    # 'once', 'each', None201    run_setup_classes = "once"202    # 'once', 'each', None203    run_setup_mappers = "each"204    classes = None205    @classmethod206    def setup_class(cls):207        cls._init_class()208        if cls.classes is None:209            cls.classes = adict()210        cls._setup_once_tables()211        cls._setup_once_classes()212        cls._setup_once_mappers()213        cls._setup_once_inserts()214    @classmethod215    def teardown_class(cls):216        cls._teardown_once_class()217        cls._teardown_once_metadata_bind()218    def setup(self):219        self._setup_each_tables()220        self._setup_each_classes()221        self._setup_each_mappers()222        self._setup_each_inserts()223    def teardown(self):224        sa.orm.session.Session.close_all()225        self._teardown_each_mappers()226        self._teardown_each_classes()227        self._teardown_each_tables()228    @classmethod229    def _teardown_once_class(cls):230        cls.classes.clear()231        _ORMTest.teardown_class()232    @classmethod233    def _setup_once_classes(cls):234        if cls.run_setup_classes == "once":235            cls._with_register_classes(cls.setup_classes)236    @classmethod237    def _setup_once_mappers(cls):238        if cls.run_setup_mappers == "once":239            cls._with_register_classes(cls.setup_mappers)240    def _setup_each_mappers(self):241        if self.run_setup_mappers == "each":242            self._with_register_classes(self.setup_mappers)243    def _setup_each_classes(self):244        if self.run_setup_classes == "each":245            self._with_register_classes(self.setup_classes)246    @classmethod247    def _with_register_classes(cls, fn):248        """Run a setup method, framing the operation with a Base class249        that will catch new subclasses to be established within250        the "classes" registry.251        """252        cls_registry = cls.classes253        class FindFixture(type):254            def __init__(cls, classname, bases, dict_):255                cls_registry[classname] = cls256                return type.__init__(cls, classname, bases, dict_)257        class _Base(util.with_metaclass(FindFixture, object)):258            pass259        class Basic(BasicEntity, _Base):260            pass261        class Comparable(ComparableEntity, _Base):262            pass263        cls.Basic = Basic264        cls.Comparable = Comparable265        fn()266    def _teardown_each_mappers(self):267        # some tests create mappers in the test bodies268        # and will define setup_mappers as None -269        # clear mappers in any case270        if self.run_setup_mappers != "once":271            sa.orm.clear_mappers()272    def _teardown_each_classes(self):273        if self.run_setup_classes != "once":274            self.classes.clear()275    @classmethod276    def setup_classes(cls):277        pass278    @classmethod279    def setup_mappers(cls):280        pass281class DeclarativeMappedTest(MappedTest):282    run_setup_classes = "once"283    run_setup_mappers = "once"284    @classmethod285    def _setup_once_tables(cls):286        pass287    @classmethod288    def _with_register_classes(cls, fn):289        cls_registry = cls.classes290        class FindFixtureDeclarative(DeclarativeMeta):291            def __init__(cls, classname, bases, dict_):292                cls_registry[classname] = cls293                return DeclarativeMeta.__init__(cls, classname, bases, dict_)294        class DeclarativeBasic(object):295            __table_cls__ = schema.Table296        _DeclBase = declarative_base(297            metadata=cls.metadata,298            metaclass=FindFixtureDeclarative,299            cls=DeclarativeBasic,300        )301        cls.DeclarativeBasic = _DeclBase302        fn()303        if cls.metadata.tables and cls.run_create_tables:...

test_artifact_bundles.py

Source:test_artifact_bundles.py

...30        self.host = '127.0.0.1'31        self.connection = connection(self.host)32        self._client = ArtifactBundles(self.connection)33    @mock.patch.object(ResourceClient, 'get_all')34    def test_get_all_called_once(self, mock_get_all):35        filter = 'name=TestName'36        sort = 'name: ascending'37        self._client.get_all(2, 500, filter, sort)38        mock_get_all.assert_called_once_with(2, 500, filter=filter, sort=sort)39    @mock.patch.object(ResourceClient, 'get_all')40    def test_get_all_called_once_with_default_values(self, mock_get_all):41        self._client.get_all()42        mock_get_all.assert_called_once_with(0, -1, filter='', sort='')43    @mock.patch.object(ResourceClient, 'get')44    def test_get_called_once(self, mock_get):45        self._client.get('78836581-2b6f-4e26-9969-5667fb5837b4')46        mock_get.assert_called_once_with(47            '78836581-2b6f-4e26-9969-5667fb5837b4')48    @mock.patch.object(ResourceClient, 'get')49    def test_get_with_uri_called_once(self, mock_get):50        uri = '/rest/artifact-bundles/78836581-2b6f-4e26-9969-5667fb5837b4'51        self._client.get(uri)52        mock_get.assert_called_once_with(uri)53    @mock.patch.object(ResourceClient, 'get_by')54    def test_get_by_called_once(self, mock_get_by):55        self._client.get_by('name', 'ArtifactBundle')56        mock_get_by.assert_called_once_with(57            'name', 'ArtifactBundle')58    @mock.patch.object(ResourceClient, 'get_by_name')59    def test_get_by_name_called_once(self, mock_get_by):60        self._client.get_by_name('ArtifactBundle')61        mock_get_by.assert_called_once_with('ArtifactBundle')62    @mock.patch.object(ResourceClient, 'create')63    def test_create_called_once(self, mock_create):64        information = {65            "name": "ArtifactBundle"66        }67        mock_create.return_value = {}68        self._client.create(information)69        mock_create.assert_called_once_with(information.copy(), timeout=-1)70    @mock.patch.object(ResourceClient, 'update')71    def test_update_called_once(self, mock_update):72        information = {73            "name": "ArtifactBundleUpdate",74            "uri": "/rest/artifact-bundles/78836581-2b6f-4e26-9969-5667fb5837b4"75        }76        mock_update.return_value = {}77        self._client.update(information)78        default_values = {u'300': {u'type': u'ArtifactsBundle'}}79        mock_update.assert_called_once_with(information.copy(), default_values=default_values, timeout=-1)80    @mock.patch.object(ResourceClient, 'delete')81    def test_remove_called_once(self, mock_delete):82        id = '78836581-2b6f-4e26-9969-5667fb5837b4'83        self._client.delete(id)84        mock_delete.assert_called_once_with(id, timeout=-1)85    @mock.patch.object(ResourceClient, 'get_collection')86    def test_get_all_backups_called_once(self, mock_get_collection):87        uri = '/rest/artifact-bundles/backups'88        self._client.get_all_backups()89        mock_get_collection.assert_called_once_with(id_or_uri=uri)90    @mock.patch.object(ResourceClient, 'get')91    def test_get_backups_by_id_called_once(self, mock_get):92        uri = '/rest/artifact-bundles/backups'93        id = '78836581-2b6f-4e26-9969-5667fb5837b4'94        self._client.get_backup(id)95        mock_get.assert_called_once_with(id_or_uri=uri + '/' + id)96    @mock.patch.object(ResourceClient, 'download')97    def test_download_called_once_by_id(self, mock_download):98        destination = '~/image.zip'99        self._client.download_artifact_bundle('0ABDE00534F', destination)100        mock_download.assert_called_once_with('/rest/artifact-bundles/download/0ABDE00534F', destination)101    @mock.patch.object(ResourceClient, 'download')102    def test_download_called_once_by_uri(self, mock_download):103        uri = '/rest/artifact-bundles/0ABDE00534F'104        destination = '~/image.zip'105        self._client.download_artifact_bundle(uri, destination)106        mock_download.assert_called_once_with('/rest/artifact-bundles/download/0ABDE00534F', destination)107    @mock.patch.object(ResourceClient, 'download')108    def test_download_archive_artifact_bundle_by_id_called_once(self, mock_download):109        id = '78836581-2b6f-4e26-9969-5667fb5837b4'110        destination = '~/image.zip'111        self._client.download_archive_artifact_bundle(id, destination)112        expected_uri = '/rest/artifact-bundles/backups/archive/78836581-2b6f-4e26-9969-5667fb5837b4'113        mock_download.assert_called_once_with(expected_uri, destination)114    @mock.patch.object(ResourceClient, 'download')115    def test_download_archive_artifact_bundle_by_uri_called_once(self, mock_download):116        uri = '/rest/artifact-bundles/78836581-2b6f-4e26-9969-5667fb5837b4'117        destination = '~/image.zip'118        self._client.download_archive_artifact_bundle(uri, destination)119        expected_uri = '/rest/artifact-bundles/backups/archive/78836581-2b6f-4e26-9969-5667fb5837b4'120        mock_download.assert_called_once_with(expected_uri, destination)121    @mock.patch.object(ResourceClient, 'create')122    def test_create_backup_called_once(self, mock_create):123        information = {124            "deploymentGroupURI": "/rest/deployment-groups/00c1344d-e4dd-43c3-a733-1664e159a36f"125        }126        mock_create.return_value = {}127        self._client.create_backup(information)128        uri = '/rest/artifact-bundles/backups'129        mock_create.assert_called_once_with(information.copy(), uri=uri, timeout=-1)130    @mock.patch.object(ResourceClient, 'upload')131    def test_upload_artifact_bundle_called_once(self, mock_upload):132        filepath = "~/HPE-ImageStreamer-Developer-2016-09-12.zip"133        self._client.upload_bundle_from_file(filepath)134        mock_upload.assert_called_once_with(filepath)135    @mock.patch.object(ResourceClient, 'upload')136    def test_upload_backup_artifact_bundle_called_once(self, mock_upload):137        filepath = "~/HPE-ImageStreamer-Developer-2016-09-12.zip"138        deployment_groups = "/rest/deployment-groups/00c1344d-e4dd-43c3-a733-1664e159a36f"139        self._client.upload_backup_bundle_from_file(filepath, deployment_groups)140        expected_uri = '/rest/artifact-bundles/backups/archive?deploymentGrpUri=' + deployment_groups141        mock_upload.assert_called_once_with(filepath, expected_uri)142    @mock.patch.object(ResourceClient, 'upload')143    def test_upload_backup_artifact_bundle_called_once_with_id(self, mock_upload):144        filepath = "~/HPE-ImageStreamer-Developer-2016-09-12.zip"145        deployment_groups = "00c1344d-e4dd-43c3-a733-1664e159a36f"146        self._client.upload_backup_bundle_from_file(filepath, deployment_groups)147        expected_uri = '/rest/artifact-bundles/backups/archive?' \148                       'deploymentGrpUri=/rest/deployment-groups/00c1344d-e4dd-43c3-a733-1664e159a36f'149        mock_upload.assert_called_once_with(filepath, expected_uri)150    @mock.patch.object(ResourceClient, 'update')151    def test_extract_called_once(self, mock_update):152        mock_update.return_value = {}153        uri = {'uri': '/rest/artifact-bundles/78836581-2b6f-4e26-9969-5667fb5837b4'}154        custom_headers = {'Content-Type': 'text/plain'}155        self._client.extract_bundle(uri)156        mock_update.assert_called_once_with(uri, custom_headers=custom_headers, timeout=-1)157    @mock.patch.object(ResourceClient, 'update')158    def test_extract_backup_bundle_called_once(self, mock_update):159        mock_update.return_value = {}160        data = {161            'deploymentGroupURI': '/rest/deployment-groups/00c1344d-e4dd-43c3-a733-1664e159a36f'162        }163        uri = '/rest/artifact-bundles/backups/archive'164        self._client.extract_backup_bundle(data)165        mock_update.assert_called_once_with(data, uri=uri, timeout=-1)166    @mock.patch.object(ResourceClient, 'update')167    def test_stop_creation_called_once(self, mock_update):168        mock_update.return_value = {}169        artifact_uri = "/rest/artifact-bundles/04939e89-bcb0-49fc-814f-1a6bc0a2f63c"170        task_uri = "/rest/tasks/A15F9270-46FC-48DF-94A9-D11EDB52877E"171        self._client.stop_artifact_creation(artifact_uri, task_uri)172        uri = artifact_uri + '/stopArtifactCreate'173        task_uri = {174            'taskUri': task_uri175        }...

test_power_devices.py

Source:test_power_devices.py

...46    def test_get_utilization_with_defaults(self, mock_get):47        self._power_devices.get_utilization('35323930-4936-4450-5531-303153474820')48        mock_get.assert_called_once_with('35323930-4936-4450-5531-303153474820', None, None, False, None)49    @mock.patch.object(ResourceClient, 'get_all')50    def test_get_all_called_once(self, mock_get_all):51        filter = 'name=TestName'52        sort = 'name:ascending'53        self._power_devices.get_all(2, 500, filter=filter, sort=sort)54        mock_get_all.assert_called_once_with(2, 500, filter=filter, sort=sort, query='')55    @mock.patch.object(ResourceClient, 'get_all')56    def test_get_all_called_once_with_default(self, mock_get_all):57        self._power_devices.get_all()58        mock_get_all.assert_called_once_with(0, -1, filter='', sort='', query='')59    @mock.patch.object(ResourceClient, 'get')60    def test_get_by_id_called_once(self, mock_get):61        rack_id = "f0a0a113-ec97-41b4-83ce-d7c92b900e7c"62        self._power_devices.get(rack_id)63        mock_get.assert_called_once_with(rack_id)64    @mock.patch.object(ResourceClient, 'get')65    def test_get_by_uri_called_once(self, mock_get):66        rack_uri = "/rest/racks/f0a0a113-ec97-41b4-83ce-d7c92b900e7c"67        self._power_devices.get(rack_uri)68        mock_get.assert_called_once_with(rack_uri)69    @mock.patch.object(ResourceClient, 'get_by')70    def test_get_by_called_once(self, mock_get_by):71        self._power_devices.get_by("name", "test name")72        mock_get_by.assert_called_once_with("name", "test name")73    @mock.patch.object(ResourceClient, 'create')74    def test_add_called_once_with_defaults(self, mock_create):75        rack = {76            "name": "MyRack"77        }78        self._power_devices.add(rack)79        mock_create.assert_called_once_with(rack, timeout=-1)80    @mock.patch.object(ResourceClient, 'create')81    def test_create_called_once(self, mock_create):82        rack = {83            "name": "MyRack"84        }85        self._power_devices.add(rack, 70)86        mock_create.assert_called_once_with(rack, timeout=70)87    @mock.patch.object(ResourceClient, 'update')88    def test_update_called_once_with_defaults(self, update):89        rack = {90            "id": "4b4b87e2-eea8-4c90-8eca-b92eaaeecfff",91            "uuid": "4b4b87e2-eea8-4c90-8eca-b92eaaeecfff",92            "depth": 20,93            "height": 30,94            "width": 2095        }96        self._power_devices.update(rack)97        update.assert_called_once_with(rack.copy(), timeout=-1)98    @mock.patch.object(ResourceClient, 'update')99    def test_update_called_once(self, mock_update):100        rack = {101            "id": "4b4b87e2-eea8-4c90-8eca-b92eaaeecfff",102            "uuid": "4b4b87e2-eea8-4c90-8eca-b92eaaeecfff",103            "depth": 20,104            "height": 30,105            "width": 20106        }107        self._power_devices.update(rack, 70)108        mock_update.assert_called_once_with(rack.copy(), timeout=70)109    @mock.patch.object(ResourceClient, 'delete')110    def test_remove_called_once(self, mock_delete):111        id = 'ad28cf21-8b15-4f92-bdcf-51cb2042db32'112        self._power_devices.remove(id, force=True, timeout=50)113        mock_delete.assert_called_once_with(id, force=True, timeout=50)114    @mock.patch.object(ResourceClient, 'delete')115    def test_remove_called_once_with_defaults(self, mock_delete):116        id = 'ad28cf21-8b15-4f92-bdcf-51cb2042db32'117        self._power_devices.remove(id)118        mock_delete.assert_called_once_with(id, force=False, timeout=-1)119    @mock.patch.object(ResourceClient, 'create')120    def test_add_ipdu_called_once_with_defaults(self, mock_create):121        self._power_devices.add_ipdu({'name': 'name'})122        mock_create.assert_called_once_with({'name': 'name'}, timeout=-1, uri='/rest/power-devices/discover')123    @mock.patch.object(ResourceClient, 'get')124    def test_get_power_state_called_once_with_defaults(self, mock_get):...

test_send_file.py

Source:test_send_file.py

...10    file.return_value.__enter__.return_value.read.side_effect = lambda _: next(f)11    # ===== invoke =====12    send_file(filename="mytext.txt", testing=True)13    # ===== ensurance =====14    sock.assert_called_once()15    sock.return_value.bind.assert_called_once()16    sock.return_value.listen.assert_called_once()17    sock.return_value.accept.assert_called_once()18    conn.recv.assert_called_once()19    file.return_value.__enter__.assert_called_once()20    file.return_value.__enter__.return_value.read.assert_called()21    conn.send.assert_called_once()22    conn.close.assert_called_once()23    sock.return_value.shutdown.assert_called_once()...

Using AI Code Generation

1const chromy = require('chromy')2chromy.chain()3  .once('load', () => {4    console.log('page loaded')5  })6  .evaluate(() => {7  })8  .result(title => {9    console.log(title)10  })11  .end()12  .then(() => {13    console.log('done')14  })15  .catch(e => {16    console.log(e)17  })18const chromy = require('chromy')19chromy.chain()20  .on('load', () => {21    console.log('page loaded')22  })23  .evaluate(() => {24  })25  .result(title => {26    console.log(title)27  })28  .end()29  .then(() => {30    console.log('done')31  })32  .catch(e => {33    console.log(e)34  })35#### chromy([options])

Using AI Code Generation

1const chromy = new Chromy({ port: 9222, visible: true });2  .wait('input[name="q"]')3  .type('input[name="q"]', 'chrome headless')4  .click('input[name="btnK"]')5  .wait('#resultStats')6  .evaluate(() => document.querySelector('#resultStats').textContent)7  .result(stats => console.log(stats))8  .end()9  .then(() => console.log('done'))10  .catch(e => console.log(e));11const chromy = new Chromy({ port: 9222, visible: true });12  .chain()13  .wait('input[name="q"]')14  .type('input[name="q"]', 'chrome headless')15  .click('input[name="btnK"]')16  .wait('#resultStats')17  .evaluate(() => document.querySelector('#resultStats').textContent)18  .result(stats => console.log(stats))19  .end()20  .then(() => console.log('done'))21  .catch(e => console.log(e));22const chromy = new Chromy({ port: 9222, visible: true });23  .run(async () => {24    await chromy.wait('input[name="q"]');25    await chromy.type('input[name="q"]', 'chrome headless');26    await chromy.click('input[name="btnK"]');27    await chromy.wait('#resultStats');28    const stats = await chromy.evaluate(() => document.querySelector('#resultStats').textContent);29    console.log(stats);30  })31  .then(() => console.log('done'))32  .catch(e => console.log(e

Using AI Code Generation

1const Chromy = require('chromy')2const chromy = new Chromy({ port: 9222 })3  .once('Network.responseReceived', msg => {4    console.log('responseReceived', msg)5  })6  .once('Network.requestWillBeSent', msg => {7    console.log('requestWillBeSent', msg)8  })9  .evaluate(() => {10    document.querySelector('input[name=q]').value = 'Hello world'11    document.querySelector('input[name=q]').form.submit()12  })13  .wait('h3')14  .result(h3 => {15    console.log(h3)16  })17  .end()18  .then(() => {19    console.log('All done')20  })21  .catch(e => console.error(e))22#### constructor(options)23#### goto(url, [options])24#### reload([options])25#### back()26#### forward()27#### click(selector)28#### tap(selector)29#### focus(selector)30#### type(selector, text)

Using AI Code Generation

1const chromy = new Chromy({ port: 9222 });2  .once('Network.responseReceived', event => {3    console.log(event);4  })5  .end()6  .then(() => chromy.close());7const chromy = new Chromy({ port: 9222 });8  .evaluate(() => {9    return document.title;10  })11  .end()12  .then(title => {13    console.log(title);14    return chromy.close();15  });16const chromy = new Chromy({ port: 9222 });17  .evaluateAsync(() => {18    return new Promise(resolve => {19      resolve(document.title);20    });21  })22  .end()23  .then(title => {24    console.log(title);25    return chromy.close();26  });27const chromy = new Chromy({ port: 9222 });28  .evaluate(() => {29    return $('title').text();30  })31  .end()32  .then(title => {33    console.log(title);34    return chromy.close();35  });36const chromy = new Chromy({ port: 9222 });37  .screenshot({ path: 'screenshot.png' })38  .end()39  .then(() => chromy.close());40const chromy = new Chromy({ port: 9222 });41  .pdf({ path: 'pdf.pdf'

Using AI Code Generation

1chromy.once('Network.responseReceived', function(event) {2    console.log(event)3})4chromy.on('Network.responseReceived', function(event) {5    console.log(event)6})7chromy.off('Network.responseReceived')8chromy.removeAllListeners()9chromy.removeListener('Network.responseReceived')10chromy.addListener('Network.responseReceived', function(event) {11    console.log(event)12})13chromy.on('Network.responseReceived', function(event) {14    console.log(event)15})16chromy.once('Network.responseReceived', function(event) {17    console.log(event)18})19chromy.off('Network.responseReceived')20chromy.removeAllListeners()21chromy.removeListener('Network.responseReceived')22chromy.addListener('Network.responseReceived', function(event) {23    console.log(event)24})25chromy.once('Network.responseReceived', function(event) {26    console.log(event)27})28chromy.on('Network.responseReceived', function(event) {29    console.log(event)30})31chromy.off('Network.responseReceived')32chromy.removeAllListeners()33chromy.removeListener('Network.responseReceived')34chromy.addListener('Network.responseReceived', function(event) {35    console.log(event)36})37chromy.once('Network.responseReceived', function(event) {38    console.log(event)39})40chromy.on('Network.responseReceived', function(event) {41    console.log(event)42})43chromy.off('Network.responseReceived')44chromy.removeAllListeners()

Using AI Code Generation

1var chromy = new Chromy();2chromy.chain()3  .once('Network.responseReceived', function() {4    console.log('received');5  })6  .end()7  .then(function() {8    chromy.close();9  });10var chromy = new Chromy();11chromy.chain()12  .onceEvent('Network.responseReceived', function() {13    console.log('received');14  })15  .end()16  .then(function() {17    chromy.close();18  });19This method is same as `on` method of [chrome-remote-interface](

Using AI Code Generation

1var chromy = require('./chromy.js');2chromy.once('Page.frameNavigated', function() { console.log('Page.frameNavigated'); });3chromy.close();4var CDP = require('chrome-remote-interface');5var chromy = new CDP();6chromy.once = function(event, callback) {7  chromy.on(event, function() {8    chromy.off(event);9    callback.apply(chromy, arguments);10  });11};12module.exports = chromy;

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.