How to use _get_mountpoint method in autotest

Best Python code snippet using autotest_python

hashicorp_vault_connector.py

Source:hashicorp_vault_connector.py

...84        self.save_progress('Getting vault URL from asset configuration..._get_url()')85        config = self.get_config()86        url = config['vault_url']87        return url88    def _get_mountpoint(self):89        self.save_progress('Getting vault mountpoint from asset configuration..._get_mountpoint()')90        config = self.get_config()91        mountpoint = config['vault_mountpoint']92        return mountpoint93    def _create_vault_client(self, action_result):94        url = self._get_url()95        token = self._get_token()96        if url and token:97            try:98                vault_client = hvac.Client(url=url, token=token)99                return RetVal(action_result.set_status(phantom.APP_SUCCESS, 'Successfully created Hashicorp Vault Client'), vault_client)100            except Exception as e:101                err = self._get_error_message_from_exception(e)102                err = urllib.parse.unquote(err)103                return RetVal(104                    action_result.set_status(105                        phantom.APP_ERROR,106                        "Error in getting the Hashicorp Vault Client. {0}".format(err)107                    ), None108                )109        else:110            return RetVal(action_result.set_status(phantom.APP_ERROR, "Error in fetching url and token"), None)111    def _test_connectivity(self, action_result):112        ret_val, hvac_client = self._create_vault_client(action_result)113        if phantom.is_fail(ret_val):114            return action_result.get_status()115        if hvac_client:116            try:117                is_authenticated = hvac_client.is_authenticated()118                if is_authenticated:119                    self.save_progress('Successfully connected to Hashicorp vault with given credentials')120                    return action_result.set_status(phantom.APP_SUCCESS, 'Successfully connected to Hashicorp Vault')121                else:122                    self.save_progress('Failed to connect to Hashicorp vault with given credentials')123                    return action_result.set_status(phantom.APP_ERROR, 'Failed to connect to Hashicorp Vault')124            except Exception as e:125                err = self._get_error_message_from_exception(e)126                err = urllib.parse.unquote(err)127                return action_result.set_status(phantom.APP_ERROR, 'Error in authenticating Hashicorp Vault Client. {0}'.format(err))128        else:129            self.save_progress('Failed to create Hashicorp Vault client')130            return action_result.set_status(phantom.APP_ERROR, 'Failed to create Hashicorp Vault client')131    def _set_secret(self, param, action_result):132        ret_val, hvac_client = self._create_vault_client(action_result)133        if phantom.is_fail(ret_val):134            return action_result.get_status()135        mountpoint = self._get_mountpoint()136        path = param.get('location')137        secret = param.get('secret_json')138        self.save_progress(secret)139        try:140            secret = json.loads(secret)141            try:142                create_response = hvac_client.secrets.kv.v2.create_or_update_secret(mount_point=mountpoint, path=path, secret=secret)143                if create_response:144                    self.save_progress("Successfully added the secret")145                    action_result.add_data({"succeeded": True})146                    return action_result.set_status(phantom.APP_SUCCESS, 'Successfully added the secret')147                else:148                    self.save_progress("Failed to add the secret to Hashicorp Vault")149                    return action_result.set_status(phantom.APP_ERROR, "Failed to add the secret to Hashicorp Vault")150            except Exception as e:151                err = self._get_error_message_from_exception(e)152                err = urllib.parse.unquote(err)153                self.save_progress("Error occurred while storing the secret in Hashicorp vault. {}".format(err))154                return action_result.set_status(phantom.APP_ERROR, "Error occurred while storing the secret in Hashicorp vault. {}".format(err))155        except Exception as e:156            err = self._get_error_message_from_exception(e)157            self.save_progress("Please verify 'secret_json' action parameter. {}".format(err))158            return action_result.set_status(phantom.APP_ERROR, "Please verify 'secret_json' action parameter. {}".format(err))159    def _get_secret(self, param, action_result):160        ret_val, hvac_client = self._create_vault_client(action_result)161        if phantom.is_fail(ret_val):162            return action_result.get_status()163        mountpoint = self._get_mountpoint()164        path = param.get('location')165        try:166            read_response = hvac_client.secrets.kv.v2.read_secret_version(mount_point=mountpoint, path=path)167            if read_response:168                try:169                    secret_value = read_response['data']['data']170                    if secret_value:171                        self.save_progress("Secret value retrieved successfully")172                        action_result.add_data({"succeeded": True, "secret_value": secret_value})173                        return action_result.set_status(phantom.APP_SUCCESS, 'Successfully retrieved secret value')174                    else:175                        self.save_progress("No secret value retrieved from Hashicorp Vault for the specified path")176                        return action_result.set_status(phantom.APP_ERROR, "No secret value retrieved from Hashicorp Vault for the specified path")177                except Exception as e:178                    err = self._get_error_message_from_exception(e)179                    return action_result.set_status(phantom.APP_ERROR, "Error in getting secret value from the API response. {}".format(err))180            else:181                self.save_progress("Error in retrieving secret value from Hashicorp Vault")182                return action_result.set_status(phantom.APP_ERROR, "Error in retrieving secret value from Hashicorp Vault")183        except Exception as e:184            err = self._get_error_message_from_exception(e)185            err = urllib.parse.unquote(err)186            self.save_progress("Error in retrieving secret value from Hashicorp Vault. {}".format(err))187            return action_result.set_status(phantom.APP_ERROR, "Error in retrieving secret value from Hashicorp Vault. {}".format(err))188    def _list_secrets(self, param, action_result):189        ret_val, hvac_client = self._create_vault_client(action_result)190        if phantom.is_fail(ret_val):191            return action_result.get_status()192        mountpoint = self._get_mountpoint()193        path = param.get('location')194        try:195            list_secrets = hvac_client.secrets.kv.v2.list_secrets(mount_point=mountpoint, path=path)196            if list_secrets:197                try:198                    secrets = list_secrets['data']['keys']199                    if secrets:200                        self.save_progress("Secrets retrieved successfully")201                        action_result.add_data({"succeeded": True, "secret_values": secrets})202                        return action_result.set_status(phantom.APP_SUCCESS, 'Successfully retrieved secret values')203                    else:204                        self.save_progress("No secrets retrieved from Hashicorp Vault for the specified path")205                        return action_result.set_status(phantom.APP_ERROR, "No secrets retrieved from Hashicorp Vault for the specified path")206                except Exception as e:...

cgroups_test.py

Source:cgroups_test.py

1"""Unit test for cgroups module.2"""3from __future__ import absolute_import4from __future__ import division5from __future__ import print_function6from __future__ import unicode_literals7import io8import os9import shutil10import tempfile11import unittest12import mock13# Disable W0611: Unused import14import treadmill.tests.treadmill_test_skip_windows  # pylint: disable=W061115import treadmill16from treadmill import cgroups17from treadmill.fs import linux as fs_linux18_PROC_CGROUPS = """#subsys_name    hierarchy       num_cgroups     enabled19cpuset  4       1       020ns      10      3       021cpu     2       3       122cpuacct 3       3       123memory  7       3       124devices 5       1       025freezer 6       1       026net_cls 8       1       027blkio   1       1       028perf_event      11      1       029net_prio        9       1       030"""31_PROC_CGROUP = """3211:pids:/3310:cpuacct,cpu:/349:blkio:/358:freezer:/367:memory:/376:perf_event:/385:hugetlb:/394:cpuset:/403:devices:/system.slice/sshd.service412:net_prio,net_cls:/421:name=systemd:/system.slice/sshd.service43"""44class CGroupsTest(unittest.TestCase):45    """Tests for teadmill.cgroups.46    """47    def setUp(self):48        self.root = tempfile.mkdtemp()49    def tearDown(self):50        if self.root and os.path.isdir(self.root):51            shutil.rmtree(self.root)52    @mock.patch('treadmill.cgroups._available_subsystems',53                mock.Mock(return_value=[54                    'cpu', 'cpuacct', 'cpuset', 'memory', 'blkio'55                ]))56    @mock.patch('treadmill.fs.linux.list_mounts', mock.Mock(spec_set=True))57    def test_read_mounted_cgroups(self):58        """Test get mounted point.59        """60        fs_linux.list_mounts.return_value = [61            fs_linux.MountEntry(62                mount_id=24, parent_id=17,63                source='tmpfs', target='/sys/fs/cgroup',64                fs_type='tmpfs',65                mnt_opts={66                    'mode=755',67                    'nodev',68                    'noexec',69                    'nosuid',70                    'ro',71                }72            ),73            fs_linux.MountEntry(74                mount_id=24, parent_id=17,75                source='cgroup', target='/sys/fs/cgroup/devices',76                fs_type='cgroup',77                mnt_opts={78                    'clone_children',79                    'noexec',80                    'relatime',81                    'rw',82                    'nosuid',83                    'devices',84                    'nodev',85                }86            ),87            fs_linux.MountEntry(88                mount_id=24, parent_id=17,89                source='cgroup', target='/sys/fs/cgroup/memory',90                fs_type='cgroup',91                mnt_opts={92                    'clone_children',93                    'noexec',94                    'relatime',95                    'rw',96                    'nosuid',97                    'memory',98                    'nodev',99                }100            ),101            fs_linux.MountEntry(102                mount_id=24, parent_id=17,103                source='cgroup', target='/sys/fs/cgroup/blkio',104                fs_type='cgroup',105                mnt_opts={106                    'clone_children',107                    'blkio',108                    'noexec',109                    'relatime',110                    'rw',111                    'nosuid',112                    'nodev',113                }114            ),115            fs_linux.MountEntry(116                mount_id=24, parent_id=17,117                source='cgroup', target='/sys/fs/cgroup/net_cls,net_prio',118                fs_type='cgroup',119                mnt_opts={120                    'clone_children',121                    'noexec',122                    'relatime',123                    'net_cls',124                    'rw',125                    'nosuid',126                    'nodev',127                    'net_prio',128                }129            ),130            fs_linux.MountEntry(131                mount_id=24, parent_id=17,132                source='cgroup', target='/sys/fs/cgroup/pids',133                fs_type='cgroup',134                mnt_opts={135                    'pids',136                    'clone_children',137                    'noexec',138                    'relatime',139                    'rw',140                    'nosuid',141                    'nodev',142                }143            ),144            fs_linux.MountEntry(145                mount_id=24, parent_id=17,146                source='cgroup', target='/sys/fs/cgroup/hugetlb',147                fs_type='cgroup',148                mnt_opts={149                    'hugetlb',150                    'clone_children',151                    'noexec',152                    'relatime',153                    'rw',154                    'nosuid',155                    'nodev',156                }157            ),158            fs_linux.MountEntry(159                mount_id=24, parent_id=17,160                source='cgroup', target='/sys/fs/cgroup/freezer',161                fs_type='cgroup',162                mnt_opts={163                    'clone_children',164                    'freezer',165                    'noexec',166                    'relatime',167                    'rw',168                    'nosuid',169                    'nodev',170                }171            ),172            fs_linux.MountEntry(173                mount_id=24, parent_id=17,174                source='cgroup', target='/sys/fs/cgroup/cpu,cpuacct',175                fs_type='cgroup',176                mnt_opts={177                    'clone_children',178                    'noexec',179                    'relatime',180                    'cpuacct',181                    'cpu',182                    'rw',183                    'nosuid',184                    'nodev',185                }186            ),187            fs_linux.MountEntry(188                mount_id=24, parent_id=17,189                source='cgroup', target='/sys/fs/cgroup/perf_event',190                fs_type='cgroup',191                mnt_opts={192                    'clone_children',193                    'noexec',194                    'relatime',195                    'rw',196                    'perf_event',197                    'nosuid',198                    'nodev',199                }200            ),201            fs_linux.MountEntry(202                mount_id=24, parent_id=17,203                source='cgroup', target='/sys/fs/cgroup/cpuset',204                fs_type='cgroup',205                mnt_opts={206                    'clone_children',207                    'cpuset',208                    'noexec',209                    'relatime',210                    'rw',211                    'nosuid',212                    'nodev',213                }214            ),215        ]216        subsystems2mounts = cgroups.read_mounted_cgroups()217        self.assertEqual(218            subsystems2mounts,219            {220                'blkio': ['/sys/fs/cgroup/blkio'],221                'cpu': ['/sys/fs/cgroup/cpu,cpuacct'],222                'cpuacct': ['/sys/fs/cgroup/cpu,cpuacct'],223                'cpuset': ['/sys/fs/cgroup/cpuset'],224                'memory': ['/sys/fs/cgroup/memory'],225            }226        )227    @mock.patch('io.open', mock.mock_open(read_data=_PROC_CGROUP.strip()))228    def test_proc_cgroups(self):229        """Test reading a process' cgroups.230        """231        io.open.return_value.__iter__ = lambda x: iter(x.readlines())232        res = cgroups.proc_cgroups()233        self.assertEqual(234            res,235            {236                'blkio': '/',237                'cpuacct,cpu': '/',238                'cpuset': '/',239                'devices': '/system.slice/sshd.service',240                'freezer': '/',241                'hugetlb': '/',242                'memory': '/',243                'name=systemd': '/system.slice/sshd.service',244                'net_prio,net_cls': '/',245                'perf_event': '/',246                'pids': '/',247            }248        )249    @mock.patch('treadmill.cgroups.get_data',250                mock.Mock(side_effect=['2', '1\n2', '-1', '']))251    def test_get_value(self):252        """Test cgroup value fetching.253        """254        value = cgroups.get_value('memory', 'foo', 'memory.usage_in_bytes')255        self.assertEqual(value, 2)256        value = cgroups.get_value('memory', 'foo', 'memory.usage_in_bytes')257        self.assertEqual(value, 1)258        value = cgroups.get_value('memory', 'foo', 'memory.usage_in_bytes')259        self.assertEqual(value, 0)260        value = cgroups.get_value('memory', 'foo', 'memory.usage_in_bytes')261        self.assertEqual(value, 0)262    @mock.patch('treadmill.cgroups._get_mountpoint',263                mock.Mock(return_value='/cgroups'))264    @mock.patch('treadmill.fs.mkdir_safe', mock.Mock(set_spec=True))265    def test_create(self):266        """Tests cgroup creation.267        """268        group = os.path.join('treadmill', 'apps', 'test1')269        cgroups.create('cpu', group)270        cgroups.create('memory', group)271        cgroups.create('cpuacct', group)272        treadmill.fs.mkdir_safe.assert_has_calls(273            [274                mock.call('/cgroups/treadmill/apps/test1'),275                mock.call('/cgroups/treadmill/apps/test1'),276                mock.call('/cgroups/treadmill/apps/test1'),277            ]278        )279    @mock.patch('treadmill.cgroups._get_mountpoint', mock.Mock())280    def test_extractpath(self):281        """Test cgroup name from a cgroup path.282        """283        # pylint: disable=W0212284        treadmill.cgroups._get_mountpoint.return_value = '/fs/cgroup/memory'285        cgrp = cgroups.extractpath('/fs/cgroup/memory/treadmill/core',286                                   'memory')287        self.assertEqual(cgrp, 'treadmill/core')288        cgrp = cgroups.extractpath('/fs/cgroup/memory/treadmill/core/foo',289                                   'memory', 'foo')290        self.assertEqual(cgrp, 'treadmill/core')291        with self.assertRaises(ValueError):292            cgroups.extractpath('/cgroup/memory/treadmill/core', 'memory')293        with self.assertRaises(ValueError):294            cgroups.extractpath('/fs/cgroup/memory/treadmill/core/foo',295                                'cpu', 'bar')296    @mock.patch('treadmill.cgroups._get_mountpoint', mock.Mock())297    @mock.patch('os.rmdir', mock.Mock())298    def test_delete(self):299        """Tests cgroup deletion.300        """301        # pylint: disable=W0212302        cgroups_dir = os.path.join(self.root, 'cgroups')303        treadmill.cgroups._get_mountpoint.return_value = cgroups_dir304        group = os.path.join('treadmill', 'apps', 'test1')305        # Create a directory for the cgroup306        os.makedirs(os.path.join(cgroups_dir, group))307        cgroups.delete('cpu', group)308        os.rmdir.assert_called_once_with(309            os.path.join(cgroups_dir, group)310        )311    @mock.patch('treadmill.cgroups._get_mountpoint',312                mock.Mock(return_value='/cgroups'))313    @mock.patch('io.open', mock.mock_open())314    def test_join(self):315        """Tests joining the cgroup.316        """317        group = os.path.join('treadmill', 'apps', 'test1')318        cgroups.join('cpu', group, '1234')319        io.open.assert_called_once_with(320            '/cgroups/treadmill/apps/test1/tasks', 'w')321        io.open().write.assert_called_once_with('1234')322    @mock.patch('io.open', mock.mock_open(read_data=_PROC_CGROUPS))323    def test__available_subsystems(self):324        """Test parsince available subsystems.325        """326        # pylint: disable=W0212327        io.open.return_value.__iter__ = lambda x: iter(x.readlines())328        subsystems = cgroups._available_subsystems()329        self.assertEqual(['cpu', 'cpuacct', 'memory'], subsystems)330if __name__ == '__main__':...

fssync.py

Source:fssync.py

...34        raise NotImplementedError()35class XFSSync(FSSync):36    """ Sync application for XFS. """37    ext = availability.XFSFREEZE_APP38    def _get_mountpoint(self, root=None):39        mountpoint = self.fs.system_mountpoint40        if root is not None and root.replace('/', ''):41            if mountpoint == root:42                mountpoint = '/'43            else:44                mountpoint = mountpoint[len(root):]45        return mountpoint46    def _freeze_command(self, root=None):47        return [str(self.ext), "-f", self._get_mountpoint(root=root)]48    def _unfreeze_command(self, root=None):49        return [str(self.ext), "-u", self._get_mountpoint(root=root)]50    def do_task(self, root="/"):51        # pylint: disable=arguments-differ52        error_msgs = self.availability_errors53        if error_msgs:54            raise FSError("\n".join(error_msgs))55        error_msg = None56        try:57            rc = util.run_program(self._freeze_command(root=root), root=root)58        except OSError as e:59            error_msg = "failed to sync filesytem: %s" % e60        error_msg = error_msg or rc61        try:62            rc = util.run_program(self._unfreeze_command(root=root), root=root)63        except OSError as e:...

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.