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

How to use lp_job method in yandex-tank

Best Python code snippet using yandex-tank

plugin.py

Source:plugin.py

...482                      connection_timeout=int(self.get_option('connection_timeout', 30)),483                      user_agent=self._get_user_agent(),484                      api_token=self.api_token)485    @property486    def lp_job(self):487        if self._lp_job is None:488            self._lp_job = self.__get_lp_job()489        return self._lp_job490    def __get_lp_job(self):491        api_client = self.__get_api_client()492        info = self.generator_info493        port = info.port494        loadscheme = [] if isinstance(info.rps_schedule,495                                      str) else info.rps_schedule496        return LPJob(client=api_client,497                     target_host=self.target,498                     target_port=port,499                     number=self.get_option('jobno', ''),500                     token=self.get_option('upload_token', ''),501                     person=self.__get_operator(),502                     task=self.task,503                     name=self.get_option('job_name', 'none').decode('utf8'),504                     description=self.get_option('job_dsc', '').decode('utf8'),...

cli.py

Source:cli.py

1import ConfigParser2import argparse3import glob4import json5import logging6import os7import socket8import sys9import pwd10from StringIO import StringIO11from urlparse import urljoin12from datetime import datetime13import pkg_resources14from .client import APIClient15from .plugin import LPJob16CONFIG_FILE = 'saved_conf.ini'17DATA_LOG = 'test_data.log'18MONITORING_LOG = 'monitoring.log'19SECTION = 'meta'20logger = logging.getLogger('')21logger.setLevel(logging.DEBUG)22handler = logging.StreamHandler(stream=sys.stdout)23handler.setLevel(logging.INFO)24logger.addHandler(handler)25verbose_handler = logging.FileHandler(26    datetime.now().strftime("post_loader_%Y-%m-%d_%H-%M-%S.log"), 'w')27verbose_handler.setLevel(logging.DEBUG)28logger.addHandler(verbose_handler)29def read_config(shooting_dir):30    config_file = glob.glob(os.path.join(shooting_dir, CONFIG_FILE))[0]31    logger.info('Config file found: %s' % config_file)32    config = ConfigParser.ConfigParser()33    config.read(config_file)34    return config35def get_lp_config(config):36    """37    looks for config file in shooting_dir,38    returns config dict of section 'meta'39    :rtype: dict40    """41    lp_config = dict(config.items(SECTION))42    for key in sorted(lp_config.keys()):43        logger.debug('%s: %s' % (key, lp_config[key]))44    return lp_config45def check_log(log_name):46    assert os.path.exists(log_name), \47        'Data log {} not found\n'.format(log_name) + \48        'JsonReport plugin should be enabled when launching Yandex-tank'49def upload_data(shooting_dir, log_name, lp_job):50    data_log = os.path.join(shooting_dir, log_name)51    check_log(data_log)52    sys.stdout.write('Uploading test data')53    with open(data_log, 'r') as f:54        for line in f:55            data = json.loads(line.strip())56            lp_job.push_test_data(data['data'], data['stats'])57            sys.stdout.write('.')58            sys.stdout.flush()59    sys.stdout.write('\n')60def upload_monitoring(shooting_dir, log_name, lp_job):61    data_log = os.path.join(shooting_dir, log_name)62    check_log(data_log)63    sys.stdout.write('Uploading monitoring data')64    with open(data_log, 'r') as f:65        for line in f:66            lp_job.push_monitoring_data(line.strip())67            sys.stdout.write('.')68            sys.stdout.flush()69    sys.stdout.write('\n')70def send_config_snapshot(config, lp_job):71    config.set(SECTION, 'launched_from', 'post-loader')72    output = StringIO()73    config.write(output)74    lp_job.send_config_snapshot(output.getvalue())75def edit_metainfo(lp_config, lp_job):76    lp_job.edit_metainfo(is_regression=lp_config.get('regress'),77                         regression_component=lp_config.get('component'),78                         cmdline=lp_config.get('cmdline'),79                         ammo_path=lp_config.get('ammo_path'),80                         loop_count=lp_config.get('loop_count'))81def get_plugin_dir(shooting_dir):82    DIRNAME = 'lunapark'83    parent = os.path.abspath(os.path.join(shooting_dir, os.pardir))84    if os.path.basename(parent) == DIRNAME:85        return parent86    else:87        plugin_dir = os.path.join(parent, DIRNAME)88        if not os.path.exists(plugin_dir):89            os.makedirs(plugin_dir)90        return plugin_dir91def make_symlink(shooting_dir, name):92    plugin_dir = get_plugin_dir(shooting_dir)93    link_name = os.path.join(plugin_dir, str(name))94    os.symlink(os.path.relpath(shooting_dir, plugin_dir), link_name)95    logger.info('Symlink created: {}'.format(os.path.abspath(link_name)))96def post_loader():97    parser = argparse.ArgumentParser()98    parser.add_argument('shooting_dir',99                        help='Directory containing shooting artifacts')100    shooting_dir = parser.parse_args().shooting_dir101    assert os.path.exists(shooting_dir), 'Directory not found'102    config = read_config(shooting_dir)103    lp_config = get_lp_config(config)104    api_client = APIClient(base_url=lp_config['api_address'],105                           user_agent='Lunapark/{}'.format(pkg_resources.require('yatank-internal-lunapark')[0].version)106                           # todo: add timeouts107                           )108    lp_job = LPJob(109        client=api_client,110        target_host=lp_config.get('target_host'),111        target_port=lp_config.get('target_port'),112        person=lp_config.get(113            'operator',114            '') or pwd.getpwuid(115            os.geteuid())[0],116        task=lp_config['task'],117        name=lp_config['job_name'],118        description=lp_config['job_dsc'],119        tank=socket.getfqdn())120    edit_metainfo(lp_config, lp_job)121    upload_data(shooting_dir, DATA_LOG, lp_job)122    send_config_snapshot(config, lp_job)123    try:124        upload_monitoring(shooting_dir, MONITORING_LOG, lp_job)125    except AssertionError as e:126        logger.error(e)127    lp_job.close(0)128    make_symlink(shooting_dir, lp_job.number)129    logger.info(130        'LP job created: {}'.format(131            urljoin(132                api_client.base_url, str(133                    lp_job.number))))134if __name__ == '__main__':...

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.