Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use pick_runner_command method in avocado

Best Python code snippet using avocado_python

runnable.py

Source:runnable.py

...135        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        """170        with open(recipe_path, encoding="utf-8") as recipe_file:171            recipe = json.load(recipe_file)172        config = ConfigDecoder.decode_set(recipe.get("config", {}))173        return cls.from_avocado_config(174            recipe.get("kind"),175            recipe.get("uri"),176            *recipe.get("args", ()),177            config=config,178            **recipe.get("kwargs", {}),179        )180    @classmethod181    def from_avocado_config(cls, kind, uri, *args, config=None, **kwargs):182        """Creates runnable with only essential config for runner of specific kind."""183        if not config:184            config = {}185        config = cls.filter_runnable_config(kind, config)186        return cls(kind, uri, *args, config=config, **kwargs)187    @staticmethod188    def filter_runnable_config(kind, config):189        """190        Returns only essential values for specific runner.191        It will use configuration from argument completed by values from192        config file and avocado default configuration.193        :param kind: Kind of runner which should use the configuration.194        :type kind: str195        :param config: Configuration values for runner. If some values will be196                       missing the default ones and from config file will be197                       used.198        :type config: dict199        :returns: Config dict, which has only values essential for runner200                  based on STANDALONE_EXECUTABLE_CONFIG_USED201        :rtype: dict202        """203        command = Runnable.pick_runner_command(kind)204        if command is not None:205            whole_config = settings.as_dict()206            whole_config.update(config)207            command = " ".join(command)208            configuration_used = STANDALONE_EXECUTABLE_CONFIG_USED.get(command)209            filtered_config = {}210            for config_item in configuration_used:211                filtered_config[config_item] = whole_config.get(config_item)212            return filtered_config213        else:214            raise ValueError(f"Unsupported kind of runnable: {kind}")215    def get_command_args(self):216        """217        Returns the command arguments that adhere to the runner interface218        This is useful for building 'runnable-run' and 'task-run' commands219        that can be executed on a command line interface.220        :returns: the arguments that can be used on an avocado-runner command221        :rtype: list222        """223        args = ["-k", self.kind]224        if self.uri is not None:225            args.append("-u")226            args.append(self.uri)227        if self.config:228            args.append("-c")229            args.append(json.dumps(self.config, cls=ConfigEncoder))230        for arg in self.args:231            args.append("-a")232            if arg.startswith("-"):233                arg = f"base64:{base64.b64encode(arg.encode()).decode('ascii')}"234            args.append(arg)235        if self.tags is not None:236            args.append(f"tags=json:{json.dumps(self.get_serializable_tags())}")237        if self.variant is not None:238            args.append(f"variant=json:{json.dumps(self.variant)}")239        if self.output_dir is not None:240            args.append(f"output_dir={self.output_dir}")241        for key, val in self.kwargs.items():242            if not isinstance(val, str) or isinstance(val, int):243                val = f"json:{json.dumps(val)}"244            args.append(f"{key}={val}")245        return args246    def get_dict(self):247        """248        Returns a dictionary representation for the current runnable249        This is usually the format that will be converted to a format250        that can be serialized to disk, such as JSON.251        :rtype: :class:`collections.OrderedDict`252        """253        recipe = collections.OrderedDict(kind=self.kind)254        if self.uri is not None:255            recipe["uri"] = self.uri256        recipe["config"] = self.config257        if self.args is not None:258            recipe["args"] = self.args259        kwargs = self.kwargs.copy()260        if self.tags is not None:261            kwargs["tags"] = self.get_serializable_tags()262        if self.variant is not None:263            kwargs["variant"] = self.variant264        if self.output_dir is not None:265            kwargs["output_dir"] = self.output_dir266        if kwargs:267            recipe["kwargs"] = kwargs268        return recipe269    def get_json(self):270        """271        Returns a JSON representation272        :rtype: str273        """274        return json.dumps(self.get_dict(), cls=ConfigEncoder)275    def get_serializable_tags(self):276        if self.tags is None:277            return {}278        tags = {}279        # sets are not serializable in json280        for key, val in self.tags.items():281            if isinstance(val, set):282                val = list(val)283            tags[key] = val284        return tags285    def write_json(self, recipe_path):286        """287        Writes a file with a JSON representation (also known as a recipe)288        """289        with open(recipe_path, "w", encoding="utf-8") as recipe_file:290            recipe_file.write(self.get_json())291    @staticmethod292    def get_capabilities_from_runner_command(runner_command, env=None):293        """Returns the capabilities of a given runner from a command.294        In case of failures, an empty capabilities dictionary is returned.295        When the capabilities are obtained, it also updates the296        :data:`STANDALONE_EXECUTABLE_CONFIG_USED` info.297        """298        cmd = runner_command + ["capabilities"]299        try:300            process = subprocess.Popen(301                cmd,302                stdin=subprocess.DEVNULL,303                stdout=subprocess.PIPE,304                stderr=subprocess.DEVNULL,305                env=env,306            )307        except (FileNotFoundError, PermissionError):308            return {}309        out, _ = process.communicate()310        try:311            capabilities = json.loads(out.decode())312        except json.decoder.JSONDecodeError:313            capabilities = {}314        # lists are not hashable, and here it'd make more sense to have315        # a command as it'd be seen in a command line anyway316        cmd = " ".join(runner_command)317        if cmd not in STANDALONE_EXECUTABLE_CONFIG_USED:318            STANDALONE_EXECUTABLE_CONFIG_USED[cmd] = capabilities.get(319                "configuration_used", []320            )321        return capabilities322    @staticmethod323    def is_kind_supported_by_runner_command(324        kind, runner_cmd, capabilities=None, env=None325    ):326        """Checks if a runner command that seems a good fit declares support."""327        if capabilities is None:328            capabilities = Runnable.get_capabilities_from_runner_command(329                runner_cmd, env330            )331        return kind in capabilities.get("runnables", [])332    @staticmethod333    def _module_exists(module_name):334        """Returns whether a nrunner "runner" module exists."""335        module_filename = f"{module_name}.py"336        mod_path = os.path.join("plugins", "runners", module_filename)337        return pkg_resources.resource_exists("avocado", mod_path)338    @staticmethod339    def pick_runner_command(kind, runners_registry=None):340        """Selects a runner command based on the runner kind.341        And when finding a suitable runner, keeps found runners in registry.342        This utility function will look at the given kind and try to find343        a matching runner.  The matching runner probe results are kept in344        a registry (that is modified by this function) so that further345        executions take advantage of previous probes.346        This is related to the :data:`SpawnMethod.STANDALONE_EXECUTABLE`347        :param kind: runners' kind348        :type kind: str349        :param runners_registry: a registry with previously found (and not350                                 found) runners keyed by runnable kind351        :type runners_registry: dict352        :returns: command line arguments to execute the runner353        :rtype: list of str or None354        """355        if runners_registry is None:356            runners_registry = RUNNERS_REGISTRY_STANDALONE_EXECUTABLE357        runner_cmd = runners_registry.get(kind)358        if runner_cmd is False:359            return None360        if runner_cmd is not None:361            return runner_cmd362        # When running Avocado Python modules, the interpreter on the new363        # process needs to know where Avocado can be found.364        env = os.environ.copy()365        env["PYTHONPATH"] = ":".join(p for p in sys.path)366        standalone_executable_cmd = [f"avocado-runner-{kind}"]367        if Runnable.is_kind_supported_by_runner_command(368            kind, standalone_executable_cmd369        ):370            runners_registry[kind] = standalone_executable_cmd371            return standalone_executable_cmd372        # attempt to find Python module files that are named after the373        # runner convention within the avocado.plugins.runners namespace dir.374        # Looking for the file only avoids an attempt to load the module375        # and should be a lot faster376        module_name = kind.replace("-", "_")377        if Runnable._module_exists(module_name):378            full_module_name = f"avocado.plugins.runners.{module_name}"379            candidate_cmd = [sys.executable, "-m", full_module_name]380            if Runnable.is_kind_supported_by_runner_command(381                kind, candidate_cmd, env=env382            ):383                runners_registry[kind] = candidate_cmd384                return candidate_cmd385        # look for the runner commands implemented in the base nrunner module386        candidate_cmd = [sys.executable, "-m", "avocado.core.nrunner"]387        if Runnable.is_kind_supported_by_runner_command(kind, candidate_cmd, env=env):388            runners_registry[kind] = candidate_cmd389            return candidate_cmd390        # exhausted probes, let's save the negative on the cache and avoid391        # future similar problems392        runners_registry[kind] = False393    def runner_command(self, runners_registry=None):394        """Selects a runner command based on the runner.395        And when finding a suitable runner, keeps found runners in registry.396        This utility function will look at the given task and try to find397        a matching runner.  The matching runner probe results are kept in398        a registry (that is modified by this function) so that further399        executions take advantage of previous probes.400        This is related to the :data:`SpawnMethod.STANDALONE_EXECUTABLE`401        :param runners_registry: a registry with previously found (and not402                                 found) runners keyed by runnable kind403        :type runners_registry: dict404        :returns: command line arguments to execute the runner405        :rtype: list of str or None406        """407        return Runnable.pick_runner_command(self.kind, runners_registry)408    def pick_runner_class_from_entry_point(self):409        """Selects a runner class from entry points based on kind.410        This is related to the :data:`SpawnMethod.PYTHON_CLASS`.411        :returns: a class that inherits from :class:`BaseRunner` or None412        """413        namespace = "avocado.plugins.runnable.runner"414        for ep in pkg_resources.iter_entry_points(namespace, self.kind):415            try:416                obj = ep.load()417                return obj418            except ImportError:419                return420    def pick_runner_class(self):421        """Selects a runner class from the registry based on kind....

runner.py

Source:runner.py

...24        runners_registry = RUNNERS_REGISTRY_STANDALONE_EXECUTABLE25    ok = []26    missing = []27    for runnable in runnables:28        runner = runnable.pick_runner_command(runnable.kind, runners_registry)29        if runner:30            ok.append(runnable)31        else:32            missing.append(runnable)33    return (ok, missing)34class BaseRunner(RunnableRunner):35    #: The "main Avocado" configuration keys (AKA namespaces) that36    #: this runners makes use of.37    CONFIGURATION_USED = []38    @staticmethod39    def prepare_status(status_type, additional_info=None):40        """Prepare a status dict with some basic information.41        This will add the keyword 'status' and 'time' to all status.42        :param: status_type: The type of event ('started', 'running',...

process.py

Source:process.py

...12            return False13        return runtime_task.spawner_handle.returncode is None14    async def spawn_task(self, runtime_task):15        task = runtime_task.task16        runner = task.runnable.pick_runner_command()17        args = runner[1:] + ['task-run'] + task.get_command_args()18        runner = runner[0]19        # pylint: disable=E113320        try:21            runtime_task.spawner_handle = await asyncio.create_subprocess_exec(22                runner,23                *args,24                stdout=asyncio.subprocess.PIPE,25                stderr=asyncio.subprocess.PIPE)26        except (FileNotFoundError, PermissionError):27            return False28        asyncio.ensure_future(self._collect_task(runtime_task.spawner_handle))29        return True30    @staticmethod...

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.