Best Python code snippet using localstack_python
test_es.py
Source:test_es.py
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 ]...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!