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

How to use _deploy_nodes method in lisa

Best Python code snippet using lisa_python

node_manager.py

Source:node_manager.py

...47        }48    def deploy_nodes(self):49        ret = False50        if self.params.ela_params.enable:51            ret = self._deploy_nodes("ela", self.params.ela_params.number)52            self.later_start_nodes = \53                self.ela_nodes[self.params.ela_params.number - self.params.ela_params.later_start_number + 1:]54        if self.params.did_params.enable:55            ret = self._deploy_nodes("did", self.params.did_params.number)56        if self.params.token_params.enable:57            ret = self._deploy_nodes("token", self.params.token_params.number)58        if self.params.neo_params.enable:59            ret = self._deploy_nodes("neo", self.params.neo_params.number)60        return ret61    def start_nodes(self):62        if self.params.ela_params.enable:63            for i in range(len(self.ela_nodes) - self.params.ela_params.later_start_number):64                self.ela_nodes[i].start()65                time.sleep(0.2)66            self.wait_rpc_ready(self.ela_nodes[0].rpc_port)67        if self.params.did_params.enable:68            for i in range(len(self.did_nodes)):69                self.did_nodes[i].start()70                time.sleep(0.2)71            self.wait_rpc_ready(self.did_nodes[0].rpc_port)72            self.create_side_info("did")73        if self.params.token_params.enable:74            for i in range(len(self.token_nodes)):75                self.token_nodes[i].start()76                time.sleep(0.2)77            self.wait_rpc_ready(self.token_nodes[0].rpc_port)78            self.create_side_info("token")79        if self.params.neo_params.enable:80            for i in range(len(self.neo_nodes)):81                self.neo_nodes[i].start()82                time.sleep(0.2)83            self.wait_rpc_ready(self.neo_nodes[0].rpc_port)84            self.create_side_info("neo")85        if self.params.arbiter_params.enable:86            if self.params.arbiter_params.enable:87                self._deploy_nodes("arbiter", self.params.arbiter_params.number)88                time.sleep(2)89                self.start_arbiter_nodes()90    def wait_rpc_ready(self, port: int, content=1, timeout=60):91        time.sleep(5)92        stop_time = time.time() + timeout93        while time.time() <= stop_time:94            result = []95            for i in range(self.params.ela_params.number):96                count = rpc.get_connection_count(port)97                Logger.debug('{} wait for rpc service ready, connection count: {}'.format(self.tag, count))98                if count and count >= content:99                    result.append(True)100                else:101                    result.append(False)102                if result.count(True) == self.params.ela_params.number:103                    Logger.debug('{} Nodes connect with each other, '104                                 'rpc service is started on success!.'.format(self.tag))105                    return True106                time.sleep(0.2)107        Logger.error('{} Node can not connect with each other, wait rpc service timed out!')108        return False109    def start_arbiter_nodes(self):110        arbiter_node_number = len(self.arbiter_nodes)111        if arbiter_node_number <= 0:112            return False113        for i in range(arbiter_node_number):114            self.arbiter_nodes[i].start()115        return True116    def stop_nodes(self):117        if self.params.ela_params.enable:118            for i in range(len(self.ela_nodes)):119                self.ela_nodes[i].stop()120                time.sleep(0.5)121        if self.params.arbiter_params.enable:122            for i in range(len(self.arbiter_nodes)):123                self.arbiter_nodes[i].stop()124                time.sleep(0.5)125        if self.params.did_params.enable:126            for i in range(len(self.did_nodes)):127                self.did_nodes[i].stop()128                time.sleep(0.5)129        if self.params.token_params.enable:130            for i in range(len(self.token_nodes)):131                self.token_nodes[i].stop()132                time.sleep(0.5)133        if self.params.neo_params.enable:134            for i in range(len(self.neo_nodes)):135                self.neo_nodes[i].stop()136                time.sleep(0.5)137    def _init_nodes(self, category: str, config, index: int, cwd_dir: str, ela_type="normal"):138        if category == "ela":139            node = ElaNode(index, config, self.params.ela_params, self.keystore_manager, cwd_dir, ela_type)140        elif category == "arbiter":141            node = ArbiterNode(index, config, self.params.arbiter_params, self.keystore_manager, cwd_dir)142        elif category == "did":143            node = DidNode(index, config, self.params.did_params, self.keystore_manager, cwd_dir)144        elif category == "token":145            node = TokenNode(index, config, self.params.token_params, self.keystore_manager, cwd_dir)146        elif category == "neo":147            node = NeoNode(index, config, self.params.neo_params, self.keystore_manager, cwd_dir)148        else:149            node = None150        return node151    def _deploy_nodes(self, category: str, num: int):152        src_path = os.path.join(self.env_manager.elastos_path, self.env_manager.src_path_dict[category])153        print("src path: ", src_path)154        if not os.path.exists(src_path):155            return False156        Logger.debug("{} src_path: {}".format(self.tag, src_path))157        config_path = os.path.join(src_path, "config.json.sample")158        if os.path.exists(config_path):159            Logger.debug("{} config.json will generate from the sample".format(self.tag))160            config_dict = util.read_config_file(config_path)161        else:162            Logger.debug("{} config.json will generate from the default".format(self.tag))163            config_dict = self.env_manager.config_dict[category]164        global ela_type165        global temp_dest_dir...

deploy_map.py

Source:deploy_map.py

1# Copyright (C) 2021, Bayerische Motoren Werke Aktiengesellschaft (BMW AG),2#   Author: Alexander Domin (Alexander.Domin@bmw.de)3# Copyright (C) 2021, ProFUSION Sistemas e SoluÃ§Ãµes LTDA,4#   Author: Gustavo Barbieri (barbieri@profusion.mobi)5#   Author: Garbiel Fernandes (g7fernandes@profusion.mobi)6#   Author: Leandro Ferlin (leandroferlin@profusion.mobi)7#   Author: Leonardo Ramos (leo.ramos@profusion.mobi)8#9# SPDX-License-Identifier: MPL-2.010#11# This Source Code Form is subject to the terms of the12# Mozilla Public License, v. 2.0. If a copy of the MPL was13# not distributed with this file, You can obtain one at14# http://mozilla.org/MPL/2.0/.15from typing import Any, Dict, Optional, Union16from .deploy_entry import VehicleDeployEntry17from .types.deploy_factory import deploy_yaml_keys18class VehicleDeployMap:19    '''20    Deploy data flat model.21    Iterable class that contains the data of YAML deploy files in22    VehicleDeployEntry objects.23    Parameters:24    -----------25        raw_data : dict26            Nested dictionary containing deploy data. It may be generated27            by `load_depl()`.28    Methods:29    --------30        get_entry(entry_name: str)31            Returns the content of the deploy node entry_name.32    '''33    _deploy_nodes: dict = {}34    def flat_deploy_model(  # noqa: C90135        self,36        depl_dict: dict,37        path: str,38        depl_data: Union[dict, list],39        is_list_item: bool = False40    ):41        if isinstance(depl_data, dict):42            if deploy_yaml_keys['constants'] in depl_data:43                self._handle_constant(path, depl_dict, depl_data)44            for key, value in depl_data.items():45                if key in deploy_yaml_keys.values():46                    continue47                entry_name = f'{path}_{key}' if path else key48                if is_list_item:49                    self._handle_list(entry_name, depl_dict, value)50                else:51                    depl_dict[entry_name] = VehicleDeployEntry(52                        entry_name,53                        value54                    )55                if (56                    key not in deploy_yaml_keys.values()57                    and (isinstance(value, list) or isinstance(value, dict))58                ):59                    self.flat_deploy_model(depl_dict,60                                           entry_name,61                                           value,62                                           is_list_item)63        elif isinstance(depl_data, list):64            for item in depl_data:65                for key in [k for k in item.keys()66                            if k not in deploy_yaml_keys.values()]:67                    entry_name = f'{path}_{key}'68                    if not depl_dict.get(entry_name):69                        depl_dict[f'{path}_{key}'] = []70                self.flat_deploy_model(depl_dict, path, item, True)71    def _handle_constant(self,72                         entry_name: str,73                         depl_dict: Dict,74                         deploy_data: Dict) -> None:75        depl_dict[entry_name] = VehicleDeployEntry(entry_name,76                                                   deploy_data)77        constants = deploy_data[deploy_yaml_keys['constants']]78        if isinstance(constants, dict):79            for key, value in constants[next(iter(constants))].items():80                constant_entry_name = f'{entry_name}_{key}'81                depl_dict[constant_entry_name] = VehicleDeployEntry(82                    constant_entry_name,83                    value84                )85        if isinstance(constants, list):86            for key, value in constants[0]:87                constant_entry_name = f'{entry_name}_{key}'88                depl_dict[constant_entry_name] = VehicleDeployEntry(89                    constant_entry_name,90                    value91                )92    def _handle_list(self,93                     entry_name: str,94                     depl_dict: Dict,95                     value: Any):96        if entry_name not in depl_dict:97            depl_dict[entry_name] = []98        depl_dict[entry_name].append(99            VehicleDeployEntry(100                entry_name,101                value102            )103        )104    def get(self, entry_name: str) -> Optional[VehicleDeployEntry]:105        return self._deploy_nodes.get(entry_name)106    # REMOVE ME: contains + get_entry is a bad pattern, double lookup107    def get_entry(self, entry_name: str) -> VehicleDeployEntry:108        return self._deploy_nodes[f'{entry_name}']109    def contains(self, entry_name: str) -> bool:110        return entry_name in self._deploy_nodes111    def __iter__(self):112        return iter(self._deploy_nodes.items())113    def __init__(self, raw_data: dict):114        self.flat_deploy_model(self._deploy_nodes, '', raw_data)115    def __repr__(self) -> str:116        keyList = []117        for key in self._deploy_nodes.keys():118            keyList.append(key)119        keyList.sort()120        return '\n'.join(121            [f'{repr(self._deploy_nodes[key])}' for key in keyList]...

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.