How to use get_install_type_and_version method in localstack

Best Python code snippet using localstack_python

cluster.py

Source:cluster.py

...65    :param data_root: the root of the data dir (will be resolved to TMP_PATH or DATA_DIR by default)66    :returns: a Directories data structure67    """68    # where to find cluster binary and the modules69    engine_type, install_version = versions.get_install_type_and_version(version)70    if engine_type == EngineType.OpenSearch:71        install_dir = install.get_opensearch_install_dir(install_version)72    else:73        # Elasticsearch version74        install_dir = install.get_elasticsearch_install_dir(install_version)75    modules_dir = os.path.join(install_dir, "modules")76    if not data_root:77        data_root = config.dirs.data or config.dirs.tmp78    if engine_type == EngineType.OpenSearch:79        data_path = os.path.join(data_root, "opensearch", cluster_path)80    else:81        data_path = os.path.join(data_root, "elasticsearch", cluster_path)82    tmp_dir = os.path.join(data_path, "tmp")83    data_dir = os.path.join(data_path, "data")84    backup_dir = os.path.join(data_path, "backup")85    return Directories(install_dir, tmp_dir, modules_dir, data_dir, backup_dir)86def build_cluster_run_command(cluster_bin: str, settings: CommandSettings) -> List[str]:87    """88    Takes the command settings dict and builds the actual command (which can then be executed as a shell command).89    :param cluster_bin: path to the OpenSearch/Elasticsearch binary (including the binary)90    :param settings: dictionary where each item will be set as a command arguments91    :return: list of strings for the command with the settings to be executed as a shell command92    """93    cmd_settings = [f"-E {k}={v}" for k, v, in settings.items()]94    return [cluster_bin] + cmd_settings95class OpensearchCluster(Server):96    """Manages an OpenSearch cluster which is installed an operated by LocalStack."""97    # TODO: legacy default port should be removed here98    def __init__(99        self, port=4571, host="localhost", version: str = None, directories: Directories = None100    ) -> None:101        super().__init__(port, host)102        self._version = version or self.default_version103        self.command_settings = {}104        self.directories = directories or self._resolve_directories()105    @property106    def default_version(self) -> str:107        return constants.OPENSEARCH_DEFAULT_VERSION108    @property109    def version(self) -> str:110        return self._version111    @property112    def install_version(self) -> str:113        _, install_version = versions.get_install_type_and_version(self._version)114        return install_version115    @property116    def bin_name(self) -> str:117        return "opensearch"118    @property119    def os_user(self):120        return constants.OS_USER_OPENSEARCH121    def health(self) -> Optional[str]:122        return get_cluster_health_status(self.url)123    def do_start_thread(self) -> FuncThread:124        self._ensure_installed()125        self._init_directories()126        cmd = self._create_run_command(additional_settings=self.command_settings)127        cmd = " ".join(cmd)...

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.