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

How to use __add_jmeter_components method in yandex-tank

Best Python code snippet using yandex-tank

plugin.py

Source:plugin.py

...62        self.core.add_artifact_file(self.jmeter_log, True)63        self.exclude_markers = set(64            filter((lambda marker: marker != ''),65                   self.get_option('exclude_markers', []).split(' ')))66        self.jmx = self.__add_jmeter_components(67            self.original_jmx, self.jtl_file, self._get_variables())68        self.core.add_artifact_file(self.jmx)69        jmeter_stderr_file = self.core.mkstemp(".log", "jmeter_stdout_stderr_")70        self.core.add_artifact_file(jmeter_stderr_file)71        self.jmeter_stderr = open(jmeter_stderr_file, 'w')72    def prepare_test(self):73        self.args = [74            self.jmeter_path, "-n", "-t", self.jmx, '-j', self.jmeter_log,75            '-Jjmeter.save.saveservice.default_delimiter=\\t',76            '-Jjmeter.save.saveservice.connect_time=true'77        ]78        self.args += splitstring(self.user_args)79        aggregator = None80        try:81            aggregator = self.core.get_plugin_of_type(AggregatorPlugin)82        except Exception as ex:83            logger.warning("No aggregator found: %s", ex)84        if aggregator:85            aggregator.reader = JMeterReader(self.jtl_file)86            aggregator.stats_reader = aggregator.reader.stats_reader87        try:88            console = self.core.get_plugin_of_type(ConsolePlugin)89        except Exception as ex:90            logger.debug("Console not found: %s", ex)91            console = None92        if console:93            widget = JMeterInfoWidget(self)94            console.add_info_widget(widget)95            if aggregator:96                aggregator.add_result_listener(widget)97    def start_test(self):98        logger.info(99            "Starting %s with arguments: %s", self.jmeter_path, self.args)100        try:101            self.jmeter_process = subprocess.Popen(102                self.args,103                executable=self.jmeter_path,104                preexec_fn=os.setsid,105                close_fds=True,106                stdout=self.jmeter_stderr,107                stderr=self.jmeter_stderr)108        except OSError:109            logger.debug(110                "Unable to start JMeter process. Args: %s, Executable: %s",111                self.args,112                self.jmeter_path,113                exc_info=True)114            raise RuntimeError(115                "Unable to access to JMeter executable file or it does not exist: %s"116                % self.jmeter_path)117        self.start_time = time.time()118    def is_test_finished(self):119        retcode = self.jmeter_process.poll()120        aggregator = self.core.get_plugin_of_type(AggregatorPlugin)121        if not aggregator.reader.jmeter_finished and retcode is not None:122            logger.info(123                "JMeter process finished with exit code: %s, waiting for aggregator",124                retcode)125            self.retries = 0126            aggregator.reader.jmeter_finished = True127            return -1128        elif aggregator.reader.jmeter_finished is True:129            if aggregator.reader.agg_finished:130                return retcode131            else:132                logger.info("Waiting for aggregator to finish")133                return -1134        else:135            return -1136    def end_test(self, retcode):137        if self.jmeter_process:138            logger.info(139                "Terminating jmeter process group with PID %s",140                self.jmeter_process.pid)141            try:142                os.killpg(self.jmeter_process.pid, signal.SIGTERM)143            except OSError as exc:144                logger.debug("Seems JMeter exited itself: %s", exc)145                # Utils.log_stdout_stderr(logger, self.jmeter_process.stdout, self.jmeter_process.stderr, "jmeter")146        if self.jmeter_stderr:147            self.jmeter_stderr.close()148        self.core.add_artifact_file(self.jmeter_log)149        return retcode150    def __add_jmeter_components(self, jmx, jtl, variables):151        """ Genius idea by Alexey Lavrenyuk """152        logger.debug("Original JMX: %s", os.path.realpath(jmx))153        with open(jmx, 'r') as src_jmx:154            source_lines = src_jmx.readlines()155        try:156            closing = source_lines.pop(-1)157            closing = source_lines.pop(-1) + closing158            closing = source_lines.pop(-1) + closing159            logger.debug("Closing statement: %s", closing)160        except Exception as exc:161            raise RuntimeError("Failed to find the end of JMX XML: %s" % exc)162        udv_tpl = resource_string(__name__, 'config/jmeter_var_template.xml')163        udv_set = []164        for var_name, var_value in variables.iteritems():...

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.