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

How to use _may_run method in Molotov

Best Python code snippet using molotov_python

jobs.py

Source:jobs.py

...111                self._abort()112            else:113                raise JobNotActive()114    def run(self):115        if not self._may_run():116            return117        try:118            self._run()119        except exception.ActionStopped:120            self._abort_completed()121        except Exception as e:122            self._run_failed(e)123        else:124            self._run_completed()125        finally:126            self._autodelete_if_required()127    def _may_run(self):128        """129        Check if the job should enter the running state.  Allowed origin states130        are aborted and pending.  If a pending job had been aborted we quietly131        refuse to run it.  The common case is to move a job from pending to132        running.133        """134        with self._status_lock:135            if self.status == STATUS.ABORTED:136                logging.debug('Refusing to run aborted job %r', self._id)137                return False138            if self.status != STATUS.PENDING:139                raise RuntimeError('Attempted to run job %r from state %r' %140                                   (self._id, self.status))141            self._status = STATUS.RUNNING...

worker.py

Source:worker.py

...43        finally:44            self.teardown()45            self.results["WORKER"] -= 146        return res47    def _may_run(self):48        if is_stopped():49            return False50        if _now() - self.worker_start > self.args.duration:51            return False52        if self._exhausted:53            return False54        if self.results["REACHED"] == 1:55            return False56        if self.args.max_runs and self.count > self.args.max_runs:57            return False58        return True59    async def setup(self):60        if self._setup is None:61            return {}62        try:63            options = await self._setup(self.wid, self.args)64        except Exception as e:65            self.console.print_error(e)66            raise FixtureError(str(e))67        if options is None:68            options = {}69        elif not isinstance(options, dict):70            msg = "The setup function needs to return a dict"71            self.console.print(msg)72            raise FixtureError(msg)73        return options74    async def session_setup(self, session):75        if self._session_setup is None:76            return77        try:78            await self._session_setup(self.wid, session)79        except Exception as e:80            self.console.print_error(e)81            raise FixtureError(str(e))82    async def session_teardown(self, session):83        if self._session_teardown is None:84            return85        try:86            await self._session_teardown(self.wid, session)87        except Exception as e:88            # we can't stop the teardown process89            self.console.print_error(e)90    async def _run(self):91        verbose = self.args.verbose92        exception = self.args.exception93        if self.args.single_mode:94            single = get_scenario(self.args.single_mode)95        elif self.args.single_run:96            single = next_scenario()97        else:98            single = None99        self.count = 1100        self.worker_start = _now()101        try:102            options = await self.setup()103        except FixtureError as e:104            self.results["SETUP_FAILED"] += 1105            stop(why=e)106            return107        async with get_session(108            self.loop, self.console, verbose, self.statsd, **options109        ) as session:110            get_context(session).args = self.args111            get_context(session).worker_id = self.wid112            try:113                await self.session_setup(session)114            except FixtureError as e:115                self.results["SESSION_SETUP_FAILED"] += 1116                stop(why=e)117                return118            while self._may_run():119                step_start = _now()120                get_context(session).step = self.count121                result = await self.step(self.count, session, scenario=single)122                if result == 1:123                    self.results["OK"] += 1124                    self.results["MINUTE_OK"] += 1125                elif result != 0:126                    self.results["FAILED"] += 1127                    self.results["MINUTE_FAILED"] += 1128                    if exception:129                        stop(why=result)130                if not is_stopped() and self._reached_tolerance(step_start):131                    stop()132                    cancellable_sleep.cancel_all()...

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.