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

How to use mock_monitor_task_finished method in avocado

Best Python code snippet using avocado_python

task-state-machine.py

Source:task-state-machine.py

...21    return true_or_false(-8)22def mock_check_task_start():23    # More success than failures, please24    return true_or_false(-6)25def mock_monitor_task_finished():26    # More failures than successes, please27    return true_or_false(5)28class Task:29    """Used here as a placeholder for an avocado.core.nrunner.Task."""30    def __init__(self, identification):31        self._identification = identification32class TaskInfo(Task):33    """Task with extra status information on its life-cycle.34    The equivalent of a StatusServer will contain this information35    in the real implementation."""36    def __init__(self, identification):37        super(TaskInfo, self).__init__(identification)38        self._status = None39        self._timeout = None40    @property41    def status(self):42        return self._status43    @status.setter44    def status(self, status):45        self._status = status46    @property47    def timeout(self):48        return self._timeout49    @timeout.setter50    def timeout(self, timeout):51        self._timeout = timeout52    def __repr__(self):53        if self._status is None:54            return '%s' % self._identification55        else:56            return '%s (%s)' % (self._identification,57                                self.status)58class TaskStateMachine:59    """Represents all phases that a task can go through its life."""60    def __init__(self, tasks):61        self._requested = tasks62        self._triaging = []63        self._ready = []64        self._started = []65        self._finished = []66        self._lock = asyncio.Lock()67    @property68    def requested(self):69        return self._requested70    @property71    def triaging(self):72        return self._triaging73    @property74    def ready(self):75        return self._ready76    @property77    def started(self):78        return self._started79    @property80    def finished(self):81        return self._finished82    @property83    def lock(self):84        return self._lock85    @property86    async def complete(self):87        async with self._lock:88            pending = any([self._requested, self._triaging,89                           self._ready, self._started])90        return not pending91    def __str__(self):92        headers = ("|_REQUESTED_|", "|_TRIAGING__|",93                   "|___READY___|", "|__STARTED__|",94                   "|______FINISHED_______|")95        data = itertools.zip_longest(self._requested, self._triaging, self._ready,96                                     self._started, self._finished, fillvalue="")97        matrix = [_ for _ in data]98        return tabular_output(matrix, headers)99async def bootstrap(lc):100    """Reads from requested, moves into triaging."""101    # fake some rate limiting102    if true_or_false(10):103        return104    try:105        async with lc.lock:106            task = lc.requested.pop()107            lc.triaging.append(task)108            debug('Moved Task %s: REQUESTED => TRIAGING' % task)109    except IndexError:110        debug('BOOTSTRAP: nothing to do')111        return112async def triage(lc):113    """Reads from triaging, moves into either: ready or finished."""114    await sleep_random()115    try:116        async with lc.lock:117            task = lc.triaging.pop()118    except IndexError:119        debug('TRIAGE done')120        return121    if mock_check_task_requirement():122        async with lc.lock:123            lc.ready.append(task)124            debug('Moving Task %s: TRIAGING => READY' % task)125    else:126        async with lc.lock:127            lc.finished.append(task)128            task.status = 'FAILED ON TRIAGE'129            debug('Moving Task %s: TRIAGING => FINISHED' % task)130async def start(lc):131    """Reads from ready, moves into either: started or finished."""132    await sleep_random()133    try:134        async with lc.lock:135            task = lc.ready.pop()136    except IndexError:137        debug('START: nothing to do')138        return139    # enforce a rate limit on the number of started (currently running) tasks.140    # this is a global limit, but the spawners can also be queried with regards141    # to their capacity to handle new tasks142    MAX_RUNNING_TASKS = 8143    async with lc.lock:144        if len(lc.started) >= MAX_RUNNING_TASKS:145            lc.ready.insert(0, task)146            task.status = 'WAITING'147            return148    # suppose we're starting the tasks149    if mock_check_task_start():150        async with lc.lock:151            task.status = None152            # Let's give each task 15 seconds from start time153            task.timeout = time.monotonic() + 15154            lc.started.append(task)155            debug('Moving Task %s: READY => STARTED' % task)156    else:157        async with lc.lock:158            lc.finished.append(task)159            task.status = 'FAILED ON START'160            debug('Moving Task %s: READY => FINISHED (ERRORED ON START)' % task)161async def monitor(lc):162    """Reads from started, moves into finished."""163    await sleep_random()164    try:165        async with lc.lock:166            task = lc.started.pop()167    except IndexError:168        debug('MONITOR: nothing to do')169        return170    if time.monotonic() > task.timeout:171        async with lc.lock:172            task.status = 'FAILED W/ TIMEOUT'173            lc.finished.append(task)174            debug('Moving Task %s: STARTED => FINISHED (FAILED ON TIMEOUT)' % task)175    elif mock_monitor_task_finished():176        async with lc.lock:177            lc.finished.append(task)178            debug('Moving Task %s: STARTED => FINISHED (COMPLETED AFTER STARTED)' % task)179    else:180        async with lc.lock:181            lc.started.insert(0, task)182        debug('Task %s: has not finished yet' % task)183def print_lc_status(lc):184    print("\033c", end="")185    print(str(lc))186async def worker(lc):187    """Pushes Tasks forward and makes them do something with their lives."""188    while True:189        complete = await lc.complete...

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.