How to use _arg_decode_base64 method in avocado

Best Python code snippet using avocado_python Github


Full Screen

...78 self.kwargs, self.tags, self.requirements)79 @classmethod80 def from_args(cls, args):81 """Returns a runnable from arguments"""82 decoded_args = [_arg_decode_base64(arg) for arg in args.get('arg', ())]83 return cls(args.get('kind'),84 args.get('uri'),85 *decoded_args,86 config=json.loads(args.get('config', '{}')),87 **_key_val_args_to_kwargs(args.get('kwargs', [])))88 @classmethod89 def from_recipe(cls, recipe_path):90 """91 Returns a runnable from a runnable recipe file92 :param recipe_path: Path to a recipe file93 :rtype: instance of :class:`Runnable`94 """95 with open(recipe_path) as recipe_file:96 recipe = json.load(recipe_file)97 return cls(recipe.get('kind'),98 recipe.get('uri'),99 *recipe.get('args', ()),100 config=recipe.get('config', {}),101 **recipe.get('kwargs', {}))102 def get_command_args(self):103 """104 Returns the command arguments that adhere to the runner interface105 This is useful for building 'runnable-run' and 'task-run' commands106 that can be executed on a command line interface.107 :returns: the arguments that can be used on an avocado-runner command108 :rtype: list109 """110 args = ['-k', self.kind]111 if self.uri is not None:112 args.append('-u')113 args.append(self.uri)114 if self.config is not None:115 args.append('-c')116 args.append(json.dumps(self.config))117 for arg in self.args:118 args.append('-a')119 if arg.startswith('-'):120 arg = 'base64:%s' % base64.b64encode(arg.encode()).decode('ascii')121 args.append(arg)122 if self.tags is not None:123 args.append('tags=json:%s' % json.dumps(self.get_serializable_tags()))124 for key, val in self.kwargs.items():125 if not isinstance(val, str) or isinstance(val, int):126 val = "json:%s" % json.dumps(val)127 args.append('%s=%s' % (key, val))128 return args129 def get_dict(self):130 """131 Returns a dictionary representation for the current runnable132 This is usually the format that will be converted to a format133 that can be serialized to disk, such as JSON.134 :rtype: :class:`collections.OrderedDict`135 """136 recipe = collections.OrderedDict(kind=self.kind)137 if self.uri is not None:138 recipe['uri'] = self.uri139 recipe['config'] = self.config140 if self.args is not None:141 recipe['args'] = self.args142 kwargs = self.kwargs.copy()143 if self.tags is not None:144 kwargs['tags'] = self.get_serializable_tags()145 if kwargs:146 recipe['kwargs'] = kwargs147 return recipe148 def get_json(self):149 """150 Returns a JSON representation151 :rtype: str152 """153 return json.dumps(self.get_dict())154 def get_serializable_tags(self):155 tags = {}156 # sets are not serializable in json157 for key, val in self.tags.items():158 if isinstance(val, set):159 val = list(val)160 tags[key] = val161 return tags162 def write_json(self, recipe_path):163 """164 Writes a file with a JSON representation (also known as a recipe)165 """166 with open(recipe_path, 'w') as recipe_file:167 recipe_file.write(self.get_json())168 def is_kind_supported_by_runner_command(self, runner_command):169 """Checks if a runner command that seems a good fit declares support."""170 cmd = runner_command + ['capabilities']171 try:172 process = subprocess.Popen(cmd,173 stdin=subprocess.DEVNULL,174 stdout=subprocess.PIPE,175 stderr=subprocess.DEVNULL)176 except (FileNotFoundError, PermissionError):177 return False178 out, _ = process.communicate()179 try:180 capabilities = json.loads(out.decode())181 except json.decoder.JSONDecodeError:182 return False183 return self.kind in capabilities.get('runnables', [])184 def pick_runner_command(self, runners_registry=None):185 """Selects a runner command based on the runner.186 And when finding a suitable runner, keeps found runners in registry.187 This utility function will look at the given task and try to find188 a matching runner. The matching runner probe results are kept in189 a registry (that is modified by this function) so that further190 executions take advantage of previous probes.191 This is related to the :data:`SpawnMethod.STANDALONE_EXECUTABLE`192 :param runners_registry: a registry with previously found (and not193 found) runners keyed by runnable kind194 :param runners_registry: dict195 :returns: command line arguments to execute the runner196 :rtype: list of str or None197 """198 if runners_registry is None:199 runners_registry = RUNNERS_REGISTRY_STANDALONE_EXECUTABLE200 runner_cmd = runners_registry.get(self.kind)201 if runner_cmd is False:202 return None203 if runner_cmd is not None:204 return runner_cmd205 standalone_executable_cmd = ['avocado-runner-%s' % self.kind]206 if self.is_kind_supported_by_runner_command(standalone_executable_cmd):207 runners_registry[self.kind] = standalone_executable_cmd208 return standalone_executable_cmd209 # attempt to find Python module files that are named after the210 # runner convention within the avocado.core namespace dir.211 # Looking for the file only avoids an attempt to load the module212 # and should be a lot faster213 core_dir = os.path.dirname(os.path.abspath(__file__))214 module_name = "nrunner_%s" % self.kind.replace('-', '_')215 module_filename = '' % module_name216 if os.path.exists(os.path.join(core_dir, module_filename)):217 full_module_name = 'avocado.core.%s' % module_name218 candidate_cmd = [sys.executable, '-m', full_module_name]219 if self.is_kind_supported_by_runner_command(candidate_cmd):220 runners_registry[self.kind] = candidate_cmd221 return candidate_cmd222 # exhausted probes, let's save the negative on the cache and avoid223 # future similar problems224 runners_registry[self.kind] = False225 def pick_runner_class_from_entry_point(self):226 """Selects a runner class from entry points based on kind.227 This is related to the :data:`SpawnMethod.PYTHON_CLASS`. This228 complements the :data:`RUNNERS_REGISTRY_PYTHON_CLASS` on systems229 that have setuptools available.230 :returns: a class that inherits from :class:`BaseRunner` or None231 """232 if not PKG_RESOURCES_AVAILABLE:233 return234 namespace = 'avocado.plugins.runnable.runner'235 for ep in pkg_resources.iter_entry_points(namespace):236 if == self.kind:237 try:238 obj = ep.load()239 return obj240 except ImportError:241 return242 def pick_runner_class(self, runners_registry=None):243 """Selects a runner class from the registry based on kind.244 This is related to the :data:`SpawnMethod.PYTHON_CLASS`245 :param runners_registry: a registry with previously registered246 runner classes, keyed by runnable kind247 :param runners_registry: dict248 :returns: a class that inherits from :class:`BaseRunner`249 :raises: ValueError if kind there's no runner from kind of runnable250 """251 if runners_registry is None:252 runners_registry = RUNNERS_REGISTRY_PYTHON_CLASS253 runner = runners_registry.get(self.kind, None)254 if runner is None:255 runner = self.pick_runner_class_from_entry_point()256 if runner is not None:257 return runner258 raise ValueError('Unsupported kind of runnable: %s' % self.kind)259class BaseRunner:260 """261 Base interface for a Runner262 """263 def __init__(self, runnable):264 self.runnable = runnable265 def prepare_status(self, status_type, additional_info=None):266 """Prepare a status dict with some basic information.267 This will add the keyword 'status' and 'time' to all status.268 :param: status_type: The type of event ('started', 'running',269 'finished')270 :param: addional_info: Any additional information that you271 would like to add to the dict. This must be a272 dict.273 :rtype: dict274 """275 status = {'status': status_type,276 'time': time.monotonic()}277 if isinstance(additional_info, dict):278 status.update(additional_info)279 return status280 def run(self):281 yield {}282class NoOpRunner(BaseRunner):283 """284 Sample runner that performs no action before reporting FINISHED status285 Runnable attributes usage:286 * uri: not used287 * args: not used288 """289 def run(self):290 yield self.prepare_status('started')291 yield self.prepare_status('finished', {'result': 'pass'})292RUNNERS_REGISTRY_PYTHON_CLASS['noop'] = NoOpRunner293class ExecRunner(BaseRunner):294 """295 Runner for standalone executables with or without arguments296 Runnable attributes usage:297 * uri: path to a binary to be executed as another process298 * args: arguments to be given on the command line to the299 binary given by path300 * kwargs: key=val to be set as environment variables to the301 process302 """303 def run(self):304 env = None305 if self.runnable.kwargs:306 current = dict(os.environ)307 current.update(self.runnable.kwargs)308 env = current309 if env and 'PATH' not in env:310 env['PATH'] = os.environ.get('PATH')311 process = subprocess.Popen(312 [self.runnable.uri] + list(self.runnable.args),313 stdin=subprocess.DEVNULL,314 stdout=subprocess.PIPE,315 stderr=subprocess.PIPE,316 env=env)317 yield self.prepare_status('started')318 most_current_execution_state_time = None319 while process.poll() is None:320 time.sleep(RUNNER_RUN_CHECK_INTERVAL)321 now = time.monotonic()322 if most_current_execution_state_time is not None:323 next_execution_state_mark = (most_current_execution_state_time +324 RUNNER_RUN_STATUS_INTERVAL)325 if (most_current_execution_state_time is None or326 now > next_execution_state_mark):327 most_current_execution_state_time = now328 yield self.prepare_status('running')329 stdout = process.stdout.close()331 stderr = process.stderr.close()333 return_code = process.returncode334 yield self.prepare_status('finished', {'returncode': return_code,335 'stdout': stdout,336 'stderr': stderr})337RUNNERS_REGISTRY_PYTHON_CLASS['exec'] = ExecRunner338class ExecTestRunner(ExecRunner):339 """340 Runner for standalone executables treated as tests341 This is similar in concept to the Avocado "SIMPLE" test type, in which an342 executable returning 0 means that a test passed, and anything else means343 that a test failed.344 Runnable attributes usage is identical to :class:`ExecRunner`345 """346 def run(self):347 # Since Runners are standalone, and could be executed on a remote348 # machine in an "isolated" way, there is no way to assume a default349 # value, at this moment.350 skip_codes = self.runnable.config.get('runner.exectest.exitcodes.skip',351 [])352 for most_current_execution_state in super(ExecTestRunner, self).run():353 returncode = most_current_execution_state.get('returncode')354 if returncode in skip_codes:355 most_current_execution_state['result'] = 'skip'356 elif returncode == 0:357 most_current_execution_state['result'] = 'pass'358 else:359 most_current_execution_state['result'] = 'fail'360 yield most_current_execution_state361RUNNERS_REGISTRY_PYTHON_CLASS['exec-test'] = ExecTestRunner362class PythonUnittestRunner(BaseRunner):363 """364 Runner for Python unittests365 The runnable uri is used as the test name that the native unittest366 TestLoader will use to find the test. A native unittest test367 runner (TextTestRunner) will be used to execute the test.368 Runnable attributes usage:369 * uri: a "dotted name" that can be given to Python standard370 library's :meth:`unittest.TestLoader.loadTestsFromName`371 method. While it's not enforced, it's highly recommended372 that this is "a test method within a test case class" within373 a test module. Example is: "module.Class.test_method".374 * args: not used375 * kwargs: not used376 """377 @staticmethod378 def _uri_to_unittest_name(uri):379 if ':' in uri:380 module, class_method = uri.rsplit(':', 1)381 else:382 module = uri383 class_method = None384 if module.endswith('.py'):385 module = module[:-3]386 if module.startswith(os.path.curdir):387 module = module[1:]388 if module.startswith(os.path.sep):389 module = module[1:]390 module = module.replace(os.path.sep, ".")391 if class_method:392 return '%s.%s' % (module, class_method)393 return module394 @staticmethod395 def _run_unittest(uri, queue):396 sys.path.insert(0, ".")397 stream = io.StringIO()398 unittest_name = PythonUnittestRunner._uri_to_unittest_name(uri)399 suite = unittest.TestLoader().loadTestsFromName(unittest_name)400 runner = unittest.TextTestRunner(stream=stream, verbosity=0)401 unittest_result = if len(unittest_result.errors) > 0:403 result = 'error'404 elif len(unittest_result.failures) > 0:405 result = 'fail'406 elif len(unittest_result.skipped) > 0:407 result = 'skip'408 else:409 result = 'pass'410 output = {'status': 'finished',412 'result': result,413 'output':}414 stream.close()415 queue.put(output)416 def run(self):417 if not self.runnable.uri:418 error_msg = 'uri is required but was not given'419 yield self.prepare_status('finished', {'result': 'error',420 'output': error_msg})421 return422 queue = multiprocessing.SimpleQueue()423 process = multiprocessing.Process(target=self._run_unittest,424 args=(self.runnable.uri, queue))425 process.start()426 yield self.prepare_status('started')427 most_current_execution_state_time = None428 while queue.empty():429 time.sleep(RUNNER_RUN_CHECK_INTERVAL)430 now = time.monotonic()431 if most_current_execution_state_time is not None:432 next_execution_state_mark = (most_current_execution_state_time +433 RUNNER_RUN_STATUS_INTERVAL)434 if (most_current_execution_state_time is None or435 now > next_execution_state_mark):436 most_current_execution_state_time = now437 yield self.prepare_status('running')438 status = queue.get()439 status['time'] = time.monotonic()440 yield status441RUNNERS_REGISTRY_PYTHON_CLASS['python-unittest'] = PythonUnittestRunner442def _parse_key_val(argument):443 key_value = argument.split('=', 1)444 if len(key_value) < 2:445 msg = ('Invalid keyword parameter: "%s". Valid option must '446 'be a "KEY=VALUE" like expression' % argument)447 raise argparse.ArgumentTypeError(msg)448 return tuple(key_value)449def _arg_decode_base64(arg):450 """451 Decode arguments possibly encoded as base64452 :param arg: the possibly encoded argument453 :type arg: str454 :returns: the decoded argument455 :rtype: str456 """457 prefix = 'base64:'458 if arg.startswith(prefix):459 content = arg[len(prefix):]460 return base64.decodebytes(content.encode()).decode()461 return arg462def _kwarg_decode_json(value):463 """...

Full Screen

Full Screen Github


Full Screen

...13#: SpawnMethod.STANDALONE_EXECUTABLE compatible spawners14RUNNERS_REGISTRY_STANDALONE_EXECUTABLE = {}15#: The configuration that is known to be used by standalone runners16STANDALONE_EXECUTABLE_CONFIG_USED = {}17def _arg_decode_base64(arg):18 """19 Decode arguments possibly encoded as base6420 :param arg: the possibly encoded argument21 :type arg: str22 :returns: the decoded argument23 :rtype: str24 """25 prefix = "base64:"26 if arg.startswith(prefix):27 content = arg[len(prefix) :]28 return base64.decodebytes(content.encode()).decode()29 return arg30def _kwarg_decode_json(value):31 """32 Decode arguments possibly encoded as base6433 :param value: the possibly encoded argument34 :type value: str35 :returns: the decoded keyword argument as Python object36 """37 prefix = "json:"38 if value.startswith(prefix):39 content = value[len(prefix) :]40 return json.loads(content)41 return value42def _key_val_args_to_kwargs(kwargs):43 result = {}44 for key, val in kwargs:45 result[key] = _kwarg_decode_json(val)46 return result47class Runnable:48 """49 Describes an entity that be executed in the context of a task50 A instance of :class:`BaseRunner` is the entity that will actually51 execute a runnable.52 """53 def __init__(self, kind, uri, *args, config=None, **kwargs):54 if config is None:55 config = self.filter_runnable_config(kind, {})56 self.kind = kind57 #: The main reference to what needs to be run. This is free58 #: form, but commonly set to the path to a file containing the59 #: test or being the test, or an actual URI with multiple60 #: parts61 self.uri = uri62 #: This attributes holds configuration from Avocado proper63 #: that is passed to runners, as long as a runner declares64 #: its interest in using them with65 #: attr:`avocado.core.nrunner.runner.BaseRunner.CONFIGURATION_USED`66 self._config = {}67 self.config = config or {}68 self.args = args69 self.tags = kwargs.pop("tags", None)70 self.dependencies = kwargs.pop("dependencies", None)71 self.variant = kwargs.pop("variant", None)72 self.output_dir = kwargs.pop("output_dir", None)73 self.kwargs = kwargs74 self._identifier_format = config.get("runner.identifier_format", "{uri}")75 def __repr__(self):76 fmt = (77 '<Runnable kind="{}" uri="{}" config="{}" args="{}" '78 'kwargs="{}" tags="{}" dependencies="{}"> variant="{}"'79 )80 return fmt.format(81 self.kind,82 self.uri,83 self.config,84 self.args,85 self.kwargs,86 self.tags,87 self.dependencies,88 self.variant,89 )90 @property91 def identifier(self):92 """Runnable identifier respecting user's format string.93 This is still experimental and we have room for improvements.94 This property it will return an unique identifier for this runnable.95 Please use this property in order to respect user's customization.96 By default runnables has its '{uri}' as identifier.97 Custom formatter can be configured and currently we accept the98 following values as normal f-strings replacements: {uri}, {args},99 and {kwargs}. "args" and "kwargs" are special cases.100 For args, since it is a list, you can use in two different ways:101 "{args}" for the entire list, or "{args[n]}" for a specific element102 inside this list. The same is valid when using "{kwargs}". With103 kwargs, since it is a dictionary, you have to specify a key as index104 and then the values are used. For instance if you have a kwargs value105 named 'DEBUG', a valid usage could be: "{kwargs[DEBUG]}" and this will106 print the current value to this variable (i.e: True or False).107 Since this is formatter, combined values can be used. Example:108 "{uri}-{args}".109 """110 fmt = self._identifier_format111 # For the cases where there is no config (when calling the Runnable112 # directly113 if not fmt:114 return self.uri115 # For args we can use the entire list of arguments or with a specific116 # index.117 args = "-".join(self.args)118 if "args" in fmt and "[" in fmt:119 args = self.args120 # For kwargs we can use the entire list of values or with a specific121 # index.122 kwargs = "-".join(self.kwargs.values())123 if "kwargs" in fmt and "[" in fmt:124 kwargs = self.kwargs125 options = {"uri": self.uri, "args": args, "kwargs": kwargs}126 return fmt.format(**options)127 @property128 def config(self):129 return self._config130 @config.setter131 def config(self, config):132 """Sets the config values based on the runnable kind.133 This is not avocado config, it is a runnable config which is a subset134 of avocado config based on `STANDALONE_EXECUTABLE_CONFIG_USED` which135 describes essential configuration values for each runner kind.136 :param config: A config dict with new values for Runnable.137 :type config: dict138 """139 command = self.pick_runner_command(self.kind)140 if command is not None:141 command = " ".join(command)142 configuration_used = STANDALONE_EXECUTABLE_CONFIG_USED.get(command)143 if not set(configuration_used).issubset(set(config.keys())):144 LOG.warning(145 "The runnable config should have only values "146 "essential for its runner. In the next version of "147 "avocado, this will raise a Value Error. Please "148 "use avocado.core.nrunner.runnable.Runnable.filter_runnable_config "149 "or avocado.core.nrunner.runnable.Runnable.from_avocado_config"150 )151 self._config = config152 @classmethod153 def from_args(cls, args):154 """Returns a runnable from arguments"""155 decoded_args = [_arg_decode_base64(arg) for arg in args.get("arg", ())]156 return cls.from_avocado_config(157 args.get("kind"),158 args.get("uri"),159 *decoded_args,160 config=json.loads(args.get("config", "{}"), cls=ConfigDecoder),161 **_key_val_args_to_kwargs(args.get("kwargs", [])),162 )163 @classmethod164 def from_recipe(cls, recipe_path):165 """166 Returns a runnable from a runnable recipe file167 :param recipe_path: Path to a recipe file168 :rtype: instance of :class:`Runnable`169 """...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?