TestMu Conference is Back! Future of Testing + AI LambdaTest G2 Announcement

How to use _set_driver method in Selene

Best Python code snippet using selene_python

test_nfs.py

Source:test_nfs.py

...344        self.configuration.nas_share_path = None345        self.configuration.nas_mount_options = None346        self.configuration.volume_dd_blocksize = '1M'347        self.context = context.get_admin_context()348    def _set_driver(self, extra_confs=None):349        # Overide the default configs350        if extra_confs:351            for config_name, config_value in extra_confs.items():352                setattr(self.configuration, config_name, config_value)353        self._driver = nfs.NfsDriver(configuration=self.configuration)354        self._driver.shares = {}355        self.mock_object(self._driver, '_execute')356    @ddt.data(NFS_CONFIG1, NFS_CONFIG2, NFS_CONFIG3, NFS_CONFIG4)357    def test_local_path(self, nfs_config):358        """local_path common use case."""359        self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE360        self._set_driver(extra_confs=nfs_config)361        drv = self._driver362        volume = fake_volume.fake_volume_obj(363            self.context,364            provider_location=self.TEST_NFS_EXPORT1)365        self.assertEqual(366            '/mnt/test/2f4f60214cf43c595666dd815f0360a4/%s' % volume.name,367            drv.local_path(volume))368    @ddt.data(NFS_CONFIG1, NFS_CONFIG2, NFS_CONFIG3, NFS_CONFIG4)369    def test_copy_image_to_volume(self, nfs_config):370        """resize_image common case usage."""371        mock_resize = self.mock_object(image_utils, 'resize_image')372        mock_fetch = self.mock_object(image_utils, 'fetch_to_raw')373        self._set_driver()374        drv = self._driver375        volume = fake_volume.fake_volume_obj(self.context,376                                             size=self.TEST_SIZE_IN_GB)377        test_img_source = 'volume-%s' % volume.id378        self.mock_object(drv, 'local_path', return_value=test_img_source)379        data = mock.Mock()380        data.virtual_size = 1 * units.Gi381        self.mock_object(image_utils, 'qemu_img_info', return_value=data)382        drv.copy_image_to_volume(None, volume, None, None)383        mock_fetch.assert_called_once_with(384            None, None, None, test_img_source, mock.ANY, run_as_root=True,385            size=self.TEST_SIZE_IN_GB)386        mock_resize.assert_called_once_with(test_img_source,387                                            self.TEST_SIZE_IN_GB,388                                            run_as_root=True)389    def test_get_mount_point_for_share(self):390        """_get_mount_point_for_share should calculate correct value."""391        self._set_driver()392        drv = self._driver393        self.configuration.nfs_mount_point_base = self.TEST_MNT_POINT_BASE394        self.assertEqual('/mnt/test/2f4f60214cf43c595666dd815f0360a4',395                         drv._get_mount_point_for_share(self.TEST_NFS_EXPORT1))396    def test_get_mount_point_for_share_given_extra_slash_in_state_path(self):397        """_get_mount_point_for_share should calculate correct value."""398        # This test gets called with the extra slash399        self.configuration.nfs_mount_point_base = (400            self.TEST_MNT_POINT_BASE_EXTRA_SLASH)401        # The driver gets called with the correct configuration and removes402        # the extra slash403        drv = nfs.NfsDriver(configuration=self.configuration)404        self.assertEqual('/opt/stack/data/cinder/mnt', drv.base)405        self.assertEqual(406            '/opt/stack/data/cinder/mnt/2f4f60214cf43c595666dd815f0360a4',407            drv._get_mount_point_for_share(self.TEST_NFS_EXPORT1))408    def test_get_capacity_info(self):409        """_get_capacity_info should calculate correct value."""410        self._set_driver()411        drv = self._driver412        stat_total_size = 2620544413        stat_avail = 2129984414        stat_output = '1 %d %d' % (stat_total_size, stat_avail)415        du_used = 490560416        du_output = '%d /mnt' % du_used417        with mock.patch.object(418                drv, '_get_mount_point_for_share') as mock_get_mount:419            mock_get_mount.return_value = self.TEST_MNT_POINT420            drv._execute.side_effect = [(stat_output, None),421                                        (du_output, None)]422            self.assertEqual((stat_total_size, stat_avail, du_used),423                             drv._get_capacity_info(self.TEST_NFS_EXPORT1))424            mock_get_mount.assert_called_once_with(self.TEST_NFS_EXPORT1)425            calls = [mock.call('stat', '-f', '-c', '%S %b %a',426                               self.TEST_MNT_POINT, run_as_root=True),427                     mock.call('du', '-sb', '--apparent-size',428                               '--exclude', '*snapshot*',429                               self.TEST_MNT_POINT, run_as_root=True)]430            drv._execute.assert_has_calls(calls)431    def test_get_capacity_info_for_share_and_mount_point_with_spaces(self):432        """_get_capacity_info should calculate correct value."""433        self._set_driver()434        drv = self._driver435        stat_total_size = 2620544436        stat_avail = 2129984437        stat_output = '1 %d %d' % (stat_total_size, stat_avail)438        du_used = 490560439        du_output = '%d /mnt' % du_used440        with mock.patch.object(441                drv, '_get_mount_point_for_share') as mock_get_mount:442            mock_get_mount.return_value = self.TEST_MNT_POINT_SPACES443            drv._execute.side_effect = [(stat_output, None),444                                        (du_output, None)]445            self.assertEqual((stat_total_size, stat_avail, du_used),446                             drv._get_capacity_info(447                                 self.TEST_NFS_EXPORT_SPACES))448            mock_get_mount.assert_called_once_with(449                self.TEST_NFS_EXPORT_SPACES)450            calls = [mock.call('stat', '-f', '-c', '%S %b %a',451                               self.TEST_MNT_POINT_SPACES, run_as_root=True),452                     mock.call('du', '-sb', '--apparent-size',453                               '--exclude', '*snapshot*',454                               self.TEST_MNT_POINT_SPACES, run_as_root=True)]455            drv._execute.assert_has_calls(calls)456    def test_load_shares_config(self):457        self._set_driver()458        drv = self._driver459        drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE460        with mock.patch.object(461                drv, '_read_config_file') as mock_read_config:462            config_data = []463            config_data.append(self.TEST_NFS_EXPORT1)464            config_data.append('#' + self.TEST_NFS_EXPORT2)465            config_data.append('')466            config_data.append(self.TEST_NFS_EXPORT2 + ' ' +467                               self.TEST_NFS_EXPORT2_OPTIONS)468            config_data.append('broken:share_format')469            mock_read_config.return_value = config_data470            drv._load_shares_config(drv.configuration.nfs_shares_config)471            mock_read_config.assert_called_once_with(472                self.TEST_SHARES_CONFIG_FILE)473            self.assertIn(self.TEST_NFS_EXPORT1, drv.shares)474            self.assertIn(self.TEST_NFS_EXPORT2, drv.shares)475            self.assertEqual(2, len(drv.shares))476            self.assertEqual(self.TEST_NFS_EXPORT2_OPTIONS,477                             drv.shares[self.TEST_NFS_EXPORT2])478    def test_load_shares_config_nas_opts(self):479        self._set_driver()480        drv = self._driver481        drv.configuration.nas_host = self.TEST_NFS_HOST482        drv.configuration.nas_share_path = self.TEST_NFS_SHARE_PATH483        drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE484        drv._load_shares_config(drv.configuration.nfs_shares_config)485        self.assertIn(self.TEST_NFS_EXPORT1, drv.shares)486        self.assertEqual(1, len(drv.shares))487    def test_ensure_shares_mounted_should_save_mounting_successfully(self):488        """_ensure_shares_mounted should save share if mounted with success."""489        self._set_driver()490        drv = self._driver491        config_data = []492        config_data.append(self.TEST_NFS_EXPORT1)493        drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE494        with mock.patch.object(495                drv, '_read_config_file') as mock_read_config:496            with mock.patch.object(497                    drv, '_ensure_share_mounted') as mock_ensure:498                mock_read_config.return_value = config_data499                drv._ensure_share_mounted(self.TEST_NFS_EXPORT1)500                mock_ensure.assert_called_once_with(self.TEST_NFS_EXPORT1)501    @mock.patch.object(remotefs, 'LOG')502    def test_ensure_shares_mounted_should_not_save_mounting_with_error(self,503                                                                       LOG):504        """_ensure_shares_mounted should not save share if failed to mount."""505        self._set_driver()506        drv = self._driver507        config_data = []508        config_data.append(self.TEST_NFS_EXPORT1)509        drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE510        with mock.patch.object(511                drv, '_read_config_file') as mock_read_config:512            with mock.patch.object(513                    drv, '_ensure_share_mounted') as mock_ensure:514                mock_read_config.return_value = config_data515                drv._ensure_share_mounted()516                self.assertEqual(0, len(drv._mounted_shares))517                mock_ensure.assert_called_once_with()518    def test_find_share_should_throw_error_if_there_is_no_mounted_share(self):519        """_find_share should throw error if there is no mounted shares."""520        self._set_driver()521        drv = self._driver522        drv._mounted_shares = []523        self.assertRaises(exception.NfsNoSharesMounted, drv._find_share,524                          self._simple_volume())525    def test_find_share(self):526        """_find_share simple use case."""527        self._set_driver()528        drv = self._driver529        drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]530        volume = fake_volume.fake_volume_obj(self.context,531                                             size=self.TEST_SIZE_IN_GB)532        with mock.patch.object(533                drv, '_get_capacity_info') as mock_get_capacity_info:534            mock_get_capacity_info.side_effect = [535                (5 * units.Gi, 2 * units.Gi, 2 * units.Gi),536                (10 * units.Gi, 3 * units.Gi, 1 * units.Gi)]537            self.assertEqual(self.TEST_NFS_EXPORT2,538                             drv._find_share(volume))539            calls = [mock.call(self.TEST_NFS_EXPORT1),540                     mock.call(self.TEST_NFS_EXPORT2)]541            mock_get_capacity_info.assert_has_calls(calls)542            self.assertEqual(2, mock_get_capacity_info.call_count)543    def test_find_share_should_throw_error_if_there_is_not_enough_space(self):544        """_find_share should throw error if there is no share to host vol."""545        self._set_driver()546        drv = self._driver547        drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]548        with mock.patch.object(549                drv, '_get_capacity_info') as mock_get_capacity_info:550            mock_get_capacity_info.side_effect = [551                (5 * units.Gi, 0, 5 * units.Gi),552                (10 * units.Gi, 0, 10 * units.Gi)]553            self.assertRaises(exception.NfsNoSuitableShareFound,554                              drv._find_share, self._simple_volume())555            calls = [mock.call(self.TEST_NFS_EXPORT1),556                     mock.call(self.TEST_NFS_EXPORT2)]557            mock_get_capacity_info.assert_has_calls(calls)558            self.assertEqual(2, mock_get_capacity_info.call_count)559    def _simple_volume(self, size=10):560        loc = self.TEST_NFS_EXPORT1561        return fake_volume.fake_volume_obj(self.context,562                                           display_name='volume_name',563                                           provider_location=loc,564                                           size=size)565    def test_create_sparsed_volume(self):566        self._set_driver()567        drv = self._driver568        volume = self._simple_volume()569        self.override_config('nfs_sparsed_volumes', True)570        with mock.patch.object(571                drv, '_create_sparsed_file') as mock_create_sparsed_file:572            with mock.patch.object(573                    drv, '_set_rw_permissions') as mock_set_rw_permissions:574                drv._do_create_volume(volume)575                mock_create_sparsed_file.assert_called_once_with(mock.ANY,576                                                                 mock.ANY)577                mock_set_rw_permissions.assert_called_once_with(mock.ANY)578    def test_create_nonsparsed_volume(self):579        self._set_driver()580        drv = self._driver581        self.configuration.nfs_sparsed_volumes = False582        volume = self._simple_volume()583        self.override_config('nfs_sparsed_volumes', False)584        with mock.patch.object(585                drv, '_create_regular_file') as mock_create_regular_file:586            with mock.patch.object(587                    drv, '_set_rw_permissions') as mock_set_rw_permissions:588                drv._do_create_volume(volume)589                mock_create_regular_file.assert_called_once_with(mock.ANY,590                                                                 mock.ANY)591                mock_set_rw_permissions.assert_called_once_with(mock.ANY)592    @mock.patch.object(nfs, 'LOG')593    def test_create_volume_should_ensure_nfs_mounted(self, mock_log):594        """create_volume ensures shares provided in config are mounted."""595        self._set_driver()596        drv = self._driver597        drv._find_share = mock.Mock()598        drv._find_share.return_value = self.TEST_NFS_EXPORT1599        drv._do_create_volume = mock.Mock()600        with mock.patch.object(601                drv, '_ensure_share_mounted') as mock_ensure_share:602            drv._ensure_share_mounted()603            volume = fake_volume.fake_volume_obj(self.context,604                                                 size=self.TEST_SIZE_IN_GB)605            drv.create_volume(volume)606            mock_ensure_share.assert_called_once_with()607    @mock.patch.object(nfs, 'LOG')608    def test_create_volume_should_return_provider_location(self, mock_log):609        """create_volume should return provider_location with found share."""610        self._set_driver()611        drv = self._driver612        drv._ensure_shares_mounted = mock.Mock()613        drv._do_create_volume = mock.Mock()614        with mock.patch.object(drv, '_find_share') as mock_find_share:615            mock_find_share.return_value = self.TEST_NFS_EXPORT1616            volume = fake_volume.fake_volume_obj(self.context,617                                                 size=self.TEST_SIZE_IN_GB)618            result = drv.create_volume(volume)619            self.assertEqual(self.TEST_NFS_EXPORT1,620                             result['provider_location'])621            mock_find_share.assert_called_once_with(volume)622    def test_delete_volume(self):623        """delete_volume simple test case."""624        self._set_driver()625        drv = self._driver626        drv._ensure_share_mounted = mock.Mock()627        volume = fake_volume.fake_volume_obj(628            self.context,629            display_name='volume-123',630            provider_location=self.TEST_NFS_EXPORT1)631        with mock.patch.object(drv, 'local_path') as mock_local_path:632            mock_local_path.return_value = self.TEST_LOCAL_PATH633            drv.delete_volume(volume)634            mock_local_path.assert_called_with(volume)635            drv._execute.assert_called_once()636    def test_delete_should_ensure_share_mounted(self):637        """delete_volume should ensure that corresponding share is mounted."""638        self._set_driver()639        drv = self._driver640        volume = fake_volume.fake_volume_obj(641            self.context,642            display_name='volume-123',643            provider_location=self.TEST_NFS_EXPORT1)644        with mock.patch.object(drv, '_ensure_share_mounted'):645            drv.delete_volume(volume)646    def test_delete_should_not_delete_if_provider_location_not_provided(self):647        """delete_volume shouldn't delete if provider_location missed."""648        self._set_driver()649        drv = self._driver650        volume = fake_volume.fake_volume_obj(self.context,651                                             name='volume-123',652                                             provider_location=None)653        with mock.patch.object(drv, '_ensure_share_mounted'):654            drv.delete_volume(volume)655            self.assertFalse(drv._execute.called)656    def test_get_volume_stats(self):657        """get_volume_stats must fill the correct values."""658        self._set_driver()659        drv = self._driver660        drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]661        with mock.patch.object(662                drv, '_ensure_shares_mounted') as mock_ensure_share:663            with mock.patch.object(664                    drv, '_get_capacity_info') as mock_get_capacity_info:665                mock_get_capacity_info.side_effect = [666                    (10 * units.Gi, 2 * units.Gi, 2 * units.Gi),667                    (20 * units.Gi, 3 * units.Gi, 3 * units.Gi)]668                drv._ensure_shares_mounted()669                drv.get_volume_stats()670                calls = [mock.call(self.TEST_NFS_EXPORT1),671                         mock.call(self.TEST_NFS_EXPORT2)]672                mock_get_capacity_info.assert_has_calls(calls)673                self.assertTrue(mock_ensure_share.called)674                self.assertEqual(30.0, drv._stats['total_capacity_gb'])675                self.assertEqual(5.0, drv._stats['free_capacity_gb'])676                self.assertEqual(5, drv._stats['reserved_percentage'])677                self.assertTrue(drv._stats['sparse_copy_volume'])678    def test_get_volume_stats_with_non_zero_reserved_percentage(self):679        """get_volume_stats must fill the correct values."""680        self.configuration.reserved_percentage = 10.0681        drv = nfs.NfsDriver(configuration=self.configuration)682        drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]683        with mock.patch.object(684                drv, '_ensure_shares_mounted') as mock_ensure_share:685            with mock.patch.object(686                    drv, '_get_capacity_info') as mock_get_capacity_info:687                mock_get_capacity_info.side_effect = [688                    (10 * units.Gi, 2 * units.Gi, 2 * units.Gi),689                    (20 * units.Gi, 3 * units.Gi, 3 * units.Gi)]690                drv._ensure_shares_mounted()691                drv.get_volume_stats()692                calls = [mock.call(self.TEST_NFS_EXPORT1),693                         mock.call(self.TEST_NFS_EXPORT2)]694                mock_get_capacity_info.assert_has_calls(calls)695                self.assertTrue(mock_ensure_share.called)696                self.assertEqual(30.0, drv._stats['total_capacity_gb'])697                self.assertEqual(5.0, drv._stats['free_capacity_gb'])698                self.assertEqual(10.0, drv._stats['reserved_percentage'])699    @ddt.data(True, False)700    def test_update_volume_stats(self, thin):701        self._set_driver()702        self._driver.configuration.max_over_subscription_ratio = 20.0703        self._driver.configuration.reserved_percentage = 5.0704        self._driver.configuration.nfs_sparsed_volumes = thin705        remotefs_volume_stats = {706            'volume_backend_name': 'fake_backend_name',707            'vendor_name': 'fake_vendor',708            'driver_version': 'fake_version',709            'storage_protocol': 'NFS',710            'total_capacity_gb': 100.0,711            'free_capacity_gb': 20.0,712            'reserved_percentage': 5.0,713            'QoS_support': False,714        }715        self.mock_object(remotefs.RemoteFSDriver, '_update_volume_stats')716        self._driver._stats = remotefs_volume_stats717        mock_get_provisioned_capacity = self.mock_object(718            self._driver, '_get_provisioned_capacity', return_value=25.0)719        self._driver._update_volume_stats()720        nfs_added_volume_stats = {721            'provisioned_capacity_gb': 25.0 if thin else 80.0,722            'max_over_subscription_ratio': 20.0,723            'reserved_percentage': 5.0,724            'thin_provisioning_support': thin,725            'thick_provisioning_support': not thin,726        }727        expected = remotefs_volume_stats728        expected.update(nfs_added_volume_stats)729        self.assertEqual(expected, self._driver._stats)730        self.assertEqual(thin, mock_get_provisioned_capacity.called)731    def _check_is_share_eligible(self, total_size, total_available,732                                 total_allocated, requested_volume_size):733        self._set_driver()734        with mock.patch.object(self._driver, '_get_capacity_info')\735                as mock_get_capacity_info:736            mock_get_capacity_info.return_value = (total_size,737                                                   total_available,738                                                   total_allocated)739            return self._driver._is_share_eligible('fake_share',740                                                   requested_volume_size)741    def test_is_share_eligible(self):742        self._set_driver()743        total_size = 100.0 * units.Gi744        total_available = 90.0 * units.Gi745        total_allocated = 10.0 * units.Gi746        requested_volume_size = 1  # GiB747        self.assertTrue(self._check_is_share_eligible(total_size,748                                                      total_available,749                                                      total_allocated,750                                                      requested_volume_size))751    def test_share_eligibility_with_reserved_percentage(self):752        self._set_driver()753        total_size = 100.0 * units.Gi754        total_available = 4.0 * units.Gi755        total_allocated = 96.0 * units.Gi756        requested_volume_size = 1  # GiB757        # Check used > used_ratio statement entered758        self.assertFalse(self._check_is_share_eligible(total_size,759                                                       total_available,760                                                       total_allocated,761                                                       requested_volume_size))762    def test_is_share_eligible_above_oversub_ratio(self):763        self._set_driver()764        total_size = 100.0 * units.Gi765        total_available = 10.0 * units.Gi766        total_allocated = 90.0 * units.Gi767        requested_volume_size = 10  # GiB768        # Check apparent_available <= requested_volume_size statement entered769        self.assertFalse(self._check_is_share_eligible(total_size,770                                                       total_available,771                                                       total_allocated,772                                                       requested_volume_size))773    def test_is_share_eligible_reserved_space_above_oversub_ratio(self):774        self._set_driver()775        total_size = 100.0 * units.Gi776        total_available = 10.0 * units.Gi777        total_allocated = 100.0 * units.Gi778        requested_volume_size = 1  # GiB779        # Check total_allocated / total_size >= oversub_ratio780        # statement entered781        self.assertFalse(self._check_is_share_eligible(total_size,782                                                       total_available,783                                                       total_allocated,784                                                       requested_volume_size))785    def test_extend_volume(self):786        """Extend a volume by 1."""787        self._set_driver()788        drv = self._driver789        volume = fake_volume.fake_volume_obj(790            self.context,791            id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',792            size=1,793            provider_location='nfs_share')794        path = 'path'795        newSize = volume['size'] + 1796        with mock.patch.object(image_utils, 'resize_image') as resize:797            with mock.patch.object(drv, 'local_path', return_value=path):798                with mock.patch.object(drv, '_is_share_eligible',799                                       return_value=True):800                    with mock.patch.object(drv, '_is_file_size_equal',801                                           return_value=True):802                        drv.extend_volume(volume, newSize)803                        resize.assert_called_once_with(path, newSize,804                                                       run_as_root=True)805    def test_extend_volume_failure(self):806        """Error during extend operation."""807        self._set_driver()808        drv = self._driver809        volume = fake_volume.fake_volume_obj(810            self.context,811            id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',812            size=1,813            provider_location='nfs_share')814        with mock.patch.object(image_utils, 'resize_image'):815            with mock.patch.object(drv, 'local_path', return_value='path'):816                with mock.patch.object(drv, '_is_share_eligible',817                                       return_value=True):818                    with mock.patch.object(drv, '_is_file_size_equal',819                                           return_value=False):820                        self.assertRaises(exception.ExtendVolumeError,821                                          drv.extend_volume, volume, 2)822    def test_extend_volume_insufficient_space(self):823        """Insufficient space on nfs_share during extend operation."""824        self._set_driver()825        drv = self._driver826        volume = fake_volume.fake_volume_obj(827            self.context,828            id='80ee16b6-75d2-4d54-9539-ffc1b4b0fb10',829            size=1,830            provider_location='nfs_share')831        with mock.patch.object(image_utils, 'resize_image'):832            with mock.patch.object(drv, 'local_path', return_value='path'):833                with mock.patch.object(drv, '_is_share_eligible',834                                       return_value=False):835                    with mock.patch.object(drv, '_is_file_size_equal',836                                           return_value=False):837                        self.assertRaises(exception.ExtendVolumeError,838                                          drv.extend_volume, volume, 2)839    def test_is_file_size_equal(self):840        """File sizes are equal."""841        self._set_driver()842        drv = self._driver843        path = 'fake/path'844        size = 2845        data = mock.MagicMock()846        data.virtual_size = size * units.Gi847        with mock.patch.object(image_utils, 'qemu_img_info',848                               return_value=data):849            self.assertTrue(drv._is_file_size_equal(path, size))850    def test_is_file_size_equal_false(self):851        """File sizes are not equal."""852        self._set_driver()853        drv = self._driver854        path = 'fake/path'855        size = 2856        data = mock.MagicMock()857        data.virtual_size = (size + 1) * units.Gi858        with mock.patch.object(image_utils, 'qemu_img_info',859                               return_value=data):860            self.assertFalse(drv._is_file_size_equal(path, size))861    @mock.patch.object(nfs, 'LOG')862    def test_set_nas_security_options_when_true(self, LOG):863        """Test higher level setting of NAS Security options.864        The NFS driver overrides the base method with a driver specific865        version.866        """867        self._set_driver()868        drv = self._driver869        drv._mounted_shares = [self.TEST_NFS_EXPORT1]870        is_new_install = True871        drv._ensure_shares_mounted = mock.Mock()872        drv._get_mount_point_for_share = mock.Mock(873            return_value=self.TEST_MNT_POINT)874        drv._determine_nas_security_option_setting = mock.Mock(875            return_value='true')876        drv.set_nas_security_options(is_new_install)877        self.assertEqual('true', drv.configuration.nas_secure_file_operations)878        self.assertEqual('true', drv.configuration.nas_secure_file_permissions)879        self.assertFalse(LOG.warning.called)880    @mock.patch.object(nfs, 'LOG')881    def test_set_nas_security_options_when_false(self, LOG):882        """Test higher level setting of NAS Security options.883        The NFS driver overrides the base method with a driver specific884        version.885        """886        self._set_driver()887        drv = self._driver888        drv._mounted_shares = [self.TEST_NFS_EXPORT1]889        is_new_install = False890        drv._ensure_shares_mounted = mock.Mock()891        drv._get_mount_point_for_share = mock.Mock(892            return_value=self.TEST_MNT_POINT)893        drv._determine_nas_security_option_setting = mock.Mock(894            return_value='false')895        drv.set_nas_security_options(is_new_install)896        self.assertEqual('false', drv.configuration.nas_secure_file_operations)897        self.assertEqual('false',898                         drv.configuration.nas_secure_file_permissions)899        self.assertTrue(LOG.warning.called)900    def test_set_nas_security_options_exception_if_no_mounted_shares(self):901        """Ensure proper exception is raised if there are no mounted shares."""902        self._set_driver()903        drv = self._driver904        drv._ensure_shares_mounted = mock.Mock()905        drv._mounted_shares = []906        is_new_cinder_install = 'does not matter'907        self.assertRaises(exception.NfsNoSharesMounted,908                          drv.set_nas_security_options,909                          is_new_cinder_install)910    def test_ensure_share_mounted(self):911        """Case where the mount works the first time."""912        self._set_driver()913        self.mock_object(self._driver._remotefsclient, 'mount')914        drv = self._driver915        drv.configuration.nfs_mount_attempts = 3916        drv.shares = {self.TEST_NFS_EXPORT1: ''}917        drv._ensure_share_mounted(self.TEST_NFS_EXPORT1)918        drv._remotefsclient.mount.called_once()919    @mock.patch('time.sleep')920    def test_ensure_share_mounted_exception(self, _mock_sleep):921        """Make the configured number of attempts when mounts fail."""922        num_attempts = 3923        self._set_driver()924        self.mock_object(self._driver._remotefsclient, 'mount',925                         side_effect=Exception)926        drv = self._driver927        drv.configuration.nfs_mount_attempts = num_attempts928        drv.shares = {self.TEST_NFS_EXPORT1: ''}929        self.assertRaises(exception.NfsException, drv._ensure_share_mounted,930                          self.TEST_NFS_EXPORT1)931        self.assertEqual(num_attempts, drv._remotefsclient.mount.call_count)932    def test_ensure_share_mounted_at_least_one_attempt(self):933        """Make at least one mount attempt even if configured for less."""934        min_num_attempts = 1935        num_attempts = 0936        self._set_driver()937        self.mock_object(self._driver._remotefsclient, 'mount',938                         side_effect=Exception)939        drv = self._driver940        drv.configuration.nfs_mount_attempts = num_attempts941        drv.shares = {self.TEST_NFS_EXPORT1: ''}942        self.assertRaises(exception.NfsException, drv._ensure_share_mounted,943                          self.TEST_NFS_EXPORT1)944        self.assertEqual(min_num_attempts,945                         drv._remotefsclient.mount.call_count)946    @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3],947              [NFS_CONFIG2, QEMU_IMG_INFO_OUT4],948              [NFS_CONFIG3, QEMU_IMG_INFO_OUT3],949              [NFS_CONFIG4, QEMU_IMG_INFO_OUT4])950    @ddt.unpack951    def test_copy_volume_from_snapshot(self, nfs_conf, qemu_img_info):952        self._set_driver(extra_confs=nfs_conf)953        drv = self._driver954        dest_volume = self._simple_volume()955        src_volume = self._simple_volume()956        fake_snap = fake_snapshot.fake_snapshot_obj(self.context)957        fake_snap.volume = src_volume958        img_out = qemu_img_info % {'volid': src_volume.id,959                                   'snapid': fake_snap.id,960                                   'size_gb': src_volume.size,961                                   'size_b': src_volume.size * units.Gi}962        img_info = imageutils.QemuImgInfo(img_out)963        mock_img_info = self.mock_object(image_utils, 'qemu_img_info')964        mock_img_info.return_value = img_info965        mock_convert_image = self.mock_object(image_utils, 'convert_image')966        vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,967                               drv._get_hash_str(src_volume.provider_location))968        src_vol_path = os.path.join(vol_dir, img_info.backing_file)969        dest_vol_path = os.path.join(vol_dir, dest_volume.name)970        info_path = os.path.join(vol_dir, src_volume.name) + '.info'971        snap_file = dest_volume.name + '.' + fake_snap.id972        snap_path = os.path.join(vol_dir, snap_file)973        size = dest_volume.size974        mock_read_info_file = self.mock_object(drv, '_read_info_file')975        mock_read_info_file.return_value = {'active': snap_file,976                                            fake_snap.id: snap_file}977        mock_permission = self.mock_object(drv, '_set_rw_permissions_for_all')978        drv._copy_volume_from_snapshot(fake_snap, dest_volume, size)979        mock_read_info_file.assert_called_once_with(info_path)980        mock_img_info.assert_called_once_with(snap_path, run_as_root=True)981        used_qcow = nfs_conf['nfs_qcow2_volumes']982        mock_convert_image.assert_called_once_with(983            src_vol_path, dest_vol_path, 'qcow2' if used_qcow else 'raw',984            run_as_root=True)985        mock_permission.assert_called_once_with(dest_vol_path)986    @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT3],987              [NFS_CONFIG2, QEMU_IMG_INFO_OUT4],988              [NFS_CONFIG3, QEMU_IMG_INFO_OUT3],989              [NFS_CONFIG4, QEMU_IMG_INFO_OUT4])990    @ddt.unpack991    def test_create_volume_from_snapshot(self, nfs_conf, qemu_img_info):992        self._set_driver(extra_confs=nfs_conf)993        drv = self._driver994        # Volume source of the snapshot we are trying to clone from. We need it995        # to have a different id than the default provided.996        src_volume = self._simple_volume(size=10)997        src_volume.id = six.text_type(uuid.uuid4())998        src_volume_dir = os.path.join(self.TEST_MNT_POINT_BASE,999                                      drv._get_hash_str(1000                                          src_volume.provider_location))1001        src_volume_path = os.path.join(src_volume_dir, src_volume.name)1002        fake_snap = fake_snapshot.fake_snapshot_obj(self.context)1003        # Fake snapshot based in the previous created volume1004        snap_file = src_volume.name + '.' + fake_snap.id1005        fake_snap.volume = src_volume1006        fake_snap.status = 'available'1007        fake_snap.size = 101008        # New fake volume where the snap will be copied1009        new_volume = self._simple_volume(size=10)1010        new_volume_dir = os.path.join(self.TEST_MNT_POINT_BASE,1011                                      drv._get_hash_str(1012                                          src_volume.provider_location))1013        new_volume_path = os.path.join(new_volume_dir, new_volume.name)1014        # Mocks1015        img_out = qemu_img_info % {'volid': src_volume.id,1016                                   'snapid': fake_snap.id,1017                                   'size_gb': src_volume.size,1018                                   'size_b': src_volume.size * units.Gi}1019        img_info = imageutils.QemuImgInfo(img_out)1020        mock_img_info = self.mock_object(image_utils, 'qemu_img_info')1021        mock_img_info.return_value = img_info1022        mock_ensure = self.mock_object(drv, '_ensure_shares_mounted')1023        mock_find_share = self.mock_object(drv, '_find_share',1024                                           return_value=self.TEST_NFS_EXPORT1)1025        mock_read_info_file = self.mock_object(drv, '_read_info_file')1026        mock_read_info_file.return_value = {'active': snap_file,1027                                            fake_snap.id: snap_file}1028        mock_convert_image = self.mock_object(image_utils, 'convert_image')1029        self.mock_object(drv, '_create_qcow2_file')1030        self.mock_object(drv, '_create_regular_file')1031        self.mock_object(drv, '_create_regular_file')1032        self.mock_object(drv, '_set_rw_permissions')1033        self.mock_object(drv, '_read_file')1034        ret = drv.create_volume_from_snapshot(new_volume, fake_snap)1035        # Test asserts1036        self.assertEqual(self.TEST_NFS_EXPORT1, ret['provider_location'])1037        used_qcow = nfs_conf['nfs_qcow2_volumes']1038        mock_convert_image.assert_called_once_with(1039            src_volume_path, new_volume_path, 'qcow2' if used_qcow else 'raw',1040            run_as_root=True)1041        mock_ensure.assert_called_once()1042        mock_find_share.assert_called_once_with(new_volume)1043    def test_create_volume_from_snapshot_status_not_available(self):1044        """Expect an error when the snapshot's status is not 'available'."""1045        self._set_driver()1046        drv = self._driver1047        src_volume = self._simple_volume()1048        fake_snap = fake_snapshot.fake_snapshot_obj(self.context)1049        fake_snap.volume = src_volume1050        new_volume = self._simple_volume()1051        new_volume['size'] = fake_snap['volume_size']1052        self.assertRaises(exception.InvalidSnapshot,1053                          drv.create_volume_from_snapshot,1054                          new_volume,1055                          fake_snap)1056    @ddt.data([NFS_CONFIG1, QEMU_IMG_INFO_OUT1],1057              [NFS_CONFIG2, QEMU_IMG_INFO_OUT2],1058              [NFS_CONFIG3, QEMU_IMG_INFO_OUT1],1059              [NFS_CONFIG4, QEMU_IMG_INFO_OUT2])1060    @ddt.unpack1061    def test_initialize_connection(self, nfs_confs, qemu_img_info):1062        self._set_driver(extra_confs=nfs_confs)1063        drv = self._driver1064        volume = self._simple_volume()1065        vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,1066                               drv._get_hash_str(volume.provider_location))1067        vol_path = os.path.join(vol_dir, volume.name)1068        mock_img_utils = self.mock_object(image_utils, 'qemu_img_info')1069        img_out = qemu_img_info % {'volid': volume.id, 'size_gb': volume.size,1070                                   'size_b': volume.size * units.Gi}1071        mock_img_utils.return_value = imageutils.QemuImgInfo(img_out)1072        self.mock_object(drv, '_read_info_file',1073                         return_value={'active': "volume-%s" % volume.id})1074        conn_info = drv.initialize_connection(volume, None)1075        mock_img_utils.assert_called_once_with(vol_path, run_as_root=True)1076        self.assertEqual('nfs', conn_info['driver_volume_type'])1077        self.assertEqual(volume.name, conn_info['data']['name'])1078        self.assertEqual(self.TEST_MNT_POINT_BASE,1079                         conn_info['mount_point_base'])1080    @mock.patch.object(image_utils, 'qemu_img_info')1081    def test_initialize_connection_raise_exception(self, mock_img_info):1082        self._set_driver()1083        drv = self._driver1084        volume = self._simple_volume()1085        qemu_img_output = """image: %s1086        file format: iso1087        virtual size: 1.0G (1073741824 bytes)1088        disk size: 173K1089        """ % volume['name']1090        mock_img_info.return_value = imageutils.QemuImgInfo(qemu_img_output)1091        self.assertRaises(exception.InvalidVolume,1092                          drv.initialize_connection,1093                          volume,1094                          None)1095    def test_create_snapshot(self):1096        self._set_driver()1097        drv = self._driver1098        volume = self._simple_volume()1099        self.configuration.nfs_snapshot_support = True1100        fake_snap = fake_snapshot.fake_snapshot_obj(self.context)1101        fake_snap.volume = volume1102        vol_dir = os.path.join(self.TEST_MNT_POINT_BASE,1103                               drv._get_hash_str(self.TEST_NFS_EXPORT1))1104        snap_file = volume['name'] + '.' + fake_snap.id1105        snap_path = os.path.join(vol_dir, snap_file)1106        info_path = os.path.join(vol_dir, volume['name']) + '.info'1107        with mock.patch.object(drv, '_local_path_volume_info',1108                               return_value=info_path), \1109                mock.patch.object(drv, '_read_info_file', return_value={}), \1110                mock.patch.object(drv, '_do_create_snapshot') \...

scrappa.py

Source:scrappa.py

...3from selenium.common.exceptions import TimeoutException, NoAlertPresentException, UnexpectedAlertPresentException4from selenium.webdriver.support.wait import WebDriverWait5class Scrappa:6    def _get(self, link, wait=True):7        # self._set_driver()8        self.driver.stop_client()9        try:10            self.driver.get('about:blank')11        except UnexpectedAlertPresentException as e:12            try:13                self.driver.switch_to.alert.accept()14            except NoAlertPresentException:15                pass16        try:17            self.driver.get(link)18            if wait:19                return self.driver.page_source20        except UnexpectedAlertPresentException as e:21            try:22                self.driver.switch_to.alert.accept()23            except NoAlertPresentException:24                pass25            try:26                if wait:27                    return self.driver.page_source28            except TimeoutException as e:29                self._set_driver()30                return self._get(link, wait)31        except TimeoutException as e:32            self._set_driver()33            return self._get(link, wait)34    def _set_driver(self):35        if hasattr(self, 'driver') and self.driver is not None:36            self.driver.quit()37        self.driver = self.browser(**self.browser_args)38        self.driver.set_page_load_timeout(self.timeout)39        self.driver.implicitly_wait(self.timeout)40        self.wait = WebDriverWait(self.driver, self.timeout)41    def _reload(self):42        self.driver.stop_client()43        self.driver.start_client()44    def __init__(self, timeout=5, browser=webdriver.Chrome, browser_args=None, whitelist_domains=None):45        self.timeout = timeout46        self.browser = browser47        default_browser_args = {}48        if browser == webdriver.Chrome:49            chrome_options = webdriver.ChromeOptions()50            prefs = {'profile.managed_default_content_settings.images': 2}51            chrome_options.add_experimental_option("prefs", prefs)52            if whitelist_domains is not None:53                chrome_options.add_argument(54                    f"--host-resolver-rules=MAP * 127.0.0.1, " + ", ".join([f"EXCLUDE {domain}" for domain in whitelist_domains])55                )56            default_browser_args['chrome_options'] = chrome_options57        elif browser == webdriver.Firefox:58            firefox_profile = webdriver.FirefoxProfile()59            firefox_profile.set_preference('permissions.default.image', 2)60            firefox_profile.set_preference('dom.ipc.plugins.enabled.libflashplayer.so', 'false')61            default_browser_args['firefox_profile'] = firefox_profile62        self.browser_args = default_browser_args if browser_args is None else browser_args63        self._set_driver()64    def soup(self, link):65        page_source = self._get(link)66        return BeautifulSoup(page_source, 'lxml')67    def soups(self, links):68        for link in links:69            yield self.soup(link)70    def get_soup(self):...

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.