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

How to use _ensure_domain_exists method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...322    )323    if stored_status.get("Endpoint"):324        new_status["Endpoint"] = new_status.get("Endpoint")325    return new_status326def _ensure_domain_exists(arn: ARN) -> None:327    """328    Checks if the domain for the given ARN exists. Otherwise, a ValidationException is raised.329    :param arn: ARN string to lookup the domain for330    :return: None if the domain exists, otherwise raises an exception331    :raises: ValidationException if the domain for the given ARN cannot be found332    """333    domain_key = DomainKey.from_arn(arn)334    region = OpenSearchServiceBackend.get(domain_key.region)335    domain_status = region.opensearch_domains.get(domain_key.domain_name)336    if domain_status is None:337        raise ValidationException("Invalid ARN. Domain not found.")338class OpensearchProvider(OpensearchApi):339    def create_domain(340        self,341        context: RequestContext,342        domain_name: DomainName,343        engine_version: VersionString = None,344        cluster_config: ClusterConfig = None,345        ebs_options: EBSOptions = None,346        access_policies: PolicyDocument = None,347        snapshot_options: SnapshotOptions = None,348        vpc_options: VPCOptions = None,349        cognito_options: CognitoOptions = None,350        encryption_at_rest_options: EncryptionAtRestOptions = None,351        node_to_node_encryption_options: NodeToNodeEncryptionOptions = None,352        advanced_options: AdvancedOptions = None,353        log_publishing_options: LogPublishingOptions = None,354        domain_endpoint_options: DomainEndpointOptions = None,355        advanced_security_options: AdvancedSecurityOptionsInput = None,356        tag_list: TagList = None,357        auto_tune_options: AutoTuneOptionsInput = None,358    ) -> CreateDomainResponse:359        region = OpenSearchServiceBackend.get()360        with _domain_mutex:361            if domain_name in region.opensearch_domains:362                raise ResourceAlreadyExistsException(363                    f"domain {domain_name} already exists in region {region.name}"364                )365            domain_key = DomainKey(366                domain_name=domain_name,367                region=context.region,368                account=context.account_id,369            )370            # "create" domain data371            region.opensearch_domains[domain_name] = get_domain_status(domain_key)372            # lazy-init the cluster (sets the Endpoint and Processing flag of the domain status)373            # TODO handle additional parameters (cluster config,...)374            create_cluster(domain_key, engine_version, domain_endpoint_options)375            # set the tags376            self.add_tags(context, domain_key.arn, tag_list)377            # get the (updated) status378            status = get_domain_status(domain_key)379        # record event380        event_publisher.fire_event(381            event_publisher.EVENT_OPENSEARCH_CREATE_DOMAIN,382            payload={"n": event_publisher.get_hash(domain_name)},383        )384        return CreateDomainResponse(DomainStatus=status)385    def delete_domain(386        self, context: RequestContext, domain_name: DomainName387    ) -> DeleteDomainResponse:388        domain_key = DomainKey(389            domain_name=domain_name,390            region=context.region,391            account=context.account_id,392        )393        region = OpenSearchServiceBackend.get(domain_key.region)394        with _domain_mutex:395            if domain_name not in region.opensearch_domains:396                raise ResourceNotFoundException(f"Domain not found: {domain_name}")397            status = get_domain_status(domain_key, deleted=True)398            _remove_cluster(domain_key)399        # record event400        event_publisher.fire_event(401            event_publisher.EVENT_OPENSEARCH_DELETE_DOMAIN,402            payload={"n": event_publisher.get_hash(domain_name)},403        )404        return DeleteDomainResponse(DomainStatus=status)405    def describe_domain(406        self, context: RequestContext, domain_name: DomainName407    ) -> DescribeDomainResponse:408        domain_key = DomainKey(409            domain_name=domain_name,410            region=context.region,411            account=context.account_id,412        )413        region = OpenSearchServiceBackend.get(domain_key.region)414        with _domain_mutex:415            if domain_name not in region.opensearch_domains:416                raise ResourceNotFoundException(f"Domain not found: {domain_name}")417            status = get_domain_status(domain_key)418        return DescribeDomainResponse(DomainStatus=status)419    def describe_domains(420        self, context: RequestContext, domain_names: DomainNameList421    ) -> DescribeDomainsResponse:422        status_list = []423        with _domain_mutex:424            for domain_name in domain_names:425                try:426                    domain_status = self.describe_domain(context, domain_name)["DomainStatus"]427                    status_list.append(domain_status)428                except ResourceNotFoundException:429                    # ResourceNotFoundExceptions are ignored, we just look for the next domain.430                    # If no domain can be found, the result will just be empty.431                    pass432        return DescribeDomainsResponse(DomainStatusList=status_list)433    def list_domain_names(434        self, context: RequestContext, engine_type: EngineType = None435    ) -> ListDomainNamesResponse:436        region = OpenSearchServiceBackend.get(context.region)437        domain_names = [438            DomainInfo(439                DomainName=DomainName(domain_name),440                EngineType=versions.get_engine_type(domain["EngineVersion"]),441            )442            for domain_name, domain in region.opensearch_domains.items()443            if engine_type is None444            or versions.get_engine_type(domain["EngineVersion"]) == engine_type445        ]446        return ListDomainNamesResponse(DomainNames=domain_names)447    def list_versions(448        self,449        context: RequestContext,450        max_results: MaxResults = None,451        next_token: NextToken = None,452    ) -> ListVersionsResponse:453        return ListVersionsResponse(Versions=list(versions.install_versions.keys()))454    def get_compatible_versions(455        self, context: RequestContext, domain_name: DomainName = None456    ) -> GetCompatibleVersionsResponse:457        version_filter = None458        if domain_name:459            region = OpenSearchServiceBackend.get(context.region)460            with _domain_mutex:461                domain = region.opensearch_domains.get(domain_name)462                if not domain:463                    raise ResourceNotFoundException(f"Domain not found: {domain_name}")464                version_filter = domain.get("EngineVersion")465        compatible_versions = list(versions.compatible_versions)466        if version_filter is not None:467            compatible_versions = [468                comp469                for comp in versions.compatible_versions470                if comp["SourceVersion"] == version_filter471            ]472        return GetCompatibleVersionsResponse(CompatibleVersions=compatible_versions)473    def describe_domain_config(474        self, context: RequestContext, domain_name: DomainName475    ) -> DescribeDomainConfigResponse:476        domain_key = DomainKey(477            domain_name=domain_name,478            region=context.region,479            account=context.account_id,480        )481        region = OpenSearchServiceBackend.get(domain_key.region)482        with _domain_mutex:483            if domain_name not in region.opensearch_domains:484                raise ResourceNotFoundException(f"Domain not found: {domain_name}")485            domain_config = get_domain_config(domain_key)486        return DescribeDomainConfigResponse(DomainConfig=domain_config)487    def add_tags(self, context: RequestContext, arn: ARN, tag_list: TagList) -> None:488        _ensure_domain_exists(arn)489        OpenSearchServiceBackend.TAGS.tag_resource(arn, tag_list)490    def list_tags(self, context: RequestContext, arn: ARN) -> ListTagsResponse:491        _ensure_domain_exists(arn)492        # The tagging service returns a dictionary with the given root name493        tags = OpenSearchServiceBackend.TAGS.list_tags_for_resource(arn=arn, root_name="root")494        # Extract the actual list of tags for the typed response495        tag_list: TagList = tags["root"]496        return ListTagsResponse(TagList=tag_list)497    def remove_tags(self, context: RequestContext, arn: ARN, tag_keys: StringList) -> None:498        _ensure_domain_exists(arn)...

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.