How to use plugins_cleanup method in yandex-tank

Best Python code snippet using yandex-tank

apiworker.py

Source:apiworker.py

1# -*- coding: utf-8 -*-2""" Provides class to run TankCore from python """3import ctypes4import logging5from multiprocessing import Value, Process6from yandextank.common.util import Cleanup, Finish, Status7from yandextank.core.tankworker import TankWorker8logger = logging.getLogger()9class ApiWorker(Process, TankWorker):10    SECTION = 'core'11    FINISH_FILENAME = 'finish_status.yaml'12    def __init__(self, manager, config_paths, cli_options=None, cfg_patches=None, cli_args=None, no_local=False,13                 log_handlers=None, wait_lock=False, files=None, ammo_file=None):14        Process.__init__(self)15        TankWorker.__init__(self, configs=config_paths, cli_options=cli_options, cfg_patches=cfg_patches,16                            cli_args=cli_args, no_local=no_local, log_handlers=log_handlers,17                            wait_lock=wait_lock, files=files, ammo_file=ammo_file, api_start=True, manager=manager)18        self._status = Value(ctypes.c_char_p, Status.TEST_INITIATED)19        self._test_id = Value(ctypes.c_char_p, self.core.test_id.encode('utf8'))20        self._retcode = Value(ctypes.c_int, 0)21        self._msg = Value(ctypes.c_char_p, b'')22    @property23    def test_id(self):24        return self._test_id.value.decode('utf8')25    @property26    def status(self):27        self._status.acquire()28        res = self._status.value29        self._status.release()30        return res31    @status.setter32    def status(self, val):33        self._status.acquire()34        self._status.value = val35        self._status.release()36    @property37    def retcode(self):38        return self._retcode.value39    @retcode.setter40    def retcode(self, val):41        self._retcode.value = val42    @property43    def msg(self):44        self._msg.acquire()45        res = self._msg.value.decode('utf8')46        self._msg.release()47        return res48    @msg.setter49    def msg(self, val):50        value = val.encode('utf8')51        self._msg.acquire()52        self._msg.value = value53        self._msg.release()54    def run(self):55        with Cleanup(self) as add_cleanup:56            lock = self.get_lock()57            add_cleanup('release lock', lock.release)58            self.status = Status.TEST_PREPARING59            logger.info('Created a folder for the test. %s' % self.folder)60            self.core.plugins_configure()61            add_cleanup('plugins cleanup', self.core.plugins_cleanup)62            self.core.plugins_prepare_test()63            with Finish(self):64                self.status = Status.TEST_RUNNING65                self.core.plugins_start_test()66                self.retcode = self.core.wait_for_finish()67            self.status = Status.TEST_POST_PROCESS68            self.retcode = self.core.plugins_post_process(self.retcode)69class SingleLevelFilter(logging.Filter):70    """Exclude or approve one msg type at a time.    """71    def __init__(self, passlevel, reject):72        logging.Filter.__init__(self)73        self.passlevel = passlevel74        self.reject = reject75    def filter(self, record):76        if self.reject:77            return record.levelno != self.passlevel78        else:...

plugins_h.py

Source:plugins_h.py

...20                                            void *user_data);21extern void plugins_get_descriptions(plugin_description_callback callback, void *user_data);22extern void plugins_dump_all(void);23extern int plugins_get_count(void);24extern void plugins_cleanup(plugins_t *plugins);...

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.