Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _combine_configs method in yandex-tank

Best Python code snippet using yandex-tank

auto.py

Source:auto.py

...200                raise ValueError(f"parameter '{key}' is not configurable")201        self._config = (202            {**self._base_config, **config}  # config takes precedence (PEP 448)203            if reset204            else self._combine_configs(input_op, downstream_config=config)205        )206    @classmethod207    def register_configurable(cls, names):208        """Globally registers the parameter names as being configurable."""209        cls._configurable |= set(names)210    @classmethod211    def resolve_config(cls, input_ops):212        """Returns the config for an operator with the given input operators."""213        # Combine from left -> right so that the precedence goes from left -> right.214        upstream_config = cls._base_config215        for input_op in input_ops:216            upstream_config = cls._combine_configs(217                upstream_op=input_op, downstream_config=upstream_config218            )219        return upstream_config220    @classmethod221    def _combine_configs(cls, upstream_op, downstream_config):222        """Combines downstream with all upstream configs, following precedence rules."""223        if isinstance(upstream_op, cls):224            upstream_config = upstream_op._config  # dict is copied below225        elif upstream_op in cls._cache:226            upstream_config = cls._cache[upstream_op]227        else:228            # Note that downstream_config is not used here, which makes this229            # context-free. Combining this fact with the immutability of operators230            # makes this cacheable.231            upstream_config = cls._cache[upstream_op] = cls.resolve_config(232                input_ops=upstream_op.input_ops233            )234        # Downstream takes precedence over upstream (see PEP 448).235        return {**upstream_config, **downstream_config}...

tankworker.py

Source:tankworker.py

...30        self.ammo_file = ammo_file31        self.config_paths = configs32        self.interrupted = ProcessEvent() if api_start else ThreadEvent()33        self.info = TankInfo(manager.dict()) if api_start else TankInfo(dict())34        self.config_list = self._combine_configs(configs, cli_options, cfg_patches, cli_args, no_local)35        self.core = TankCore(self.config_list, self.interrupted, self.info)36        self.folder = self.init_folder()37        self.init_logging(debug or self.core.get_option(self.core.SECTION, 'debug'))38        is_locked = Lock.is_locked(self.core.lock_dir)39        if is_locked and not self.core.config.get_option(self.SECTION, 'ignore_lock'):40            raise LockError(is_locked)41    @staticmethod42    def _combine_configs(run_cfgs, cli_options=None, cfg_patches=None, cli_args=None, no_local=False):43        if cli_options is None:44            cli_options = []45        if cfg_patches is None:46            cfg_patches = []47        if cli_args is None:48            cli_args = []49        run_cfgs = run_cfgs if len(run_cfgs) > 0 else [TankWorker.DEFAULT_CONFIG]50        if no_local:51            configs = [load_cfg(cfg) for cfg in run_cfgs] + \52                parse_options(cli_options) + \53                parse_and_check_patches(cfg_patches) + \54                cli_args55        else:56            configs = [load_core_base_cfg()] + \...

bridge.py

Source:bridge.py

...8try:9    import hamster.client10except ImportError:11    raise ImportError('Can not find hamster')12def _combine_configs(*configs):13    """14    Combines all configs (instances of RawConfigParser) into a single one15    containing all sections and values. If duplicates appear the later configs16    will overwrite the earlier values.17    Returns a RawConfigParser() instance.18    """19    result = ConfigParser.RawConfigParser()20    for config in configs:21        for section in config.sections():22            try:23                result.add_section(section)24            except ConfigParser.DuplicateSectionError:25                # Ignore it. We simply want to include all sections from26                # the source configs27                pass28            for option in config.options(section):29                value = config.get(section, option)30                result.set(section, option, value)31    return result32class HamsterBridge(hamster.client.Storage):33    """34    Connects to the running hamster instance via dbus. But as the notification does not work reliable there is a35    polling-based loop in the run()-method that will trigger all registered listeners.36    """37    def __init__(self, save_passwords=False):38        super(HamsterBridge, self).__init__()39        self._listeners = []40        self.save_passwords = save_passwords41    def add_listener(self, listener):42        """43        Registers the given HamsterListener instance. It will then be notified about changes.44        :param listener: the HamsterListener instance45        :type  listener: HamsterListener46        """47        if listener not in self._listeners:48            self._listeners.append(listener)49    def configure(self, config_path):50        """51        Gives each listener the chance to do something before we start the bridge's runtime loops.52        :param config_path: path to config file53        :type config_path:  str54        """55        path = os.path.expanduser(config_path)56        config = ConfigParser.RawConfigParser()57        sensitive_config = ConfigParser.RawConfigParser()58        # read from file if exists59        if os.path.exists(path):60            logger.debug('Reading config file from %s', path)61            config.read(path)62        # let listeners extend63        for listener in self._listeners:64            logger.debug('Configuring listener %s', listener)65            listener.configure(config, sensitive_config)66        # save to file67        with open(path, 'wb') as configfile:68            logger.debug('Writing back configuration to %s', path)69            if self.save_passwords:70                all_configs = _combine_configs(config, sensitive_config)71                all_configs.write(configfile)72            else:73                config.write(configfile)74        # as we store passwords in clear text, let's at least set correct file permissions75        logger.debug('Setting owner only file permissions to %s', path)76        os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)77    def run(self, polling_intervall=1):78        """79        Starts the polling loop that will run until receive common exit signals.80        :param polling_intervall: how often the connector polls data from haster in seconds (default: 1)81        :type  polling_intervall: int82        """83        try:84            for listener in self._listeners:...

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.