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

How to use __graceful_shutdown method in yandex-tank

Best Python code snippet using yandex-tank

consoleworker.py

Source:consoleworker.py

...201            sys.stderr.write(RealConsoleMarkup.RESET)202            sys.stderr.write(RealConsoleMarkup.TOTAL_RESET)203            self.core.release_lock()204            raise ex205    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(250                    "Caught KeyboardInterrupt again: %s",251                    traceback.format_exc(ex))252                self.log.info(253                    "User insists on exiting, aborting graceful shutdown...")254                retcode = 1255        except Exception as ex:256            self.log.info("Exception: %s", traceback.format_exc(ex))257            sys.stderr.write(RealConsoleMarkup.RED)258            self.log.error("%s", ex)259            sys.stderr.write(RealConsoleMarkup.RESET)260            sys.stderr.write(RealConsoleMarkup.TOTAL_RESET)261            retcode = self.__graceful_shutdown()262            self.core.release_lock()263        finally:264            self.core.close()265        self.log.info("Done performing test with code %s", retcode)266        return retcode267class DevNullOpts:268    def __init__(self):269        pass270    log = "/dev/null"271class CompletionHelperOptionParser(OptionParser):272    def __init__(self):273        OptionParser.__init__(self, add_help_option=False)274        self.add_option(275            '--bash-switches-list',...

plugin.py

Source:plugin.py

...136        else:137            return -1138    def end_test(self, retcode):139        if self.process:140            gracefully_shutdown = self.__graceful_shutdown()141            if not gracefully_shutdown:142                self.__kill_jmeter()143        if self.process_stderr:144            self.process_stderr.close()145        self.core.add_artifact_file(self.jmeter_log)146        self.reader.close()147        return retcode148    def __discover_jmeter_udp_port(self):149        """Searching for line in jmeter.log such as150        Waiting for possible shutdown message on port 4445151        """152        r = re.compile(self.DISCOVER_PORT_PATTERN)153        with open(self.process_stderr.name, 'r') as f:154            cnt = 0155            while self.process.pid and cnt < 10:156                line = f.readline()157                m = r.match(line)158                if m is None:159                    cnt += 1160                    time.sleep(1)161                else:162                    port = int(m.group('port'))163                    return port164            else:165                logger.warning('JMeter UDP port wasn\'t discovered')166                return None167    def __kill_jmeter(self):168        logger.info(169            "Terminating jmeter process group with PID %s",170            self.process.pid)171        try:172            os.killpg(self.process.pid, signal.SIGTERM)173        except OSError as exc:174            logger.debug("Seems JMeter exited itself: %s", exc)175            # Utils.log_stdout_stderr(logger, self.process.stdout, self.process.stderr, "jmeter")176    def __add_jmeter_components(self, jmx, jtl, variables):177        """ Genius idea by Alexey Lavrenyuk """178        logger.debug("Original JMX: %s", os.path.realpath(jmx))179        with open(jmx, 'r') as src_jmx:180            source_lines = src_jmx.readlines()181        try:182            # In new Jmeter version (3.2 as example) WorkBench's plugin checkbox enabled by default183            # It totally crashes Yandex tank injection and raises XML Parse Exception184            closing = source_lines.pop(-1)185            if "WorkBenchGui" in source_lines[-5]:186                logger.info("WorkBench checkbox enabled...bypassing")187                last_string_count = 6188            else:189                last_string_count = 2190            while last_string_count > 0:191                closing = source_lines.pop(-1) + closing192                last_string_count -= 1193            logger.debug("Closing statement: %s", closing)194        except Exception as exc:195            raise RuntimeError("Failed to find the end of JMX XML: %s" % exc)196        udv_tpl = resource_string(__name__, 'config/jmeter_var_template.xml').decode('utf8')197        udv_set = []198        for var_name, var_value in variables.items():199            udv_set.append(udv_tpl % (var_name, var_name, var_value))200        udv = "\n".join(udv_set)201        if self.jmeter_ver >= 2.13:202            save_connect = '<connectTime>true</connectTime>'203        else:204            save_connect = ''205        if self.ext_log in ['errors', 'all']:206            level_map = {'errors': 'true', 'all': 'false'}207            tpl_resource = 'jmeter_writer_ext.xml'208            tpl_args = {209                'jtl': self.jtl_file,210                'udv': udv,211                'ext_log': self.ext_log_file,212                'ext_level': level_map[self.ext_log],213                'save_connect': save_connect214            }215        else:216            tpl_resource = 'jmeter_writer.xml'217            tpl_args = {218                'jtl': self.jtl_file,219                'udv': udv,220                'save_connect': save_connect221            }222        tpl = resource_string(__name__, 'config/' + tpl_resource).decode('utf8')223        try:224            new_jmx = self.core.mkstemp(225                '.jmx', 'modified_', os.path.dirname(os.path.realpath(jmx)))226        except OSError as exc:227            logger.debug("Can't create modified jmx near original: %s", exc)228            new_jmx = self.core.mkstemp('.jmx', 'modified_')229        logger.debug("Modified JMX: %s", new_jmx)230        with open(new_jmx, "w") as fh:231            fh.write(''.join(source_lines))232            fh.write(tpl % tpl_args)233            fh.write(closing)234        return new_jmx235    def __graceful_shutdown(self):236        if self.jmeter_udp_port is None:237            return False238        shutdown_test_started = time.time()239        while time.time() - shutdown_test_started < self.shutdown_timeout:240            self.__send_udp_message(self.SHUTDOWN_TEST)241            if self.process.poll() is not None:242                return True243            else:244                time.sleep(1)245        self.log.info('Graceful shutdown failed after %s' % str(time.time() - shutdown_test_started))246        stop_test_started = time.time()247        while time.time() - stop_test_started < self.shutdown_timeout:248            self.__send_udp_message(self.STOP_TEST_NOW)249            if self.process.poll() is not None:...

apiworker.py

Source:apiworker.py

...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):...

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.