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

How to use get_progress_logger method in yandex-tank

Best Python code snippet using yandex-tank

benchmark_pipeline_interval_trenscend.py

Source:benchmark_pipeline_interval_trenscend.py

...28            self.list_of_parameter_to_modified_value_dictionary = self.generate_parameter_to_modified_value_dictionary_for_each_pipeline(base_nameserver_port=self.base_nameserver_port, max_tuning_interval=self.max_tuning_interval, number_of_repeat_for_each_pipeline=self.number_of_repeat_for_each_pipeline)29            if self.run_multiprocessing:30                self.start_multiprocessing()31    def do_a_run_of_pipeline(self, parameter_to_modified_value_dictionary, file_path_for_saving_result, do_multiprocessing_dry_run=False):32        progress_logger = get_progress_logger()33        progress_logger.info('start doing one run of pipeline: ' + file_path_for_saving_result)34        progress_logger.info('the param dict for ' + file_path_for_saving_result + ' is: ' + str(parameter_to_modified_value_dictionary))35        print('start running pipeline: ')36        print(parameter_to_modified_value_dictionary)37        if not do_multiprocessing_dry_run:38            pipeLineTuningIntervalTranscend = PipeLineTuningIntervalTranscend(**parameter_to_modified_value_dictionary)39            AnalyzerPipeLineAutoForecastVA.save_analyzer_result_in_a_memory_efficient_way(pipeLineTuningIntervalTranscend.pipeLineAutoForecastVA, file_name=file_path_for_saving_result)40        progress_logger.info('end doing one run of pipeline: ' + file_path_for_saving_result)41        print('completed: ' + file_path_for_saving_result)42        print(parameter_to_modified_value_dictionary)43    def start_multiprocessing(self):44        progress_logger = get_progress_logger()45        progress_logger.info('starting multiprocessing')46        multiprocess_manager = ProcessLoom(max_runner_cap=self.max_runner_cap)47        for parameter_to_modified_value_dictionary in self.list_of_parameter_to_modified_value_dictionary:48            md5_code = ComputeCharacterizationCodeAndMd5Code.get_class_MD5_code(target_class=PipeLineTuningIntervalTranscend, parameter_to_modified_value_dictionary=parameter_to_modified_value_dictionary, store=True, md5_to_parameter_to_value_mapping_file_path=self.md5_to_parameter_to_value_mapping_file_path)49            file_path_for_saving_result = 'results/'+md5_code50            multiprocess_manager.add_function(self.do_a_run_of_pipeline, kwargs={'parameter_to_modified_value_dictionary': parameter_to_modified_value_dictionary, 'file_path_for_saving_result': file_path_for_saving_result, 'do_multiprocessing_dry_run': self.do_multiprocessing_dry_run}, key=md5_code)51        self.output = multiprocess_manager.execute()52    @staticmethod53    def generate_parameter_to_modified_value_dictionary_for_each_pipeline(base_nameserver_port=65100, max_tuning_interval=999999999, number_of_repeat_for_each_pipeline=5):54        '''Generate a list of kwargs, one for each pipeline'''55        # tuning interval56        list_of_tuning_interval_dict = []57        for interval in [5]:58            parameter_to_modified_value_dictionary = {}...

benchmark_pipeline_autova.py

Source:benchmark_pipeline_autova.py

...27        if not self.lazy:28            if self.run_multiprocessing:29                self.start_multiprocessing()30    def do_a_run_of_pipeline(self, parameter_to_modified_value_dictionary, file_path_for_saving_result, do_multiprocessing_dry_run=False):31        progress_logger = get_progress_logger(logger_name=self.logger_name, log_output_file_path=os.path.join(self.folder_name_to_save_results, self.log_output_file_name))32        progress_logger.info('start doing one run of pipeline: ' + file_path_for_saving_result)33        progress_logger.info('the param dict for ' + file_path_for_saving_result + ' is: ' + str(parameter_to_modified_value_dictionary))34        print('start running pipeline: ')35        print(parameter_to_modified_value_dictionary)36        timerDefault = TimerDefault(time_unit='hour')37        timerDefault.start()38        if not do_multiprocessing_dry_run:39            pipeLineTuningInterval = PipeLineTuningInterval(**parameter_to_modified_value_dictionary)40            AnalyzerPipeLineAutoForecastVA.save_analyzer_result_in_a_memory_efficient_way(pipeLineTuningInterval.pipeLineAutoForecastVA, file_name=file_path_for_saving_result)41        timerDefault.end()42        progress_logger.info('end doing one run of pipeline: ' + file_path_for_saving_result + ' ' + 'duration(hours): ' + str(timerDefault.get_duration()))43        print('completed: ' + file_path_for_saving_result)44        print(parameter_to_modified_value_dictionary)45    def start_multiprocessing(self):46        # progress_logger = get_progress_logger(logger_name=self.logger_name, log_output_file_path=os.path.join(self.folder_name_to_save_results, self.log_output_file_name))47        # progress_logger.info('starting multiprocessing')  # cannot do logging untill creates the folder48        multiprocess_manager = ProcessLoom(max_runner_cap=self.max_runner_cap)49        for parameter_to_modified_value_dictionary in self.list_of_input_dictionary:50            md5_code = ComputeCharacterizationCodeAndMd5Code.get_class_MD5_code(target_class=PipeLineTuningInterval, parameter_to_modified_value_dictionary=parameter_to_modified_value_dictionary, store=True, md5_to_parameter_to_value_mapping_file_path=os.path.join(self.folder_name_to_save_results, self.md5_to_parameter_to_value_mapping_file_name))51            file_path_for_saving_result = self.folder_name_to_save_results + '/' + md5_code + '.dat'52            multiprocess_manager.add_function(self.do_a_run_of_pipeline, kwargs={'parameter_to_modified_value_dictionary': parameter_to_modified_value_dictionary, 'file_path_for_saving_result': file_path_for_saving_result, 'do_multiprocessing_dry_run': self.do_multiprocessing_dry_run}, key=md5_code)...

queue_deduplicate.py

Source:queue_deduplicate.py

...27                ),28            )29        original_size, new_size = backend.deduplicate(30            queue,31            progress_logger=self.get_progress_logger(),32        )33        if original_size == new_size:34            self.stdout.write(35                "No duplicate jobs detected (queue length remains {})".format(36                    original_size,37                ),38            )39        else:40            self.stdout.write(41                "Deduplication reduced the queue from {} jobs to {} job(s)".format(42                    original_size,43                    new_size,44                ),45            )46    def get_progress_logger(self) -> ProgressLogger:47        try:48            import tqdm49            progress = tqdm.tqdm50        except ImportError:51            def progress(iterable: T) -> T:52                return iterable...

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.