How to use _arg_decode_base64 method in avocado

Best Python code snippet using avocado_python

nrunner.py

Source:nrunner.py

...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 = '%s.py' % 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 ep.name == 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.read()330        process.stdout.close()331        stderr = process.stderr.read()332        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 = runner.run(suite)402        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        stream.seek(0)411        output = {'status': 'finished',412                  'result': result,413                  'output': stream.read()}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    """...

runnable.py

Source:runnable.py

...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        """...

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.