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

How to use mock_check_task_start method in avocado

Best Python code snippet using avocado_python

task-state-machine.py

Source:task-state-machine.py

...18    return random.choice(choices)19def mock_check_task_requirement():20    # More success than failures, please21    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()...

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.