How to use _clusterconfig_to_opensearch method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py Github

copy

Full Screen

...117 )118 result.pop("EngineVersion", None)119 result.pop("ClusterConfig", None)120 return result121def _clusterconfig_to_opensearch(122 elasticsearch_cluster_config: Optional[ElasticsearchClusterConfig],123) -> Optional[ClusterConfig]:124 if elasticsearch_cluster_config is not None:125 result = cast(ClusterConfig, elasticsearch_cluster_config)126 result["InstanceType"] = _instancetype_to_opensearch(result.get("InstanceType"))127 result["DedicatedMasterType"] = _instancetype_to_opensearch(128 result.get("DedicatedMasterType")129 )130 result["WarmType"] = _instancetype_to_opensearch(result.get("WarmType"))131 return result132def _domainconfig_from_opensearch(133 domain_config: Optional[DomainConfig],134) -> Optional[ElasticsearchDomainConfig]:135 if domain_config is not None:136 result = cast(ElasticsearchDomainConfig, domain_config)137 engine_version = domain_config.get("EngineVersion", {})138 result["ElasticsearchVersion"] = ElasticsearchVersionStatus(139 Options=_version_from_opensearch(engine_version.get("Options")),140 Status=cast(OptionStatus, engine_version.get("Status")),141 )142 cluster_config = domain_config.get("ClusterConfig", {})143 result["ElasticsearchClusterConfig"] = ElasticsearchClusterConfigStatus(144 Options=_clusterconfig_from_opensearch(cluster_config.get("Options")),145 Status=cluster_config.get("Status"),146 )147 result.pop("EngineVersion", None)148 result.pop("ClusterConfig", None)149 return result150def _compatible_version_list_from_opensearch(151 compatible_version_list: Optional[CompatibleVersionsList],152) -> Optional[CompatibleElasticsearchVersionsList]:153 if compatible_version_list is not None:154 return [155 CompatibleVersionsMap(156 SourceVersion=_version_from_opensearch(version_map["SourceVersion"]),157 TargetVersions=[158 _version_from_opensearch(target_version)159 for target_version in version_map["TargetVersions"]160 ],161 )162 for version_map in compatible_version_list163 ]164@contextmanager165def exception_mapper():166 """Maps an exception thrown by the OpenSearch client to an exception thrown by the ElasticSearch API."""167 try:168 yield169 except ClientError as err:170 exception_types = {171 "AccessDeniedException": AccessDeniedException,172 "BaseException": EsBaseException,173 "ConflictException": ConflictException,174 "DisabledOperationException": DisabledOperationException,175 "InternalException": InternalException,176 "InvalidPaginationTokenException": InvalidPaginationTokenException,177 "InvalidTypeException": InvalidTypeException,178 "LimitExceededException": LimitExceededException,179 "ResourceAlreadyExistsException": ResourceAlreadyExistsException,180 "ResourceNotFoundException": ResourceNotFoundException,181 "ValidationException": ValidationException,182 }183 mapped_exception_type = exception_types.get(err.response["Error"]["Code"], EsBaseException)184 raise mapped_exception_type(err.response["Error"]["Message"])185class EsProvider(EsApi):186 def create_elasticsearch_domain(187 self,188 context: RequestContext,189 domain_name: DomainName,190 elasticsearch_version: ElasticsearchVersionString = None,191 elasticsearch_cluster_config: ElasticsearchClusterConfig = None,192 ebs_options: EBSOptions = None,193 access_policies: PolicyDocument = None,194 snapshot_options: SnapshotOptions = None,195 vpc_options: VPCOptions = None,196 cognito_options: CognitoOptions = None,197 encryption_at_rest_options: EncryptionAtRestOptions = None,198 node_to_node_encryption_options: NodeToNodeEncryptionOptions = None,199 advanced_options: AdvancedOptions = None,200 log_publishing_options: LogPublishingOptions = None,201 domain_endpoint_options: DomainEndpointOptions = None,202 advanced_security_options: AdvancedSecurityOptionsInput = None,203 auto_tune_options: AutoTuneOptionsInput = None,204 tag_list: TagList = None,205 ) -> CreateElasticsearchDomainResponse:206 opensearch_client = aws_stack.connect_to_service("opensearch", region_name=context.region)207 # If no version is given, we set our default elasticsearch version208 engine_version = (209 _version_to_opensearch(elasticsearch_version)210 if elasticsearch_version211 else constants.ELASTICSEARCH_DEFAULT_VERSION212 )213 kwargs = {214 "DomainName": domain_name,215 "EngineVersion": engine_version,216 "ClusterConfig": _clusterconfig_to_opensearch(elasticsearch_cluster_config),217 "EBSOptions": ebs_options,218 "AccessPolicies": access_policies,219 "SnapshotOptions": snapshot_options,220 "VPCOptions": vpc_options,221 "CognitoOptions": cognito_options,222 "EncryptionAtRestOptions": encryption_at_rest_options,223 "NodeToNodeEncryptionOptions": node_to_node_encryption_options,224 "AdvancedOptions": advanced_options,225 "LogPublishingOptions": log_publishing_options,226 "DomainEndpointOptions": domain_endpoint_options,227 "AdvancedSecurityOptions": advanced_security_options,228 "AutoTuneOptions": auto_tune_options,229 "TagList": tag_list,230 }...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful