yandextank/plugins/Phantom/utils.py
def read_config(self):
        """        Read phantom tool specific options        """
        self.threads = self.cfg["threads"] or str(int(multiprocessing.cpu_count() / 2) + 1)
        self.phantom_modules_path = self.cfg["phantom_modules_path"]
        self.additional_libs = ' '.join(self.cfg["additional_libs"])
        self.answ_log_level = self.cfg["writelog"]
        if self.answ_log_level.lower() in ['0', 'false']:
            self.answ_log_level = 'none'
        elif self.answ_log_level.lower() in ['1', 'true']:
            self.answ_log_level = 'all'
        self.timeout = parse_duration(self.cfg["timeout"])
        if self.timeout > 120000:
            logger.warning(
                "You've set timeout over 2 minutes."
                " Are you a functional tester?")
        self.answ_log = self.core.mkstemp(".log", "answ_")
        self.core.add_artifact_file(self.answ_log)
        self.core.add_artifact_file(self.phout_file)
        self.core.add_artifact_file(self.stat_log)
        self.phantom_log = self.core.mkstemp(".log", "phantom_")
        self.core.add_artifact_file(self.phantom_log)

        main_stream = StreamConfig(
            self.core,
            len(self.streams), self.phout_file, self.answ_log,
            self.answ_log_level, self.timeout, self.cfg, True)
        self.streams.append(main_stream)

        for section in self.multi():
            self.streams.append(
                StreamConfig(
                    self.core,
                    len(self.streams), self.phout_file, self.answ_log,
                    self.answ_log_level, self.timeout, section))

        for stream in self.streams:
            stream.read_config()

        if any(stream.ssl for stream in self.streams):
            self.additional_libs += ' ssl io_benchmark_method_stream_transport_ssl'
def read_config(self):
        self.threads = self.cfg["threads"] or str(int(multiprocessing.cpu_count() / 2) + 1)
        self.phantom_modules_path = self.cfg["phantom_modules_path"]
        self.additional_libs = ' '.join(self.cfg["additional_libs"])
        self.answ_log_level = self.cfg["writelog"]
        if self.answ_log_level.lower() in ['0', 'false']:
            self.answ_log_level = 'none'
        elif self.answ_log_level.lower() in ['1', 'true']:
            self.answ_log_level = 'all'
        self.timeout = parse_duration(self.cfg["timeout"])
        if self.timeout > 120000:
            logger.warning(
                "You've set timeout over 2 minutes."
                " Are you a functional tester?")
        self.answ_log = self.core.mkstemp(".log", "answ_")
        self.core.add_artifact_file(self.answ_log)
        self.core.add_artifact_file(self.phout_file)
        self.core.add_artifact_file(self.stat_log)
        self.phantom_log = self.core.mkstemp(".log", "phantom_")
        self.core.add_artifact_file(self.phantom_log)

        main_stream = StreamConfig(
            self.core,
            len(self.streams), self.phout_file, self.answ_log,
            self.answ_log_level, self.timeout, self.cfg, True)
        self.streams.append(main_stream)

        for section in self.multi():
            self.streams.append(
                StreamConfig(
                    self.core,
                    len(self.streams), self.phout_file, self.answ_log,
                    self.answ_log_level, self.timeout, section))

        for stream in self.streams:
            stream.read_config()

        if any(stream.ssl for stream in self.streams):
            self.additional_libs += ' ssl io_benchmark_method_stream_transport_ssl'
def read_config ( self ) : self . threads = self . cfg [ "threads" ] or str ( int ( multiprocessing . cpu_count ( ) / 2 ) + 1 ) self . phantom_modules_path = self . cfg [ "phantom_modules_path" ] self . additional_libs = ' ' . join ( self . cfg [ "additional_libs" ] ) self . answ_log_level = self . cfg [ "writelog" ] if self . answ_log_level . lower ( ) in [ '0' , 'false' ] : self . answ_log_level = 'none' elif self . answ_log_level . lower ( ) in [ '1' , 'true' ] : self . answ_log_level = 'all' self . timeout = parse_duration ( self . cfg [ "timeout" ] ) if self . timeout > 120000 : logger . warning ( "You've set timeout over 2 minutes." " Are you a functional tester?" ) self . answ_log = self . core . mkstemp ( ".log" , "answ_" ) self . core . add_artifact_file ( self . answ_log ) self . core . add_artifact_file ( self . phout_file ) self . core . add_artifact_file ( self . stat_log ) self . phantom_log = self . core . mkstemp ( ".log" , "phantom_" ) self . core . add_artifact_file ( self . phantom_log ) main_stream = StreamConfig ( self . core , len ( self . streams ) , self . phout_file , self . answ_log , self . answ_log_level , self . timeout , self . cfg , True ) self . streams . append ( main_stream ) for section in self . multi ( ) : self . streams . append ( StreamConfig ( self . core , len ( self . streams ) , self . phout_file , self . answ_log , self . answ_log_level , self . timeout , section ) ) for stream in self . streams : stream . read_config ( ) if any ( stream . ssl for stream in self . streams ) : self . additional_libs += ' ssl io_benchmark_method_stream_transport_ssl'
Read phantom tool specific options
Read phantom tool specific options