How to use opensearch_domain method in localstack

Best Python code snippet using localstack_python

test_es.py

Source:test_es.py Github

copy

Full Screen

1import logging2import threading3import botocore.exceptions4import pytest5from localstack import config6from localstack.constants import ELASTICSEARCH_DEFAULT_VERSION, OPENSEARCH_DEFAULT_VERSION7from localstack.services.install import install_elasticsearch, install_opensearch8from localstack.utils.common import safe_requests as requests9from localstack.utils.common import short_uid, start_worker_thread10LOG = logging.getLogger(__name__)11# Common headers used when sending requests to OpenSearch12COMMON_HEADERS = {"content-type": "application/json", "Accept-encoding": "identity"}13# Lock and event to ensure that the installation is executed before the tests14INIT_LOCK = threading.Lock()15installed = threading.Event()16def install_async():17 """18 Installs the default elasticsearch version in a worker thread. Used by conftest.py to make19 sure elasticsearch is downloaded once the tests arrive here.20 """21 if installed.is_set():22 return23 def run_install(*args):24 with INIT_LOCK:25 if installed.is_set():26 return27 LOG.info("installing elasticsearch default version")28 install_elasticsearch()29 LOG.info("done installing elasticsearch default version")30 LOG.info("installing opensearch default version")31 install_opensearch()32 LOG.info("done installing opensearch default version")33 installed.set()34 start_worker_thread(run_install)35@pytest.fixture(autouse=True)36def elasticsearch():37 if not installed.is_set():38 install_async()39 assert installed.wait(timeout=5 * 60), "gave up waiting for elasticsearch to install"40 yield41def try_cluster_health(cluster_url: str):42 response = requests.get(cluster_url)43 assert response.ok, f"cluster endpoint returned an error: {response.text}"44 response = requests.get(f"{cluster_url}/_cluster/health")45 assert response.ok, f"cluster health endpoint returned an error: {response.text}"46 assert response.json()["status"] in [47 "orange",48 "yellow",49 "green",50 ], "expected cluster state to be in a valid state"51class TestElasticsearchProvider:52 def test_list_versions(self, es_client):53 response = es_client.list_elasticsearch_versions()54 assert "ElasticsearchVersions" in response55 versions = response["ElasticsearchVersions"]56 assert "OpenSearch_1.0" in versions57 assert "OpenSearch_1.1" in versions58 assert "7.10" in versions59 def test_get_compatible_versions(self, es_client):60 response = es_client.get_compatible_elasticsearch_versions()61 assert "CompatibleElasticsearchVersions" in response62 versions = response["CompatibleElasticsearchVersions"]63 assert len(versions) == 1864 assert {"SourceVersion": "OpenSearch_1.0", "TargetVersions": ["OpenSearch_1.1"]} in versions65 assert {66 "SourceVersion": "7.10",67 "TargetVersions": ["OpenSearch_1.0", "OpenSearch_1.1"],68 } in versions69 assert {70 "SourceVersion": "7.7",71 "TargetVersions": ["7.8", "7.9", "7.10", "OpenSearch_1.0", "OpenSearch_1.1"],72 } in versions73 @pytest.mark.skip_offline74 def test_get_compatible_version_for_domain(self, es_client, opensearch_domain):75 response = es_client.get_compatible_elasticsearch_versions(DomainName=opensearch_domain)76 assert "CompatibleElasticsearchVersions" in response77 versions = response["CompatibleElasticsearchVersions"]78 # The default version is the latest version, which is not compatible with any previous versions79 assert len(versions) == 080 @pytest.mark.skip_offline81 def test_create_domain(self, es_client, opensearch_create_domain):82 es_domain = opensearch_create_domain(EngineVersion=ELASTICSEARCH_DEFAULT_VERSION)83 response = es_client.list_domain_names(EngineType="Elasticsearch")84 domain_names = [domain["DomainName"] for domain in response["DomainNames"]]85 assert es_domain in domain_names86 @pytest.mark.skip_offline87 def test_create_existing_domain_causes_exception(self, es_client, opensearch_create_domain):88 domain_name = opensearch_create_domain(EngineVersion=ELASTICSEARCH_DEFAULT_VERSION)89 with pytest.raises(botocore.exceptions.ClientError) as exc_info:90 es_client.create_elasticsearch_domain(DomainName=domain_name)91 assert exc_info.type.__name__ == "ResourceAlreadyExistsException"92 @pytest.mark.skip_offline93 def test_describe_domains(self, es_client, opensearch_create_domain):94 opensearch_domain = opensearch_create_domain(EngineVersion=ELASTICSEARCH_DEFAULT_VERSION)95 response = es_client.describe_elasticsearch_domains(DomainNames=[opensearch_domain])96 assert len(response["DomainStatusList"]) == 197 assert response["DomainStatusList"][0]["DomainName"] == opensearch_domain98 @pytest.mark.skip_offline99 def test_domain_version(self, es_client, opensearch_domain, opensearch_create_domain):100 response = es_client.describe_elasticsearch_domain(DomainName=opensearch_domain)101 assert "DomainStatus" in response102 status = response["DomainStatus"]103 assert "ElasticsearchVersion" in status104 assert status["ElasticsearchVersion"] == OPENSEARCH_DEFAULT_VERSION105 domain_name = opensearch_create_domain(EngineVersion=ELASTICSEARCH_DEFAULT_VERSION)106 response = es_client.describe_elasticsearch_domain(DomainName=domain_name)107 assert "DomainStatus" in response108 status = response["DomainStatus"]109 assert "ElasticsearchVersion" in status110 assert status["ElasticsearchVersion"] == "7.10"111 @pytest.mark.skip_offline112 def test_path_endpoint_strategy(self, monkeypatch, opensearch_create_domain, es_client):113 monkeypatch.setattr(config, "OPENSEARCH_ENDPOINT_STRATEGY", "path")114 monkeypatch.setattr(config, "OPENSEARCH_MULTI_CLUSTER", True)115 domain_name = f"es-domain-{short_uid()}"116 opensearch_create_domain(DomainName=domain_name)117 status = es_client.describe_elasticsearch_domain(DomainName=domain_name)["DomainStatus"]118 assert "Endpoint" in status119 endpoint = status["Endpoint"]120 assert endpoint.endswith(f"/{domain_name}")121 def test_update_domain_config(self, es_client, opensearch_domain):122 initial_response = es_client.describe_elasticsearch_domain_config(123 DomainName=opensearch_domain124 )125 update_response = es_client.update_elasticsearch_domain_config(126 DomainName=opensearch_domain,127 ElasticsearchClusterConfig={"InstanceType": "r4.16xlarge.elasticsearch"},128 )129 final_response = es_client.describe_elasticsearch_domain_config(130 DomainName=opensearch_domain131 )132 assert (133 initial_response["DomainConfig"]["ElasticsearchClusterConfig"]["Options"][134 "InstanceType"135 ]136 != update_response["DomainConfig"]["ElasticsearchClusterConfig"]["Options"][137 "InstanceType"138 ]139 )140 assert (141 update_response["DomainConfig"]["ElasticsearchClusterConfig"]["Options"]["InstanceType"]142 == "r4.16xlarge.elasticsearch"143 )144 assert (145 update_response["DomainConfig"]["ElasticsearchClusterConfig"]["Options"]["InstanceType"]146 == final_response["DomainConfig"]["ElasticsearchClusterConfig"]["Options"][147 "InstanceType"148 ]...

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