KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use __setup_taskset method in yandex-tank

Best Python code snippet using yandex-tank

tankcore.py

Source:tankcore.py

...204        self.publish("core", "stage", "configure")205        logger.info("Configuring plugins...")206        self.taskset_affinity = self.get_option(self.SECTION, 'affinity')207        if self.taskset_affinity:208            self.__setup_taskset(self.taskset_affinity, pid=os.getpid())209        for plugin in self.plugins.values():210            if not self.interrupted.is_set():211                logger.debug("Configuring %s", plugin)212                plugin.configure()213                if isinstance(plugin, MonitoringDataListener):214                    self.monitoring_data_listeners.append(plugin)215    def plugins_prepare_test(self):216        """ Call prepare_test() on all plugins        """217        logger.info("Preparing test...")218        self.publish("core", "stage", "prepare")219        for plugin in self.plugins.values():220            if not self.interrupted.is_set():221                logger.debug("Preparing %s", plugin)222                plugin.prepare_test()223    def plugins_start_test(self):224        """        Call start_test() on all plugins        """225        if not self.interrupted.is_set():226            logger.info("Starting test...")227            self.publish("core", "stage", "start")228            self.job.aggregator.start_test()229            for plugin_name, plugin in self.plugins.items():230                logger.debug("Starting %s", plugin)231                start_time = time.time()232                plugin.start_test()233                logger.info("Plugin {0:s} required {1:f} seconds to start".format(plugin_name,234                                                                                  time.time() - start_time))235            self.publish('generator', 'test_start', self.job.generator_plugin.start_time)236    def wait_for_finish(self):237        """238        Call is_test_finished() on all plugins 'till one of them initiates exit239        """240        if not self.interrupted.is_set():241            logger.info("Waiting for test to finish...")242            logger.info('Artifacts dir: {dir}'.format(dir=self.artifacts_dir))243            self.publish("core", "stage", "shoot")244            if not self.plugins:245                raise RuntimeError("It's strange: we have no plugins loaded...")246        while not self.interrupted.is_set():247            begin_time = time.time()248            aggr_retcode = self.job.aggregator.is_test_finished()249            if aggr_retcode >= 0:250                return aggr_retcode251            for plugin_name, plugin in self.plugins.items():252                logger.debug("Polling %s", plugin)253                try:254                    retcode = plugin.is_test_finished()255                    if retcode >= 0:256                        return retcode257                except Exception:258                    logger.warning('Plugin {} failed:'.format(plugin_name), exc_info=True)259                    if isinstance(plugin, GeneratorPlugin):260                        return RetCode.ERROR261                    else:262                        logger.warning('Disabling plugin {}'.format(plugin_name))263                        plugin.is_test_finished = lambda: RetCode.CONTINUE264            end_time = time.time()265            diff = end_time - begin_time266            logger.debug("Polling took %s", diff)267            logger.debug("Tank status: %s", json.dumps(self.info.get_info_dict()))268            # screen refresh every 0.5 s269            if diff < 0.5:270                time.sleep(0.5 - diff)271        return 1272    def plugins_end_test(self, retcode):273        """        Call end_test() on all plugins        """274        logger.info("Finishing test...")275        self.publish("core", "stage", "end")276        self.publish('generator', 'test_end', time.time())277        logger.info("Stopping load generator and aggregator")278        retcode = self.job.aggregator.end_test(retcode)279        logger.debug("RC after: %s", retcode)280        logger.info('Stopping monitoring')281        for plugin in self.job.monitoring_plugins:282            logger.info('Stopping %s', plugin)283            retcode = plugin.end_test(retcode) or retcode284            logger.info('RC after: %s', retcode)285        for plugin in [p for p in self.plugins.values() if286                       p is not self.job.generator_plugin and p not in self.job.monitoring_plugins]:287            logger.debug("Finalize %s", plugin)288            try:289                logger.debug("RC before: %s", retcode)290                retcode = plugin.end_test(retcode)291                logger.debug("RC after: %s", retcode)292            except Exception:  # FIXME too broad exception clause293                logger.error("Failed finishing plugin %s", plugin, exc_info=True)294                if not retcode:295                    retcode = 1296        return retcode297    def plugins_post_process(self, retcode):298        """299        Call post_process() on all plugins300        """301        logger.info("Post-processing test...")302        self.publish("core", "stage", "post_process")303        for plugin in self.plugins.values():304            logger.debug("Post-process %s", plugin)305            try:306                logger.debug("RC before: %s", retcode)307                retcode = plugin.post_process(retcode)308                logger.debug("RC after: %s", retcode)309            except Exception:  # FIXME too broad exception clause310                logger.error("Failed post-processing plugin %s", plugin, exc_info=True)311                if not retcode:312                    retcode = 1313        return retcode314    def publish_monitoring_data(self, data):315        """sends pending data set to listeners"""316        for plugin in self.monitoring_data_listeners:317            # deep copy to ensure each listener gets it's own copy318            try:319                plugin.monitoring_data(copy.deepcopy(data))320            except Exception:321                logger.error("Plugin failed to process monitoring data", exc_info=True)322    def __setup_taskset(self, affinity, pid=None, args=None):323        """ if pid specified: set process w/ pid `pid` CPU affinity to specified `affinity` core(s)324            if args specified: modify list of args for Popen to start w/ taskset w/ affinity `affinity`325        """326        self.taskset_path = self.get_option(self.SECTION, 'taskset_path')327        if args:328            return [self.taskset_path, '-c', affinity] + args329        if pid:330            args = "%s -pc %s %s" % (self.taskset_path, affinity, pid)331            retcode, stdout, stderr = execute(args, shell=True, poll_period=0.1, catch_out=True)332            logger.debug('taskset for pid %s stdout: %s', pid, stdout)333            if retcode == 0:334                logger.info("Enabled taskset for pid %s with affinity %s", str(pid), affinity)335            else:336                logger.debug('Taskset setup failed w/ retcode :%s', retcode)...

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.