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

How to use _status_map method in avocado

Best Python code snippet using avocado_python

_progress_manager.py

Source:_progress_manager.py

1from __future__ import annotations2from contextlib import asynccontextmanager3from types import TracebackType4from typing import AsyncContextManager, AsyncIterator, Mapping, Optional, Type5import anyio6from anyio.streams.memory import MemoryObjectSendStream7from ._status import Cancelled, Completed, Failed, Idle, Running, Status8StatusMap = Mapping[str, Status]  # Ideally, this is an immutable mapping9StatusStream = MemoryObjectSendStream[StatusMap]10class ProgressManager(AsyncContextManager["ProgressManager"]):11    """Convenience context manager for progress reports."""12    def __init__(13        self,14        status_map: StatusMap,15        *,16        status_stream: Optional[StatusStream] = None,17    ) -> None:18        self._initial_status_map = status_map19        self._status_map = self._initial_status_map20        self._status_stream = status_stream21    @asynccontextmanager22    async def step(self, name: str) -> AsyncIterator[None]:23        """Mark the nested code as a step with the given name.24        Automatically sends out `Running`, `Cancelled`, `Failed`, etc.25        status messages.26        """27        await self._run(name)28        try:29            yield30        except anyio.get_cancelled_exc_class():31            with anyio.CancelScope(shield=True):32                await self._cancel(name)33            raise34        except (Exception, anyio.ExceptionGroup):35            with anyio.CancelScope(shield=True):36                await self._fail(name)37            raise38        else:39            await self._complete(name)40    async def skip(self, name: str) -> None:41        """Skip the step with the given name."""42        status = self._status_map[name]43        if not isinstance(status, Idle):44            raise RuntimeError('Can only skip from the "Idle" status.')45        self._status_map = {**self._status_map, name: status.skip()}46        await self._send_status_update()47    async def _run(self, name: str) -> None:48        status = self._status_map[name]49        if not isinstance(status, (Idle, Completed, Cancelled, Failed)):50            raise RuntimeError(51                'Can only run from the "Idle", "Completed", '52                '"Cancelled", or "Failed" status.'53            )54        self._status_map = {**self._status_map, name: status.run()}55        await self._send_status_update()56    async def _cancel(self, name: str) -> None:57        status = self._status_map[name]58        if not isinstance(status, Running):59            raise RuntimeError('Can only cancel from the "Running" status.')60        self._status_map = {**self._status_map, name: status.cancel()}61        await self._send_status_update()62    async def _complete(self, name: str) -> None:63        status = self._status_map[name]64        if not isinstance(status, Running):65            raise RuntimeError('Can only complete from the "Running" status.')66        self._status_map = {**self._status_map, name: status.complete()}67        await self._send_status_update()68    async def _fail(self, name: str) -> None:69        status = self._status_map[name]70        if not isinstance(status, Running):71            raise RuntimeError('Can only fail from the "Running" status.')72        self._status_map = {**self._status_map, name: status.fail()}73        await self._send_status_update()74    async def _send_status_update(self) -> None:75        if self._status_stream is not None:76            await self._status_stream.send(self._status_map)77    async def __aenter__(self) -> ProgressManager:78        self._status_map = self._initial_status_map79        await self._send_status_update()80        return self81    async def __aexit__(82        self,83        exc_type: Optional[Type[BaseException]],84        exc_value: Optional[BaseException],85        traceback: Optional[TracebackType],86    ) -> None:...

task_state_prop.py

Source:task_state_prop.py

1# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.2# Copyright (C) NIWA & British Crown (Met Office) & Contributors.3#4# This program is free software: you can redistribute it and/or modify5# it under the terms of the GNU General Public License as published by6# the Free Software Foundation, either version 3 of the License, or7# (at your option) any later version.8#9# This program is distributed in the hope that it will be useful,10# but WITHOUT ANY WARRANTY; without even the implied warranty of11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the12# GNU General Public License for more details.13#14# You should have received a copy of the GNU General Public License15# along with this program.  If not, see <http://www.gnu.org/licenses/>.16"""Task state properties for display."""17from cylc.flow.task_state import (18    TASK_STATUS_WAITING,19    TASK_STATUS_PREPARING,20    TASK_STATUS_EXPIRED,21    TASK_STATUS_SUBMITTED,22    TASK_STATUS_SUBMIT_FAILED,23    TASK_STATUS_RUNNING,24    TASK_STATUS_SUCCEEDED,25    TASK_STATUS_FAILED26)27from colorama import Style, Fore, Back28_STATUS_MAP = {29    TASK_STATUS_WAITING: {30        "ascii_ctrl": Style.BRIGHT + Fore.CYAN + Back.RESET31    },32    TASK_STATUS_PREPARING: {33        "ascii_ctrl": Style.BRIGHT + Fore.GREEN + Back.RESET34    },35    TASK_STATUS_EXPIRED: {36        "ascii_ctrl": Style.BRIGHT + Fore.WHITE + Back.BLACK37    },38    TASK_STATUS_SUBMITTED: {39        "ascii_ctrl": Style.BRIGHT + Fore.YELLOW + Back.RESET40    },41    TASK_STATUS_SUBMIT_FAILED: {42        "ascii_ctrl": Style.BRIGHT + Fore.BLUE + Back.RESET43    },44    TASK_STATUS_RUNNING: {45        "ascii_ctrl": Style.BRIGHT + Fore.WHITE + Back.GREEN46    },47    TASK_STATUS_SUCCEEDED: {48        "ascii_ctrl": Style.NORMAL + Fore.BLACK + Back.RESET49    },50    TASK_STATUS_FAILED: {51        "ascii_ctrl": Style.BRIGHT + Fore.WHITE + Back.RED52    },53}54def extract_group_state(child_states, is_stopped=False):55    """Summarise child states as a group."""56    ordered_states = [57        TASK_STATUS_SUBMIT_FAILED,58        TASK_STATUS_FAILED,59        TASK_STATUS_EXPIRED,60        TASK_STATUS_RUNNING,61        TASK_STATUS_SUBMITTED,62        TASK_STATUS_PREPARING,63        TASK_STATUS_WAITING,64        TASK_STATUS_SUCCEEDED65    ]66    if is_stopped:67        ordered_states = [68            TASK_STATUS_SUBMIT_FAILED,69            TASK_STATUS_FAILED,70            TASK_STATUS_RUNNING,71            TASK_STATUS_SUBMITTED,72            TASK_STATUS_EXPIRED,73            TASK_STATUS_PREPARING,74            TASK_STATUS_SUCCEEDED,75            TASK_STATUS_WAITING76        ]77    for state in ordered_states:78        if state in child_states:79            return state80    return None81def get_status_prop(status, key, subst=None):82    """Return property for a task status."""83    if key == "ascii_ctrl" and subst is not None:84        return "%s%s\033[0m" % (_STATUS_MAP[status][key], subst)85    elif key == "ascii_ctrl":86        return "%s%s\033[0m" % (_STATUS_MAP[status][key], status)87    else:...

node_info.py

Source:node_info.py

1# -*- coding: utf-8 -*-2import json3class NodeInfo:4    def __init__(self, node_info):5        self._node_info = node_info6        self._status_map = {}7        for k, v in node_info.status.items():8            self._status_map[k] = v9            setattr(self, "status_{}".format(k), v)10        self._config_map = {}11        for k, v in node_info.config.items():12            if k not in self._config_map:13                self._config_map[k] = {}14            for k2, v2 in v.props.items():15                self._config_map[k][k2] = v216                setattr(self, "config_{}_{}".format(k, k2), v2)17    def json(self):18        ret = {19            "status": {},20            "config": {},21        }22        for k, v in self._status_map.items():23            ret['status'][k] = v24        for k, v in self._config_map.items():25            ret['config'][k] = v26        return ret27    def __str__(self):...

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.