How to use _backend_cluster method in localstack

Best Python code snippet using localstack_python

cluster.py

Source:cluster.py

...229        return super().is_up()230    def health(self):231        """calls the health endpoint of cluster through the proxy, making sure implicitly that both are running"""232        return get_cluster_health_status(self.url)233    def _backend_cluster(self) -> OpensearchCluster:234        return OpensearchCluster(235            port=self.cluster_port,236            host=DEFAULT_BACKEND_HOST,237            version=self.version,238            directories=self.directories,239        )240    def do_run(self):241        self.cluster_port = get_free_tcp_port()242        self.cluster = self._backend_cluster()243        self.cluster.start()244        self.proxy = EndpointProxy(self.url, self.cluster.url)245        LOG.info("registering an endpoint proxy for %s => %s", self.url, self.cluster.url)246        self.proxy.register()247        self.cluster.wait_is_up()248        LOG.info("cluster on %s is ready", self.cluster.url)249        return self.cluster.join()250    def do_shutdown(self):251        if self.proxy:252            self.proxy.unregister()253        if self.cluster:254            self.cluster.shutdown()255class ElasticsearchCluster(OpensearchCluster):256    @property257    def default_version(self) -> str:258        return constants.ELASTICSEARCH_DEFAULT_VERSION259    @property260    def bin_name(self) -> str:261        return "elasticsearch"262    @property263    def os_user(self):264        return constants.OS_USER_OPENSEARCH265    def _ensure_installed(self):266        install.install_elasticsearch(self.version)267    def _base_settings(self, dirs) -> CommandSettings:268        settings = {269            "http.port": self.port,270            "http.publish_port": self.port,271            "transport.port": "0",272            "network.host": self.host,273            "http.compression": "false",274            "path.data": f'"{dirs.data}"',275            "path.repo": f'"{dirs.backup}"',276        }277        if os.path.exists(os.path.join(dirs.mods, "x-pack-ml")):278            settings["xpack.ml.enabled"] = "false"279        return settings280    def _create_env_vars(self) -> Dict:281        return {282            "ES_JAVA_OPTS": os.environ.get("ES_JAVA_OPTS", "-Xms200m -Xmx600m"),283            "ES_TMPDIR": self.directories.tmp,284        }285class EdgeProxiedElasticsearchCluster(EdgeProxiedOpensearchCluster):286    @property287    def default_version(self):288        # TODO move to constants289        return "Elasticsearch_7.10"290    def _backend_cluster(self) -> OpensearchCluster:291        return ElasticsearchCluster(292            port=self.cluster_port,293            host=DEFAULT_BACKEND_HOST,294            version=self.version,295            directories=self.directories,...

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.