KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use subcommand method in molecule

Best Python code snippet using molecule_python

__init__.py

Source:__init__.py

...133        # Equivalent to calling subcommand_name(dispatcher, "alpha", "beta", None)134        get_commands_registry()["commandname"]["subcommands"]["subcommand_name"]["worker"](dispatcher, "alpha", "beta")135    This is designed to work in combination with :func:`handle_subcommands`, below.136    """137    def subcommand(func):138        """Constructs the wrapper function for the decorated function."""139        global _commands_registry  # pylint: disable=global-statement,global-variable-not-assigned,invalid-name140        sig = inspect.signature(func)141        # What are the positional args of func?142        params = [p for p in sig.parameters.values() if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)]143        @wraps(func)144        def wrapper(*args, **kwargs):145            """Fill in None values for any omitted args, then call the decorated function."""146            args = list(args)147            while len(args) < len(params):148                args.append(None)149            return func(*args, **kwargs)150        # Register this function in the _commands_registry dictionary as a subcommand151        subcommand_name = func.__name__.replace("_", "-")152        subcommand_spec = {153            "worker": wrapper,154            # params[0] is always the "dispatcher" instance, omit that155            "params": [p.name.replace("_", "-") for p in params[1:]],156            "doc": func.__doc__.splitlines()[0],  # Split lines to get the first line of the docstring157        }158        add_subcommand(159            command_name=command, command_func=None, subcommand_name=subcommand_name, subcommand_spec=subcommand_spec160        )161        return wrapper162    return subcommand163def add_subcommand(command_name, command_func, subcommand_name, subcommand_spec):164    """Function to add additional subcommands without the need for a decorator function.165    Args:166        command_name (string): The parent command name167        command_func (function): The Parent command function168        subcommand_name (string): The name of the subcommand to add169        subcommand_spec (dict): A dictionary containing the definition of the subcommand:170    The subcommand_spec dictionary should have the following format:171    {172        'worker': worker_function,173        'params': [],174        'doc': "docstring"175    }176    """177    global _commands_registry  # pylint: disable=global-statement,global-variable-not-assigned,invalid-name178    # See issue #20 - depending on code structure, it's possible we may process a subcommand before we179    # have actually processed its parent command. This is normal and we need to handle it.180    _commands_registry.setdefault(command_name, {}).setdefault("subcommands", {})[subcommand_name] = subcommand_spec181    if not _commands_registry[command_name].get("function", False) and command_func:182        _commands_registry[command_name]["function"] = command_func183        _commands_registry[command_name]["doc"] = command_func.__doc__184@command_histogram.time()185def handle_subcommands(command, subcommand, params=(), dispatcher_class=None, context=None):186    """Generic worker function, to be used in combination with the @subcommand_of decorator above.187    ::188        @job("default")189        def commandname(subcommand, **kwargs):190            return handle_subcommands("command_name", subcommand, **kwargs)191        @subcommand_of("commandname")192        def first_subcommand(dispatcher, arg_1, arg_2):193            '''The first subcommand.'''194            pass195        @subcommand_of("commandname")196        def second_subcommand(dispatcher, arg_1):197            '''The second subcommand.'''198            pass199    chat client::200        > /commandname help201        "/commandname" subcommands:202        - /commandname first-subcommand [arg-1] [arg-2]  The first subcommand203        - /commandname second-subcommand [arg-1]         The second subcommand204    """205    if not context:206        context = {}207    registry = get_commands_registry()208    dispatcher = dispatcher_class(context=context)209    command_log = create_command_log(dispatcher, registry, command, subcommand, params)210    if subcommand == "help" or not subcommand:...

utils.py

Source:utils.py

1"""Utility functions for API view implementations."""2from datetime import datetime, timezone3import logging4import sys5from django.db.models import Q6from django.http import HttpResponse, JsonResponse7from nautobot_chatops.choices import AccessGrantTypeChoices, CommandStatusChoices8from nautobot_chatops.models import AccessGrant, CommandLog9from nautobot_chatops.metrics import request_command_cntr10logger = logging.getLogger(__name__)11try:12    from nautobot.core.celery import nautobot_task13    @nautobot_task14    def celery_worker_task(command, subcommand, params, dispatcher_module, dispatcher_name, context):15        """Task executed by Celery worker.16        Celery cannot serialize/deserialize objects, instead of passing the17        function, registry or dispatcher_class, we will have to look them up.18        """19        # import done here instead of up top to prevent circular imports.20        # Disabling cyclic-import as this does not affect the worker.21        # pylint: disable=import-outside-toplevel,cyclic-import22        from nautobot_chatops.workers import get_commands_registry23        # Looking up the function from the registry using the command.24        registry = get_commands_registry()25        function = registry[command]["function"]26        # Get dispatcher class from module, since the module is already loaded, we can use27        # sys.modules to map the dispatcher module string to the module. Then pull the class28        # using getattr.29        dispatcher_class = getattr(sys.modules[dispatcher_module], dispatcher_name)30        return function(31            subcommand,32            params=params,33            dispatcher_class=dispatcher_class,34            context=context,35        )36    def enqueue_task(*, command, subcommand, params, dispatcher_class, context, **kwargs):37        """Enqueue task with Celery worker."""38        return celery_worker_task.delay(39            command,40            subcommand,41            params=params,42            dispatcher_module=dispatcher_class.__module__,43            dispatcher_name=dispatcher_class.__name__,44            context=context,45        )46except ImportError:47    logger.info("INFO: Celery was not found - using Django RQ Worker")48    from django_rq import get_queue49    def enqueue_task(*, function, subcommand, params, dispatcher_class, context, **kwargs):50        """Enqueue task with Django RQ Worker."""51        return get_queue("default").enqueue(52            function,53            subcommand,54            params=params,55            dispatcher_class=dispatcher_class,56            context=context,57        )58def create_command_log(dispatcher, registry, command, subcommand, params=()):59    """Helper function to instantiate a new CommandLog based on the given information.60    Args:61        dispatcher (BaseDispatcher): Dispatcher subclass instance.62        registry (dict): See nautobot_chatops.workers.get_commands_registry().63        command (str): Top-level chatbot command -- *must* be present in the provided registry64        subcommand (str): Chatbot subcommand65        params (list): Any parameters passed to the command/subcommand66    Returns:67        CommandLog68    """69    if subcommand not in registry[command]["subcommands"]:70        # Valid command, invalid subcommand - e.g. "/clear meaningless-subcommand" - discard the subcommand71        subcommand = ""72    param_names = registry[command]["subcommands"].get(subcommand, {}).get("params", [])73    # The user might have specified more parameters than the command actually pays attention to,74    # e.g. "/nautobot get-circuit-providers meaningless-arg1 meaningless-arg2"75    # Only log the params that actually have meaning. Fortunately, zip() does this for us.76    params_list = list(zip(param_names, params))77    return CommandLog(78        user_name=dispatcher.context.get("user_name", "UNKNOWN"),79        user_id=dispatcher.context.get("user_id", "UNKNOWN"),80        platform=dispatcher.platform_name,81        platform_color=dispatcher.platform_color,82        command=command,83        subcommand=subcommand,84        start_time=datetime.now(timezone.utc),85        params=params_list,86    )87def check_and_enqueue_command(88    registry, command, subcommand, params, context, dispatcher_class89):  # pylint:disable=too-many-statements90    """Check whether the given command is permitted by any access grants, and enqueue it if permitted.91    For a command/subcommand to be permitted, we must have:92    - access grants for each of the AccessGrantTypeChoices (organization, channel, user)93    - that match either this command + subcommand exactly, or this command + "*" subcommand, or "*" command94    - and permit either this (organization, channel, user) exactly, or have a "*" value95    Args:96        registry (dict): See nautobot_chatops.workers.get_commands_registry().97        command (str): Top-level chatbot command98        subcommand (str): Chatbot subcommand99        params (list): Any parameters of the command/subcommand100        context (dict): Invocation context of the command101        dispatcher_class (class): Dispatcher concrete subclass102    Returns:103      HttpResponse104    """105    # Add to the context the command and subcommand issued to make available downstream106    context.update({"command": command, "subcommand": subcommand})107    dispatcher = dispatcher_class(context)108    # Mattermost requires at minimum an empty json response.109    response = HttpResponse()110    if dispatcher.platform_slug == "mattermost":111        data = {}112        response = JsonResponse(data)113    if command not in registry or "function" not in registry[command] or not callable(registry[command]["function"]):114        logger.error("Invalid/unknown command '%s'. Check registry:\n%s", command, registry)115        dispatcher.send_error(f"Something has gone wrong; I don't know how to handle command '{command}'.")116        return response117    command_log = create_command_log(dispatcher, registry, command, subcommand, params)118    if subcommand == "":119        # Command help - permit if any form of access to this command or any of its subcommands is permitted120        access_grants = AccessGrant.objects.filter(Q(command="*") | Q(command=command))121    else:122        # Actual subcommand  - permit only if this particular subcommand (or all commands/subcommands) are permitted123        access_grants = AccessGrant.objects.filter(124            Q(command="*") | Q(command=command, subcommand="*") | Q(command=command, subcommand=subcommand),125        )126    org_grants = access_grants.filter(grant_type=AccessGrantTypeChoices.TYPE_ORGANIZATION)127    if "org_id" not in context or not org_grants.filter(Q(value="*") | Q(value=context["org_id"])):128        label = context.get("org_id", "unspecified")129        if "org_name" in context:130            label += f" ({context['org_name']})"131        logger.warning(132            "Blocked %s %s: organization %s is not granted access",133            command,134            subcommand,135            label,136        )137        dispatcher.send_error(138            f"Access to this bot and/or command is not permitted in organization {label}, "139            "ask your Nautobot administrator to define an appropriate Access Grant"140        )141        request_command_cntr.labels(142            dispatcher.platform_slug,143            command,144            subcommand,145            CommandStatusChoices.STATUS_BLOCKED,146        ).inc()147        command_log.status = CommandStatusChoices.STATUS_BLOCKED148        command_log.details = f"Not permitted in organization {label}"149        command_log.runtime = datetime.now(timezone.utc) - command_log.start_time150        command_log.save()151        return response152    chan_grants = access_grants.filter(grant_type=AccessGrantTypeChoices.TYPE_CHANNEL)153    if "channel_id" not in context or not chan_grants.filter(Q(value="*") | Q(value=context["channel_id"])):154        label = context.get("channel_id", "unspecified")155        if "channel_name" in context:156            label += f" ({context['channel_name']})"157        logger.warning(158            "Blocked %s %s: channel %s is not granted access",159            command,160            subcommand,161            label,162        )163        dispatcher.send_error(164            f"Access to this bot and/or command is not permitted in channel {label}, "165            "ask your Nautobot administrator to define an appropriate Access Grant"166        )167        request_command_cntr.labels(168            dispatcher.platform_slug,169            command,170            subcommand,171            CommandStatusChoices.STATUS_BLOCKED,172        ).inc()173        command_log.status = CommandStatusChoices.STATUS_BLOCKED174        command_log.details = f"Not permitted in channel {label}"175        command_log.runtime = datetime.now(timezone.utc) - command_log.start_time176        command_log.save()177        return response178    user_grants = access_grants.filter(grant_type=AccessGrantTypeChoices.TYPE_USER)179    if "user_id" not in context or not user_grants.filter(Q(value="*") | Q(value=context["user_id"])):180        label = context.get("user_id", "unspecified")181        if "user_name" in context:182            label += f" ({context['user_name']})"183        logger.warning("Blocked %s %s: user %s is not granted access", command, subcommand, label)184        dispatcher.send_error(185            f"Access to this bot and/or command is not permitted by user {label}, "186            "ask your Nautobot administrator to define an appropriate Access Grant"187        )188        request_command_cntr.labels(189            dispatcher.platform_slug,190            command,191            subcommand,192            CommandStatusChoices.STATUS_BLOCKED,193        ).inc()194        command_log.status = CommandStatusChoices.STATUS_BLOCKED195        command_log.details = f"Not permitted by user {label}"196        command_log.runtime = datetime.now(timezone.utc) - command_log.start_time197        command_log.save()198        return response199    # If we made it here, we're permitted. Enqueue it for the worker200    enqueue_task(201        command=command,202        subcommand=subcommand,203        params=params,204        dispatcher_class=dispatcher_class,205        context=context,206        function=registry[command]["function"],207    )...

subcommand_command.py

Source:subcommand_command.py

...15        subparsers.required = True16        for subcommand in self.subcommands:17            subparsers.add_parser(subcommand)18    def handle(self, parsed_args=None, unparsed_args=None):19        self._run_subcommand(subcommand=parsed_args['subcommand'],20                             parsed_args=parsed_args,21                             unparsed_args=unparsed_args)22    def _run_subcommand(self, subcommand=None, parsed_args=None,23                        unparsed_args=None):24        try:25            subcommand_fn = self._get_subcommand_fn(subcommand=subcommand)26        except Exception as exc:27            raise self.InvalidSubcommandError(subcommand) from exc28        try:29            return self._call_subcommand_fn(subcommand_fn=subcommand_fn,30                                            parsed_args=parsed_args,31                                            unparsed_args=unparsed_args)32        except Exception as exc:33            raise self.SubcommandExecutionError(subcommand) from exc34    def _get_subcommand_fn(self, subcommand=None):35        return getattr(self, subcommand)36    def _call_subcommand_fn(self, subcommand_fn=None, parsed_args=None,...

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.