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

How to use plugins_end_test method in yandex-tank

Best Python code snippet using yandex-tank

consoleworker.py

Source:consoleworker.py

...205    def __graceful_shutdown(self):206        """ call shutdown routines """207        retcode = 1208        self.log.info("Trying to shutdown gracefully...")209        retcode = self.core.plugins_end_test(retcode)210        retcode = self.core.plugins_post_process(retcode)211        self.log.info("Done graceful shutdown")212        return retcode213    def perform_test(self):214        """215        Run the test sequence via Tank Core216        """217        self.log.info("Performing test")218        retcode = 1219        try:220            self.core.plugins_configure()221            self.core.plugins_prepare_test()222            if self.scheduled_start:223                self.log.info(224                    "Waiting scheduled time: %s...", self.scheduled_start)225                while datetime.datetime.now() < self.scheduled_start:226                    self.log.debug(227                        "Not yet: %s < %s",228                        datetime.datetime.now(), self.scheduled_start)229                    time.sleep(1)230                self.log.info("Time has come: %s", datetime.datetime.now())231            if self.options.manual_start:232                raw_input("Press Enter key to start test:")233            self.core.plugins_start_test()234            retcode = self.core.wait_for_finish()235            retcode = self.core.plugins_end_test(retcode)236            retcode = self.core.plugins_post_process(retcode)237        except KeyboardInterrupt as ex:238            sys.stdout.write(RealConsoleMarkup.YELLOW)239            self.log.info(240                "Do not press Ctrl+C again, the test will be broken otherwise")241            sys.stdout.write(RealConsoleMarkup.RESET)242            sys.stdout.write(RealConsoleMarkup.TOTAL_RESET)243            self.signal_count += 1244            self.log.debug(245                "Caught KeyboardInterrupt: %s", traceback.format_exc(ex))246            try:247                retcode = self.__graceful_shutdown()248            except KeyboardInterrupt as ex:249                self.log.debug(...

apiworker.py

Source:apiworker.py

...87                    "Manual start option specified, waiting for user actions")88                raw_input("Press Enter key to start test")89            self.core.plugins_start_test()90            retcode = self.core.wait_for_finish()91            retcode = self.core.plugins_end_test(retcode)92            retcode = self.core.plugins_post_process(retcode)93        except KeyboardInterrupt as ex:94            self.log.info(95                "Do not press Ctrl+C again, the test will be broken otherwise")96            self.log.debug(97                "Caught KeyboardInterrupt: %s", traceback.format_exc(ex))98            try:99                retcode = self.__graceful_shutdown()100            except KeyboardInterrupt as ex:101                self.log.debug(102                    "Caught KeyboardInterrupt again: %s",103                    traceback.format_exc(ex))104                self.log.info(105                    "User insists on exiting, aborting graceful shutdown...")106                retcode = 1107        except Exception as ex:108            self.log.info("Exception: %s", traceback.format_exc(ex))109            self.log.error("%s", ex)110            retcode = self.__graceful_shutdown()111            self.core.release_lock()112        self.log.info("Done performing test with code %s", retcode)113        return retcode114    def get_default_configs(self):115        """ returns default configs list, from /etc, home dir and package_data"""116        # initialize basic defaults117        configs = [resource_filename(__name__, 'config/00-base.ini')]118        try:119            conf_files = sorted(os.listdir(self.baseconfigs_location))120            for filename in conf_files:121                if fnmatch.fnmatch(filename, '*.ini'):122                    configs += [123                        os.path.realpath(124                            self.baseconfigs_location + os.sep + filename)125                    ]126        except OSError:127            self.log.warn(128                self.baseconfigs_location +129                ' is not accessible to get configs list')130        configs += [os.path.expanduser('~/.yandex-tank')]131        return configs132    def __graceful_shutdown(self):133        """ call shutdown routines """134        retcode = 1135        self.log.info("Trying to shutdown gracefully...")136        retcode = self.core.plugins_end_test(retcode)137        retcode = self.core.plugins_post_process(retcode)138        self.log.info("Done graceful shutdown")139        return retcode140class SingleLevelFilter(logging.Filter):141    """Exclude or approve one msg type at a time.    """142    def __init__(self, passlevel, reject):143        logging.Filter.__init__(self)144        self.passlevel = passlevel145        self.reject = reject146    def filter(self, record):147        if self.reject:148            return record.levelno != self.passlevel149        else:150            return record.levelno == self.passlevel

test_tankcore.py

Source:test_tankcore.py

...131def test_start_test(config):132    core = TankCore(configs=[config])133    core.plugins_prepare_test()134    core.plugins_start_test()135    core.plugins_end_test(1)136@pytest.mark.parametrize('options, expected', [137    (138        ['meta.task=LOAD-204',139         'phantom.ammofile = air-tickets-search-ammo.log',140         'meta.component = air_tickets_search [imbalance]',141         'meta.jenkinsjob = https://jenkins-load.yandex-team.ru/job/air_tickets_search/'],142        [{'uploader': {'package': 'yandextank.plugins.DataUploader', 'task': 'LOAD-204'}},143         {'phantom': {'package': 'yandextank.plugins.Phantom', 'ammofile': 'air-tickets-search-ammo.log'}},144         {'uploader': {'package': 'yandextank.plugins.DataUploader', 'component': 'air_tickets_search [imbalance]'}},145         {'uploader': {'package': 'yandextank.plugins.DataUploader',146                       'meta': {'jenkinsjob': 'https://jenkins-load.yandex-team.ru/job/air_tickets_search/'}}}]147    ),148    #     with converting/type-casting149    (...

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.