Best Python code snippet using avocado_python
runner.py
Source:runner.py  
...83            self._failed = True84            TEST_LOG.error("RUNNER: Failed to read queue: %s", details)85            return None86    @property87    def early_status(self):88        """89        Get early status90        """91        if self._early_status:92            return self._early_status93        else:94            queue = []95            while not self.queue.empty():96                msg = self._get_msg_from_queue()97                if msg is None:98                    break99                if "early_status" in msg:100                    self._early_status = msg101                    for _ in queue:     # Return all unprocessed messages back102                        self.queue.put(_)103                    return msg104                else:   # Not an early_status message105                    queue.append(msg)106    def __getattribute__(self, name):107        # Update state before returning the value108        if name in ("status", "interrupt"):109            self._tick()110        return super(TestStatus, self).__getattribute__(name)111    def wait_for_early_status(self, proc, timeout):112        """113        Wait until early_status is obtained114        :param proc: test process115        :param timeout: timeout for early_state116        :raise exceptions.TestError: On timeout/error117        """118        step = 0.01119        end = time.time() + timeout120        while not self.early_status:121            if not proc.is_alive():122                if not self.early_status:123                    raise exceptions.TestError("Process died before it pushed "124                                               "early test_status.")125            if time.time() > end and not self.early_status:...check_early.py
Source:check_early.py  
1import datetime2import traceback3from common import get_now_utc_naive, get_now_local_naive, clear_log_file4from token_early import TokenEarly5import sql_manager6def main():7    """8    check_early.py9    Looks for tokens in db with10      - Null early_status11      - recorded_date after 10mins ago sorted ascending12    Run every 5 mins at specific minutes 5,10,15... on rpi13    Inserts14        - early_status          :   ok or bad15        - early_status_cause    :   a000 -> a02216        - early_monitor_time    :   at updated time in utc17    """18    clear_log_file("check_early")19    local, utc = get_now_local_naive(), get_now_utc_naive()20    print(f"check: main @ local:{local} utc:{utc}")21    tokens = get_recent_tokens()22    for i, token in enumerate(tokens):23        print(f"start monitoring token {i+1} of {len(tokens)} pair:{token.pair}")24        try:25            token.monitor()26        except:27            print(28                "exception found in monitor. Script will continue monitoring. "29                "The error is:"30            )31            traceback.print_exc()32    local, utc = get_now_local_naive(), get_now_utc_naive()33    print(f"finished monitoring token @ local:{local} utc:{utc}\n")34def get_recent_tokens():35    """36    Looks for tokens in db with37      - Null early_status38      - recorded_date after 10mins ago sorted ascending39    Returns40        tokens (list)   :   list of TokenEarly, each yet to be monitored for ok/bad41    """42    now = get_now_utc_naive()43    dt_hour_ago = now - datetime.timedelta(minutes=10)44    start_str = dt_hour_ago.strftime("%Y-%m-%d %H:%M:%S")45    stmt = (46        "SELECT main_token, pair FROM tokens "47        f"WHERE recorder_time > '{start_str}' "48        "AND early_status IS NULL "49        "ORDER BY recorder_time ASC "50        # "LIMIT 50"51        ";"52    )53    rows = sql_manager.Select(stmt, error="no get_recent_tokens")54    tokens = []55    if rows is not None:56        for main_token, pair in rows:57            tokens.append(TokenEarly(main_token, pair))58    return tokens...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
