How to use _status_map method in avocado

Best Python code snippet using avocado_python

_progress_manager.py

Source:_progress_manager.py Github

copy

Full Screen

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

Full Screen

Full Screen

task_state_prop.py

Source:task_state_prop.py Github

copy

Full Screen

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

Full Screen

Full Screen

node_info.py

Source:node_info.py Github

copy

Full Screen

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):...

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:

YouTube

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?

Helpful

NotHelpful