Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _resolve_directories method in localstack

Best Python code snippet using localstack_python

cluster.py

Source:cluster.py

...50    def __init__(self, port=9200, host="localhost", version=None) -> None:51        super().__init__(port, host)52        self._version = version or constants.ELASTICSEARCH_DEFAULT_VERSION53        self.command_settings = {}54        self.directories = self._resolve_directories()55    @property56    def version(self):57        return self._version58    def health(self):59        return get_elasticsearch_health_status(self.url)60    def do_start_thread(self) -> FuncThread:61        # FIXME: if this fails the cluster could be left in a wonky state62        # FIXME: this is not a good place to run install, and it only works because we're63        #  assuming that there will only ever be one running Elasticsearch cluster64        install.install_elasticsearch(self.version)65        self._init_directories()66        cmd = self._create_run_command(additional_settings=self.command_settings)67        cmd = " ".join(cmd)68        user = constants.OS_USER_ELASTICSEARCH69        if is_root() and user:70            # run the elasticsearch process as a non-root user (when running in docker)71            cmd = f"su {user} -c '{cmd}'"72        env_vars = self._create_env_vars()73        LOG.info("starting elasticsearch: %s with env %s", cmd, env_vars)74        t = ShellCommandThread(75            cmd,76            env_vars=env_vars,77            strip_color=True,78            log_listener=self._log_listener,79        )80        t.start()81        return t82    def _log_listener(self, line, **_kwargs):83        LOG.info(line.rstrip())84    def _create_run_command(85        self, additional_settings: Optional[CommandSettings] = None86    ) -> List[str]:87        # delete Elasticsearch data that may be cached locally from a previous test run88        dirs = self.directories89        bin_path = os.path.join(dirs.base, "bin/elasticsearch")90        # build command settings for bin/elasticsearch91        settings = {92            "http.port": self.port,93            "http.publish_port": self.port,94            "transport.port": "0",95            "network.host": self.host,96            "http.compression": "false",97            "path.data": f'"{dirs.data}"',98            "path.repo": f'"{dirs.backup}"',99        }100        if os.path.exists(os.path.join(dirs.mods, "x-pack-ml")):101            settings["xpack.ml.enabled"] = "false"102        if additional_settings:103            settings.update(additional_settings)104        cmd = _build_elasticsearch_run_command(bin_path, settings)105        return cmd106    def _create_env_vars(self) -> Dict:107        return {108            "ES_JAVA_OPTS": os.environ.get("ES_JAVA_OPTS", "-Xms200m -Xmx600m"),109            "ES_TMPDIR": self.directories.tmp,110        }111    def _resolve_directories(self) -> Directories:112        # determine various directory paths113        base_dir = install.get_elasticsearch_install_dir(self.version)114        es_tmp_dir = os.path.join(base_dir, "tmp")115        es_mods_dir = os.path.join(base_dir, "modules")116        if config.DATA_DIR:117            es_data_dir = os.path.join(config.DATA_DIR, "elasticsearch")118        else:119            es_data_dir = os.path.join(base_dir, "data")120        backup_dir = os.path.join(config.TMP_FOLDER, "es_backup")121        return Directories(base_dir, es_tmp_dir, es_mods_dir, es_data_dir, backup_dir)122    def _init_directories(self):123        dirs = self.directories124        LOG.debug("initializing elasticsearch directories %s", dirs)125        chmod_r(dirs.base, 0o777)...

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.