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

How to use tankapi_info method in yandex-tank

Best Python code snippet using yandex-tank

plugin.py

Source:plugin.py

...77        """78        if self._data_session is None:79            config_filenames = {'validated_conf.yaml', 'configinitial.yaml'}80            self._data_session = DataSession({'clients': self.clients_cfg},81                                             tankapi_info=self.tankapi_info(),82                                             config_filenames=config_filenames,83                                             artifacts_dir=self.core.artifacts_dir,84                                             test_start=self.core.info.get_value(['generator', 'test_start'], 0) * 10**6)85            self.add_cleanup(self._cleanup)86            self._data_session.update_job(dict({'name': self.test_name,87                                                '__type': 'tank'},88                                               **self.meta))89            job_no = self._data_session.clients[0].job_number90            if job_no:91                self.publish('job_no', int(job_no))92                self.publish('web_link', urljoin(self.LUNA_LINK, job_no))93        return self._data_session94    def tankapi_info(self):95        meta = self.cfg.get('meta', {})96        return {97            'host': meta.get('tankapi_host'),98            'port': meta.get('tankapi_port'),99            'local_id': self.core.test_id100        }101    def _cleanup(self):102        self.upload_actual_rps(data=pandas.DataFrame([]), last_piece=True)103        uploader_metainfo = self.get_lp_meta()104        autostop_info = self.get_autostop_info()105        regressions = self.get_regressions_names(uploader_metainfo)106        lp_link = self.core.info.get_value(['uploader', 'web_link'])107        meta = self.meta108        meta.update(autostop_info)...

manager.py

Source:manager.py

1import logging2import uuid3import time4import os5import getpass6import six7from netort.data_manager.common.util import thread_safe_property8from typing import Callable, Dict, Text, List, Union, Optional, Set, Type9from .metrics import Metric, Event10from .clients import available_clients11from .router import MetricsRouter12from .common.interfaces import AbstractMetric, DataType, MetricData  # noqa13if six.PY3:14    from queue import Queue15else:  # six.PY216    # noinspection PyUnresolvedReferences17    from Queue import Queue18logger = logging.getLogger(__name__)19# TODO: move code that works with config to library's clients (e.g., Volta).20# Classes of this library should provide constructors with described arguments only21# noinspection PyBroadException22class DataSession(object):23    """24    Workflow:25        * create DataSession object26        * use `new_metric` to add metrics to your datasession27        * use `metric.put` to add data to the metric28        * call `close` to close your datasession29    Note:30        * send your data in chunks because it could be of bigger size that server's buffer31    Args:32        config(dict): configuration options (list of DataManager clients, test meta data etc)33    TODO:34        * move config parameters to kwargs, describe them here35        * fight performance issues (probably caused by poor pandas write_csv performance)36    """37    def __init__(self, config, tankapi_info=None, config_filenames=None, artifacts_dir=None, test_start=None):38        self.start_ts = time.time()                                                  # type: float39        self.config = config                                                         # type: Dict40        self.operator = self.__get_operator()                                        # type: Text41        self.job_id = config.get('test_id', 'job_{uuid}'.format(uuid=uuid.uuid4()))  # type: Text42        logger.info('Created new local data session: %s', self.job_id)43        self.tankapi_info = tankapi_info                                             # type: Dict44        self.config_filenames = config_filenames or set()                            # type: Set45        self.test_start = test_start if test_start else int(time.time() * 10**6)     # type: float46        self.artifacts_base_dir = artifacts_dir or './logs'                          # type: Text47        self._artifacts_dir = None                                                   # type: Union[Text, None]48        self.manager = DataManager()                                                 # type: DataManager49        self.clients = []                                                            # type: List50        self.__create_clients(config.get('clients', []))51        logger.debug('DataSession clients: %s', self.clients)52    # TODO: extract client creation as factory method53    # TODO: consider removing clients from config and add them via `new_client` method54    def __create_clients(self, clients):55        # type: (Dict) -> None56        for client_meta in clients:57            type_ = client_meta.get('type')58            if not type_:59                raise ValueError('Client type should be defined.')60            if type_ in available_clients:61                client = available_clients[type_](client_meta, self)62                self.subscribe(client.put)63                self.clients.append(client)64            else:65                raise NotImplementedError('Unknown client type: %s' % type_)66    def new_true_metric(self, meta, raw=True, aggregate=False, parent=None, case=None):67        # type: (dict, bool, bool, str, str) -> AbstractMetric68        return self.manager.new_true_metric(meta=meta,69                                            test_start=self.test_start,70                                            raw=raw, aggregate=aggregate,71                                            parent=parent, case=case)72    def new_event_metric(self, meta, raw=True, aggregate=False, parent=None, case=None):73        # type: (dict, bool, bool, str, str) -> AbstractMetric74        return self.manager.new_event_metric(meta=meta,75                                             test_start=self.test_start,76                                             raw=raw, aggregate=aggregate,77                                             parent=parent, case=case)78    def subscribe(self, callback):79        # type: (Callable) -> Any80        return self.manager.subscribe(callback)81    def get_metric_by_id(self, id_):82        # type: (Text) -> Any83        return self.manager.get_metric_by_id(id_)84    def update_job(self, meta):85        # type: (Dict) -> Any86        for client in self.clients:87            try:88                client.update_job(meta)89            except Exception:90                logger.warning('Client %s job update failed', client)91                logger.debug('Client %s job update failed', client, exc_info=True)92            else:93                logger.debug('Client job updated: %s', client)94    def update_metric(self, meta):95        # type: (Dict) -> None96        for client in self.clients:97            try:98                client.update_metric(meta)99            except Exception:100                logger.warning('Client %s metric update failed', client)101                logger.debug('Client %s metric update failed', client, exc_info=True)102            else:103                logger.debug('Client metric updated: %s', client)104    # TODO: artifacts dir should be inside "local" client. Or does it?105    @thread_safe_property106    def artifacts_dir(self):107        # type: () -> Text108        if not self._artifacts_dir:109            dir_name = "{dir}/{id}".format(dir=self.artifacts_base_dir, id=self.job_id)110            if not os.path.isdir(dir_name):111                os.makedirs(dir_name)112            os.chmod(dir_name, 0o755)113            self._artifacts_dir = os.path.abspath(dir_name)114        return self._artifacts_dir115    def __get_operator(self):116        # type: () -> Text117        try:118            return self.config.get('operator') or getpass.getuser()119        except:  # noqa: E722120            logger.error(121                "Couldn't get username from the OS. Please, set the 'operator' option explicitly in your config "122                "file.")123            raise124    def close(self, test_end):125        # type: (float) -> None126        logger.info('DataSession received close signal.')127        logger.info('Closing DataManager')128        self.manager.close()129        logger.info('Waiting the rest of data from router...')130        self.manager.router.join()131        logger.info('Sending close to DataSession clients...')132        for client in self.clients:133            try:134                client.close(test_end)135            except Exception:136                logger.warning('Client %s failed to close', client, exc_info=True)137            else:138                logger.debug('Client closed: %s', client)139        logger.info('DataSession finished!')140        logger.info('DataSession time: %s', time.time() - self.start_ts)141    def interrupt(self):142        # type: () -> None143        self.manager.interrupt()144        for client in self.clients:145            try:146                client.close()147            except Exception:148                logger.warning('Client %s failed to close', client)149            else:150                logger.debug('Client closed: %s', client)151        logger.info('DataSession finished!')152class DataManager(object):153    """DataManager routes data to subscribers using metrics meta as a filter. When someone calls154    `new_metric`, DataManager will find the subscribers that are interested in this metric (using meta).155    When someone calls `subscribe`, DataManager finds the metrics that this subscriber is interested in.156    MetricsRouter is a facility that DataManager uses for passing incoming data to subscribers.157    Attributes:158        metrics: All registered metrics for DataManager session159        subscribers: All registered subscribers for DataManager session160        callbacks: callbacks for metric ids <-> subscribers' callbacks, used by router161        routing_queue: incoming unrouted metrics data, will be processed by MetricsRouter to subscribers' callbacks162        router (MetricsRouter object): Router thread. Read routing queue, concat incoming messages by metrics.type,163            left join by callback and call callback w/ resulting dataframe164    """165    def __init__(self):166        self.metrics = {}                   # type: Dict[Text, AbstractMetric]167        self.metrics_meta = {}              # type: Dict[Text, Text]168        self.subscribers = {}               # type: Dict[Text, Callable]169        self.callbacks = {}                 # type: Dict[Text, Set]170        self.routing_queue = Queue()        # type: Queue171        self.router = MetricsRouter(self)   # type: MetricsRouter172        self.router.start()173    def new_true_metric(self, meta, test_start, raw=True, aggregate=False, parent=None, case=None):174        # type: (dict, float, bool, bool, str, str) -> AbstractMetric175        """176        Create and register metric,177        find subscribers for this metric (using meta as filter) and subscribe178        Return:179            metric: one of Metric180        """181        return self._new_metric(Metric, meta, test_start, raw, aggregate, parent=parent, case=case)182    def new_event_metric(self, meta, test_start, raw=True, aggregate=False, parent=None, case=None):183        # type: (dict, float, bool, bool, str, str) -> AbstractMetric184        return self._new_metric(Event, meta, test_start, raw, aggregate, parent=parent, case=case)185    def _new_metric(self, dtype, meta, test_start, raw=True, aggregate=False, parent=None, case=None):186        # type: (Type[AbstractMetric], dict, float, bool, bool, str, str) -> AbstractMetric187        metric_obj = dtype(meta=meta,188                           _queue=self.routing_queue,189                           test_start=test_start,190                           raw=raw, aggregate=aggregate,191                           parent=parent, case=case)  # create metric object192        self.metrics_meta = meta  # register metric meta193        self.metrics[metric_obj.local_id] = metric_obj  # register metric object194        for callback in self.callbacks:195            self.callbacks[callback].add(metric_obj.local_id)196        return metric_obj197    def subscribe(self, callback):198        # type: (Callable) -> None199        """200        Create and register metric subscriber,201        subscribe all existing metrics for it202        Args:203            callback (object method): subscriber's callback204        """205        sub_id = "subscriber_{uuid}".format(uuid=uuid.uuid4())206        # register subscriber in manager207        self.subscribers[sub_id] = callback208        self.callbacks[sub_id] = set(iter(self.metrics))209    def get_metric_by_id(self, id_):210        # type: (Text) -> Optional[AbstractMetric]211        return self.metrics.get(id_)212    def close(self):213        # type: () -> None214        self.router.close()215    def interrupt(self):216        # type: () -> None217        self.router.interrupt()218        self.router.join()219# def usage_sample():220#     import time221#     import pandas as pd222#     config = {223#         'clients': [224#             {225#                 'type': 'luna',226#                 'api_address': 'http://hostname.tld',227#                 'user_agent': 'Tank Test',228#             },229#             {230#                 'type': 'local_storage',231#             }232#         ],233#         'test_start': time.time(),234#         'artifacts_base_dir': './logs'235#     }236#     data_session = DataSession(config=config)237#238#     metric_meta = {239#         'type': 'metrics',240#         'name': 'cpu_usage',241#         'hostname': 'localhost',242#         'some_meta_key': 'some_meta_value'243#     }244#245#     metric_obj = data_session.new_true_metric('name', **metric_meta)246#     time.sleep(1)247#     df = pd.DataFrame([[123, 123.123, "trash"]], columns=['ts', 'value', 'trash'])248#     metric_obj.put(df)249#     df2 = pd.DataFrame([[456, 456.456]], columns=['ts', 'value'])250#     metric_obj.put(df2)251#     time.sleep(10)252#     df = pd.DataFrame([[123, 123.123]], columns=['ts', 'value'])253#     metric_obj.put(df)254#     df2 = pd.DataFrame([[456, 456.456]], columns=['ts', 'value'])255#     metric_obj.put(df2)256#     time.sleep(10)257#     data_session.close(test_end=time.time())258if __name__ == '__main__':259    logging.basicConfig(level='DEBUG')260    logger = logging.getLogger(__name__)...

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.