How to use determine_custom_endpoint method in localstack

Best Python code snippet using localstack_python

cluster_manager.py

Source:cluster_manager.py

...98    if config.OPENSEARCH_ENDPOINT_STRATEGY == "path":99        return f"{config.LOCALSTACK_HOSTNAME}:{config.EDGE_PORT}/{engine_domain}/{domain_key.region}/{domain_key.domain_name}"100    # or through a subdomain (domain-name.region.opensearch.localhost.localstack.cloud)101    return f"{domain_key.domain_name}.{domain_key.region}.{engine_domain}.{LOCALHOST_HOSTNAME}:{config.EDGE_PORT}"102def determine_custom_endpoint(103    domain_endpoint_options: DomainEndpointOptions,104) -> Optional[CustomEndpoint]:105    if not domain_endpoint_options:106        return107    custom_endpoint = domain_endpoint_options.get("CustomEndpoint")108    enabled = domain_endpoint_options.get("CustomEndpointEnabled", False)109    if not custom_endpoint:110        # No custom endpoint to determine111        return112    return CustomEndpoint(enabled, custom_endpoint)113class ClusterManager:114    clusters: Dict[str, Server]115    def __init__(self) -> None:116        self.clusters = {}117    def create(118        self,119        arn: str,120        version: str,121        endpoint_options: Optional[DomainEndpointOptions] = None,122        preferred_port: Optional[int] = None,123    ) -> Server:124        """125        Creates a new cluster.126        :param arn: of the cluster to create127        :param version: of the cluster to start (string including the EngineType)128        :param endpoint_options: DomainEndpointOptions (may contain information about a custom endpoint url)129        :param preferred_port: port which should be preferred (only if OPENSEARCH_ENDPOINT_STRATEGY == "port")130        :return: None131        """132        # determine custom domain endpoint133        custom_endpoint = determine_custom_endpoint(endpoint_options)134        # determine engine type135        engine_type = versions.get_engine_type(version)136        # build final endpoint and cluster url137        endpoint = build_cluster_endpoint(138            DomainKey.from_arn(arn), custom_endpoint, engine_type, preferred_port139        )140        url = f"http://{endpoint}" if "://" not in endpoint else endpoint141        # call abstract cluster factory142        cluster = self._create_cluster(arn, url, version)143        cluster.start()144        # save cluster into registry and return145        self.clusters[arn] = cluster146        return cluster147    def get(self, arn: str) -> Optional[Server]:...

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.