How to use opensearch_domain method in localstack

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            ]...

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.