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)
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!