How to use local_working_path method in lisa

Best Python code snippet using lisa_python

service_config_util.py

Source:service_config_util.py

1import argparse2from executor.utils.log_utils import LogUtils3from pathlib import Path4class ServiceConfigUtil:5    """6    This is the service config parser for parsing service configurations.7    """8    # Default options for executor service.9    DEFAULT_OPTIONS = {'service_port': 5725,10                       'use_existing_data': False,11                       'local_proving_key_path': '/home/origo-executor/trial/pk',12                       'local_code_path': '/home/origo-executor/trial/code',13                       'local_working_path': '/home/origo/working',14                       'local_abi_path': '/home/origo-executor/trial/abi',15                       'zokrates_binary_path': '/home/origo/zokrates',16                       'chain_provider_type': 'http',17                       'http_uri': 'http://host.docker.internal:7545',18                       'ipc_path': '',19                       'websocket_uri': '',20                       'account_private_key': 'fdc8a9dbfed8638949e642ad4d564683cedf3f484d4ba778e6b22c1bb222ed66',21                       'account_public_key': '0x45cb118D08d0cb4bc3b8E4338635A8BBdF907136',22                       'encryption_type': 'rsa',23                       'rsa_key_path': '/home/origo-executor/tmp/private.pem',24                       'listener_poll_interval': 2}25    def __init__(self):26        """27        Init the ServiceConfigUtil.28        """29        self.options = {}30    def build_configurations(self):31        """32        Generate configuration for executor service.33        Returns:34            dictionary, {string, object}35        """36        parser = argparse.ArgumentParser(description='Origo Executor commandline config.')37        parser.add_argument('--service-port', dest='service_port', type=int,38                            help='The port to start Origo Executor service. Default: ' +39                                 str(self.DEFAULT_OPTIONS['service_port']))40        parser.add_argument('--config-file', dest='config_file_path', type=str,41                            help='The base Origo Executor configuration file path.')42        parser.add_argument('--local-proving-key-path', dest='local_pk_path', type=str,43                            help='The local proving key path to store the downloaded proving keys for contracts. '44                                 'Default: ' + self.DEFAULT_OPTIONS['local_proving_key_path'])45        parser.add_argument('--local-code-path', dest='local_code_path', type=str,46                            help='The local code path to store the downloaded Zokrates code for contracts. '47                                 'Default: ' + self.DEFAULT_OPTIONS['local_code_path'])48        parser.add_argument('--local-working-path', dest='local_working_path', type=str,49                            help='The local working path to generate intermediate data, which is always '50                                 'cleaned up after execution. Default: ' + self.DEFAULT_OPTIONS['local_working_path'])51        parser.add_argument('--local-abi-path', dest='local_abi_path', type=str,52                            help='The local abi path to store the downloaded abi files for contracts. Default: ' +53                                 self.DEFAULT_OPTIONS['local_abi_path'])54        parser.add_argument('--zokrates_binary_path', dest='zokrates_binary_path', type=str,55                            help='The path for Zokrates binary. Default: ' + self.DEFAULT_OPTIONS[56                                'zokrates_binary_path'])57        parser.add_argument('--use-existing-data', dest='use_existing_data', action='store_true', default=None,58                            help='Whether use existing local data and skip downloading again. Default: ' +59                                 str(self.DEFAULT_OPTIONS['use_existing_data']))60        # Chain related args.61        parser.add_argument('--chain-provider-type', dest='chain_provider_type', type=str,62                            help='The block chain provide type: http, ipc, websocket. Default: ' +63                                 self.DEFAULT_OPTIONS['chain_provider_type'])64        parser.add_argument('--http-uri', dest='http_uri', type=str,65                            help='The http uri to connect block chain network. Default: ' + self.DEFAULT_OPTIONS[66                                'http_uri'])67        parser.add_argument('--ipc-path', dest='ipc_path', type=str,68                            help='The ipc path to connect block chain network. Default: ' + self.DEFAULT_OPTIONS[69                                'ipc_path'])70        parser.add_argument('--websocket-uri', dest='websocket_uri', type=str,71                            help='The websocket uri to connect block chain network. Default: ' +72                                 self.DEFAULT_OPTIONS['websocket_uri'])73        parser.add_argument('--account-public-key', dest='account_public_key', type=str,74                            help='The chain account to work with. Default: ' + self.DEFAULT_OPTIONS[75                                'account_public_key'])76        parser.add_argument('--account-private-key', dest='account_private_key', type=str,77                            help='The working account private key. Default: ' + self.DEFAULT_OPTIONS[78                                'account_private_key'])79        # Encryption related args.80        parser.add_argument('--encryption-type', dest='encryption_type', type=str,81                            help='The encryption type used, currently only support: rsa. Default: ' +82                                 self.DEFAULT_OPTIONS['encryption_type'])83        parser.add_argument('--rsa-key-path', dest='rsa_key_path', type=str,84                            help='The RSA private key path. Default: ' + self.DEFAULT_OPTIONS['rsa_key_path'])85        # Listener related args.86        parser.add_argument('--listener-poll-interval', dest='listener_poll_interval', type=int,87                            help='The listener poll interval seconds. Default: ' +88                                 str(self.DEFAULT_OPTIONS['listener_poll_interval']))89        # Debug mode90        parser.add_argument('--debug-mode', dest='debug_mode', action='store_true', default=False,91                            help='Whether enable the debug mode for Origo Executor. Default: False')92        commandline_args = parser.parse_args()93        self._generate_options_from_commandline_args(commandline_args)94    @staticmethod95    def _get_options_from_config_file(config_path):96        parsed_options = {}97        with open(config_path) as f:98            config_lines = f.readlines()99            for line in config_lines:100                config_fields = line.rstrip().split('=')101                if len(config_fields) != 2:102                    LogUtils.error("Invalid config item:" + line)103                    exit(0)104                if config_fields[0] == 'listener_poll_interval' or config_fields[0] == 'service_port':105                    parsed_options[config_fields[0]] = int(config_fields[1])106                elif config_fields[0] == 'use_existing_data' or config_fields[0] == 'debug_mode':107                    if config_fields[1] in ['true', 'True', '1']:108                        parsed_options[config_fields[0]] = True109                    else:110                        parsed_options[config_fields[0]] = False111                else:112                    parsed_options[config_fields[0]] = config_fields[1]113        return parsed_options114    def _generate_options_from_commandline_args(self, input_args):115        """116        Generate the options based on commandline args.117        Args:118            input_args: args of ArgumentParser.119        Returns:120        """121        config_options = {}122        if input_args.config_file_path is not None and input_args.config_file_path != '':123            config_file = Path(input_args.config_file_path)124            if not config_file.is_file():125                LogUtils.error("Could not find provided config file:" + input_args.config_file_path)126                exit(0)127            config_options = self._get_options_from_config_file(input_args.config_file_path)128        for arg in vars(input_args):129            if getattr(input_args, arg) is not None:130                config_options[arg] = getattr(input_args, arg)131        for key, value in self.DEFAULT_OPTIONS.items():132            if key not in config_options or config_options[key] is None:133                config_options[key] = value134        self.options['service_port'] = config_options['service_port']135        self.options['proving_key_path'] = config_options['local_proving_key_path']136        self.options['code_path'] = config_options['local_code_path']137        self.options['working_path'] = config_options['local_working_path']138        self.options['zokrates_path'] = config_options['zokrates_binary_path']139        self.options['abi_path'] = config_options['local_abi_path']140        self.options['chain_config'] = {'provider_type': config_options['chain_provider_type'],141                                        'abi_path': config_options['local_abi_path'],142                                        'private_key': config_options['account_private_key'],143                                        'public_key': config_options['account_public_key'],144                                        'default_account': config_options['account_public_key']}145        self.options['encryption_info'] = {'type': config_options['encryption_type'],146                                           'rsa_key': config_options['rsa_key_path']}147        self.options['poll_interval'] = config_options['listener_poll_interval']148        self.options['use_existing_data'] = config_options['use_existing_data']149        self.options['debug_mode'] = config_options['debug_mode']150        if config_options['chain_provider_type'] == 'http':151            self.options['chain_config']['http_uri'] = config_options['http_uri']152        elif config_options['chain_provider_type'] == 'ipc':153            self.options['chain_config']['ipc_path'] = config_options['ipc_path']154        elif config_options['chain_provider_type'] == 'websocket':155            self.options['chain_config']['websocket_uri'] = config_options['websocket_uri']156        else:157            raise Exception("Unsupported chain provider type:" + config_options['chain_provider_type'])158    def get_options(self):159        LogUtils.info("\nStart the Origo Executor with the following configurations:")160        for key, value in self.options.items():161            LogUtils.info(key + ':\t\t' + str(value))162        return self.options163    def get_service_port(self):164        return self.options['service_port']165    def get_debug_mode(self):...

paths.py

Source:paths.py

...191    def test_DefaultPath(self):192        expected = self.make_local_path('den')193        self.assertEqual(194            expected,195            target.local_working_path()196        )197        self.assertTrue(os.path.exists(expected))198    def test_SubPath(self):199        expected = self.make_local_path('den/foo')200        self.assertEqual(201            expected,202            target.local_working_path('foo')203        )204        self.assertTrue(os.path.exists(expected))205    def test_NoCreate(self):206        expected = self.make_local_path('den/foo')207        self.assertEqual(208            expected,209            target.local_working_path('foo', ensure_exists=False)210        )211        self.assertFalse(os.path.exists(expected))212    def test_WithFilename(self):213        expected = self.make_local_path('den/bar.txt')214        self.assertEqual(215            expected,216            target.local_working_path(file_name='bar.txt')217        )218class LocalConfigFileTestCase(PathTestCaseBase):219    def test_DefaultPath(self):220        self.assertEqual(221            self.make_local_path('conf/nginx.conf'),222            target.local_config_file('nginx')223        )224    def test_WithNamePrefix(self):225        self.assertEqual(226            self.make_local_path('conf/nginx/alt-test.conf'),227            target.local_config_file('nginx', 'alt')228        )229    def test_WithAlternateExtension(self):230        self.assertEqual(...

pip.py

Source:pip.py

...39     - use requirements.txt from app/requirements.txt@revision40    """41    require('project_name')42    if not bundle_file:43        bundle_file = paths.local_working_path(file_name='%s.pybundle' % revision)44    path_to_requirements = paths.local_working_path(file_name='%s-requirements.txt' % revision)45    scm.export_file(revision, 'app/requirements.txt', path_to_requirements)46    utils.local('%s bundle -r %s %s' % (get_pip_cmd(), path_to_requirements, bundle_file))47    return bundle_file48def install_bundle(bundle_file, use_sudo=True, user=None):49    """50    Install pip bundle.51    To install into a virtualenv use the :ref:`vitualenv.activate` context52    manager.53    :param bundle_file: path to bundle file54    """55    file_name = os.path.basename(bundle_file)56    remote_bundle = paths.join_paths('/tmp/', file_name)57    put(bundle_file, remote_bundle)58    utils.run_as('%s install %s' % (get_pip_cmd(), remote_bundle), use_sudo, user)

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.