Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use add_result_listener method in yandex-tank

Best Python code snippet using yandex-tank

dynamic_scheduler.py

Source:dynamic_scheduler.py

...48                                           executors=self.config.EXECUTORS, job_defaults=self.config.JOB_DEFAULTS)49        self.listener = Listener(self.scheduler, config_class)50        self.scheduler.add_listener(self.listener.add_job_listener, EVENT_JOB_ADDED)51        self.db_pool = PoolDB(self.config.DEFAULT_DB_URL)52    def add_result_listener(self):53        if self.config.IS_DYNAMIC_UPDATE_CRON:  # å¼å¯ä¼èæ¶ éä½æ§è½54            self.scheduler.add_listener(self.listener.execute_job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)55        return self56    def add_job(self, func, trigger=None, args=None, kwargs=None, id=None, name=None,57                misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,58                next_run_time=undefined, jobstore='default', executor='default',59                replace_existing=False, **trigger_args):60        self.scheduler.add_job(func, trigger, args, kwargs, id, name,61                               misfire_grace_time, coalesce, max_instances,62                               next_run_time, jobstore, executor,63                               replace_existing, **trigger_args)64    def add_job_with_default(self, func, cron_str=None,trigger=None, args=None, kwargs=None, id=None, name=None,65                             misfire_grace_time=undefined, coalesce=True, max_instances=undefined,66                             next_run_time=undefined, jobstore='default', executor='default',67                             replace_existing=True, **trigger_args):68        """69        é»è®¤åªéä¼ å¥ funcãidãcronè¡¨è¾¾å¼70        å¢å ä»»å¡æ¶è®¾å®ä¸äºé»è®¤éé¡¹ï¼é¿åéå¤code, trigger é»è®¤ä¸º cron71        """72        trigger_ = self.config.DEFAULT_TRIGGER if trigger is None else trigger73        trigger_args_ = trigger_args74        if cron_str:75            trigger_args_.update(self.listener.translate_expression(cron_str))76        self.add_job(func, trigger_, args, kwargs, id, name,77                     misfire_grace_time, coalesce, max_instances,78                     next_run_time, jobstore, executor,79                     replace_existing, **trigger_args_)80    def start(self, *args, **kwargs):81        existed_jobs = self.scheduler.get_jobs(jobstore='default')82        if existed_jobs:83            self.scheduler.start(*args, **kwargs)84    def clear_history(self):85        self.db_pool.execute(f'delete from {self.config.SCHEDULER_SCHEMA}.{self.config.SCHEDULER_TABLE}')86    def set_cron_task(self, func):87        """88        1. ç»é»è®¤ cronè¡¨ä¸å®æ¶ä»»å¡ è¿è¡å¯å¨89        2. è®¾ç½® ä¸åçfunc çå®æ¶ä»»å¡ï¼éè¿job_idå³èï¼job_idå¯è®¾ç½®ä¸ºfunc_name90        # ä¾èµself.insert_cron_table æ¹æ³ï¼åæ§è¡å®ç»æ¯ä¸ªæ¹æ³è®¾å®åå§å¼;91        # ç°å¨ä¸ä¾èµäºï¼æ·»å ä¸åå¨çä»»å¡ä¼èªå¨åå¥cronä¸92        """93        cron_task = self.db_pool[f'select * from {self.config_class.SCHEMA_TABLE} where state = 1']94        if not cron_task.empty:95            for index, row in cron_task.iterrows():96                self.add_job_with_default(func, self.listener.translate_expression(row['expression']), id=row['job_id'])97        self.add_result_listener()98        # self.start()99    def insert_cron_table(self, func_list, expression_list):100        """æ¹éä¼ å¥æ¹æ³åè¡¨ãè¡¨è¾¾å¼åè¡¨ï¼åæå¯æ¹ä¸ºdict"""101        value_str = [f"('{func.__name__}', '{express}')" for func, express in zip(func_list, expression_list)]102        value_str = ", ".join(value_str)103        insert_sql = f"insert into {self.config.SCHEMA_TABLE} (job_id, expression) values {value_str}"104        self.db_pool.insert(insert_sql)105        logger.success("åå¥cronè¡¨æå")106class StaticSchedulerProxy(DynamicSchedulerProxy):107    def __init__(self, config_class):108        super(StaticSchedulerProxy, self).__init__(config_class)109        self.scheduler.remove_listener(self.listener.add_job_listener)110        self.listener = Listener(self.scheduler, self.config)111        self.listener.IS_DYNAMIC_UPDATE_CRON = False112        self.scheduler.add_listener(self.listener.add_job_listener, EVENT_JOB_ADDED)113if __name__ == '__main__':114    def task():115        # print(datetime.datetime.now())116        # time.sleep(10)117        print('start time: ', datetime.datetime.now(), os.getpid(), threading.current_thread().name)118        return datetime.datetime.now()119    class Config(SchedulerConfig):120        DEFAULT_DB_URL = 'postgresql+psycopg2://postgres:123456@127.0.0.1:5432/postgres?utf-8'121        SCHEDULER_SCHEMA = 'public'122    s = DynamicSchedulerProxy(config_class=Config)123    s.clear_history()124    s.add_job_with_default(func=task, cron_str='* * */5', id='task')125    s.add_result_listener()...

main.py

Source:main.py

...90        self.main_layout = QVBoxLayout(self)91        self.main_layout.addWidget(self.ribbon)92        self.main_layout.addWidget(self.view)93        self.model.add_source_listener(self.view.set_left_image)94        self.model.add_result_listener(self.view.set_right_image)95        self.model.add_loading_listener(lambda loading: upscale_button.setEnabled(not loading))96        self.model.add_result_listener(lambda result: save_button.setEnabled(result is not None))97        self.model.add_source_listener(lambda source: upscale_button.setEnabled(source is not None))98    def dragEnterEvent(self, e):99        if e.mimeData().hasImage() or e.mimeData().hasUrls():100            e.accept()101        else:102            e.ignore()103            104    def dropEvent(self, e):105        if e.mimeData().hasImage():106            qtimage = e.mimeData().imageData()107            image = Image.fromqimage(qtimage)108            self.display_image(image)109        elif e.mimeData().urls()[0].scheme() == "file":110            url = e.mimeData().urls()[0].toString()...

game.py

Source:game.py

...13    def __init__(self, player1: Player, player2: Player) -> None:14        self.player1 = player115        self.player2 = player216        pass17    def add_result_listener(self, result_listener: Callable[[Result, Player], None]):18        self.result_listeners.append(result_listener)19    def move(self, player: Player, move: Move):20        if player == self.player1:21            if self.player1Move != None:22                raise Exception("Player 1 already made their move")23            self.player1Move = move24        elif player == self.player2:25            if self.player2Move != None:26                raise Exception("Player 2 already made their move")27            self.player2Move = move28        else:29            raise Exception(f'Unknown player: {player.name}')30        if self.player1Move != None and self.player2Move != None:31            self.determine_result()...

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.