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

How to use create_startup_config method in yandex-tank

Best Python code snippet using yandex-tank

client.py

Source:client.py

...41    def install(self):42        self.workdir = tempfile.mkdtemp()43        logger.info("Created temp dir %s", self.workdir)44        agent_config = self.config.create_collector_config(self.workdir)45        startup_config = self.config.create_startup_config()46        customs_script = self.config.create_custom_exec_script()47        try:48            copyfile(49                self.path['AGENT_LOCAL_FOLDER'] + '/agent.py',50                self.workdir + '/agent.py')51            copyfile(agent_config, self.workdir + '/agent.cfg')52            copyfile(startup_config, self.workdir + '/agent_startup.cfg')53            copyfile(customs_script, self.workdir + '/agent_customs.sh')54            if not os.path.isfile(self.path['TELEGRAF_LOCAL_PATH']):55                logger.error(56                    'Telegraf binary not found at specified path: %s\n'57                    'You can download telegraf binaries here: https://github.com/influxdata/telegraf\n'58                    'or install debian package: `telegraf`',59                    self.path['TELEGRAF_LOCAL_PATH'])60                return None, None, None61        except Exception:62            logger.error(63                "Failed to copy agent to %s on localhost",64                self.workdir,65                exc_info=True)66            return None, None, None67        return agent_config, startup_config, customs_script68    @staticmethod69    def popen(cmnd):70        return subprocess.Popen(71            cmnd,72            bufsize=0,73            preexec_fn=os.setsid,74            close_fds=True,75            shell=True,76            stdout=subprocess.PIPE,77            stderr=subprocess.PIPE,78            stdin=subprocess.PIPE, )79    def start(self):80        """Start local agent"""81        logger.info('Starting agent on localhost')82        command = "{python} {work_dir}/agent.py --telegraf {telegraf_path} --host {host}".format(83            python=self.python,84            work_dir=self.workdir,85            telegraf_path=self.path['TELEGRAF_LOCAL_PATH'],86            host=self.host)87        self.session = self.popen(command)88        self.reader_thread = threading.Thread(target=self.read_buffer)89        self.reader_thread.setDaemon(True)90        return self.session91    def read_buffer(self):92        while self.session:93            try:94                chunk = self.session.stdout.read(4096)95                if chunk:96                    parts = chunk.rsplit('\n', 1)97                    if len(parts) > 1:98                        ready_chunk = self.buffer + parts[0] + '\n'99                        self.buffer = parts[1]100                        self.incoming_queue.put(ready_chunk)101                    else:102                        self.buffer += parts[0]103            except ValueError:104                logger.debug(105                    'this exc most likely raised during interpreter shutdown\n'106                    'otherwise something really nasty happend',107                    exc_info=True)108    def uninstall(self):109        """110        Remove agent's files from remote host111        """112        if self.session:113            logger.info('Waiting monitoring data...')114            self.session.terminate()115            self.session.wait()116            self.session = None117        log_filename = "agent_{host}.log".format(host="localhost")118        data_filename = "agent_{host}.rawdata".format(host="localhost")119        try:120            logger.info('Saving monitoring artefacts from localhost')121            copyfile(self.workdir + "/_agent.log", log_filename)122            copyfile(self.workdir + "/monitoring.rawdata", data_filename)123            logger.info('Deleting temp directory: %s', self.workdir)124            rmtree(self.workdir)125        except Exception:126            logger.error("Exception while uninstalling agent", exc_info=True)127        logger.info("Removing agent from: localhost")128        return log_filename, data_filename129class SSHClient(object):130    """remote agent client setup """131    def __init__(self, config, old_style_configs, timeout):132        # config133        self.host = config['host']134        self.username = config['username']135        self.python = config['python']136        self.port = config['port']137        self.telegraf = config['telegraf']138        self.config = AgentConfig(config, old_style_configs)139        # connection140        self.session = None141        self.ssh = SecuredShell(self.host, self.port, self.username, timeout)142        self.incoming_queue = queue.Queue()143        self.buffer = ""144        self.reader = MonitoringReader(self.incoming_queue)145        handle, cfg_path = tempfile.mkstemp('.cfg', 'agent_')146        os.close(handle)147        self.path = {148            # Destination path on remote host149            'AGENT_REMOTE_FOLDER': '/tmp/',150            # Source path on tank151            'AGENT_LOCAL_FOLDER': os.path.dirname(__file__) + '/agent',152            'TELEGRAF_REMOTE_PATH': '/tmp/telegraf',153            'TELEGRAF_LOCAL_PATH': self.telegraf,154        }155    def install(self):156        """Create folder and copy agent and metrics scripts to remote host"""157        logger.info(158            "Installing monitoring agent at %s@%s...", self.username, self.host)159        # create remote temp dir160        cmd = self.python + ' -c "import tempfile; print tempfile.mkdtemp();"'161        logger.info("Creating temp dir on %s", self.host)162        try:163            out, errors, err_code = self.ssh.execute(cmd)164        except Exception:165            logger.error(166                "Failed to install monitoring agent to %s",167                self.host,168                exc_info=True)169            return None, None, None170        if errors:171            logging.error("[%s] error: '%s'", self.host, errors)172            return None, None, None173        if err_code:174            logging.error(175                "Failed to create remote dir via SSH at %s@%s, code %s: %s" %176                (self.username, self.host, err_code, out.strip()))177            return None, None, None178        remote_dir = out.strip()179        if remote_dir:180            self.path['AGENT_REMOTE_FOLDER'] = remote_dir181        logger.debug(182            "Remote dir at %s:%s", self.host, self.path['AGENT_REMOTE_FOLDER'])183        # create collector config184        agent_config = self.config.create_collector_config(185            self.path['AGENT_REMOTE_FOLDER'])186        startup_config = self.config.create_startup_config()187        customs_script = self.config.create_custom_exec_script()188        # trying to detect os version/architecture and get information about telegraf client189        # DO NOT DELETE indices in string format below. Python 2.6 does not190        # support string formatting without indices191        remote_cmd = 'import os; print os.path.isfile("' + self.path[192            'TELEGRAF_REMOTE_PATH'] + '")'193        cmd = self.python + ' -c \'{cmd}\''.format(cmd=remote_cmd)194        remote_telegraf_exists = "False"195        try:196            out, err, err_code = self.ssh.execute(cmd)197        except Exception:198            logger.error(199                "SSH execute error trying to check telegraf availability on host %s",200                self.host,...

test_config.py

Source:test_config.py

...49        telegraf_configs = manager.getconfig(50            'yandextank/plugins/Telegraf/tests/telegraf_mon.xml',51            'sometargethint')52        agent_config = AgentConfig(telegraf_configs[0], False)53        startup = agent_config.create_startup_config()54        cfg_parser = ConfigParser()55        cfg_parser.read(startup)56        assert cfg_parser.has_section('startup')57    def test_create_collector_configs(self):58        """ test agent config creates collector config """59        manager = ConfigManager()60        telegraf_configs = manager.getconfig(61            'yandextank/plugins/Telegraf/tests/telegraf_mon.xml',62            'sometargethint')63        agent_config = AgentConfig(telegraf_configs[0], False)64        remote_workdir = '/path/to/workdir/temp'65        collector_config = agent_config.create_collector_config(remote_workdir)66        cfg_parser = ConfigParser()67        cfg_parser.read(collector_config)...

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.