Opensearch: Add the list_domain_names method support (#6824)

docs/docs/services/opensearch.rst

@@ -61,7 +61,7 @@ opensearch
 - [ ] get_package_version_history
 - [ ] get_upgrade_history
 - [ ] get_upgrade_status
-- [ ] list_domain_names
+- [x] list_domain_names
 - [ ] list_domains_for_package
 - [ ] list_instance_type_details
 - [ ] list_packages_for_domain

moto/es/urls.py

@@ -10,6 +10,7 @@ url_paths = {
     "{0}/2015-01-01/domain$": ElasticsearchServiceResponse.list_domains,
     "{0}/2015-01-01/es/domain$": ElasticsearchServiceResponse.domains,
     "{0}/2015-01-01/es/domain/(?P<domainname>[^/]+)": ElasticsearchServiceResponse.domain,
+    "{0}/2021-01-01/domain$": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/compatibleVersions": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/domain": OpenSearchServiceResponse.dispatch,
     "{0}/2021-01-01/opensearch/domain/(?P<domainname>[^/]+)": OpenSearchServiceResponse.dispatch,

moto/opensearch/exceptions.py

@@ -5,3 +5,10 @@ from moto.core.exceptions import JsonRESTError
 class ResourceNotFoundException(JsonRESTError):
     def __init__(self, name: str):
         super().__init__("ResourceNotFoundException", f"Domain not found: {name}")
+
+
+class EngineTypeNotFoundException(JsonRESTError):
+    def __init__(self, domain_name: str):
+        super().__init__(
+            "EngineTypeNotFoundException", f"Engine Type not found: {domain_name}"
+        )

moto/opensearch/models.py

@@ -2,9 +2,9 @@ from typing import Any, Dict, List, Optional
 
 from moto.core import BaseBackend, BackendDict, BaseModel
 from moto.utilities.tagging_service import TaggingService
-from .data import compatible_versions
-from .exceptions import ResourceNotFoundException
 
+from .data import compatible_versions
+from .exceptions import ResourceNotFoundException, EngineTypeNotFoundException
 
 default_cluster_config = {
     "InstanceType": "t3.small.search",
@@ -327,5 +327,27 @@ class OpenSearchServiceBackend(BaseBackend):
     def remove_tags(self, arn: str, tag_keys: List[str]) -> None:
         self.tagger.untag_resource_using_names(arn, tag_keys)
 
+    def list_domain_names(self, engine_type: str) -> List[Dict[str, str]]:
+        domains = []
+        for domain in self.domains.values():
+            if engine_type:
+                if engine_type in domain.engine_version:
+                    domains.append(
+                        {
+                            "DomainName": domain.domain_name,
+                            "EngineType": engine_type.split("_")[0],
+                        }
+                    )
+                else:
+                    raise EngineTypeNotFoundException(domain.domain_name)
+            else:
+                domains.append(
+                    {
+                        "DomainName": domain.domain_name,
+                        "EngineType": domain.engine_version.split("_")[0],
+                    }
+                )
+        return domains
+
 
 opensearch_backends = BackendDict(OpenSearchServiceBackend, "opensearch")

moto/opensearch/responses.py

@@ -2,6 +2,7 @@
 import json
 
 from moto.core.responses import BaseResponse
+
 from .models import opensearch_backends, OpenSearchServiceBackend
 
 
@@ -138,3 +139,10 @@ class OpenSearchServiceResponse(BaseResponse):
         tag_keys = self._get_param("TagKeys")
         self.opensearch_backend.remove_tags(arn, tag_keys)
         return "{}"
+
+    def list_domain_names(self) -> str:
+        engine_type = self._get_param("engineType")
+        domain_names = self.opensearch_backend.list_domain_names(
+            engine_type=engine_type,
+        )
+        return json.dumps(dict(DomainNames=domain_names))

tests/test_opensearch/test_opensearch.py

@@ -1,9 +1,10 @@
 import boto3
 import pytest
-
 from botocore.exceptions import ClientError
+
 from moto import mock_opensearch
 
+
 # See our Development Tips on writing tests for hints on how to write good tests:
 # http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
 
@@ -139,3 +140,67 @@ def test_update_domain_config():
             "CustomEndpointEnabled": False,
         }
     }
+
+
+@mock_opensearch
+def test_list_domain_names():
+    client = boto3.client("opensearch", region_name="ap-southeast-1")
+
+    test_domain_names_list_exist = False
+
+    opensearch_domain_name = "testdn"
+    opensearch_engine_version = "OpenSearch_1.0"
+    client.create_domain(
+        DomainName=opensearch_domain_name, EngineVersion=opensearch_engine_version
+    )
+
+    resp = client.list_domain_names()
+    domain_names = resp["DomainNames"]
+
+    for domain_name in domain_names:
+        if domain_name["DomainName"] == opensearch_domain_name:
+            test_domain_names_list_exist = True
+
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert test_domain_names_list_exist
+
+
+@mock_opensearch
+def test_list_filtered_domain_names():
+    client = boto3.client("opensearch", region_name="ap-southeast-1")
+
+    test_domain_names_list_exist = False
+
+    opensearch_domain_name = "testdn"
+    opensearch_engine_version = "OpenSearch_1.0"
+    client.create_domain(
+        DomainName=opensearch_domain_name, EngineVersion=opensearch_engine_version
+    )
+
+    resp = client.list_domain_names(EngineType="OpenSearch")
+    domain_names = resp["DomainNames"]
+
+    for domain_name in domain_names:
+        if domain_name["DomainName"] == opensearch_domain_name:
+            if domain_name["EngineType"] == opensearch_engine_version.split("_")[0]:
+                test_domain_names_list_exist = True
+
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert test_domain_names_list_exist
+
+
+@mock_opensearch
+def test_list_unknown_domain_names_engine_type():
+    client = boto3.client("opensearch", region_name="ap-southeast-1")
+
+    opensearch_domain_name = "testdn"
+    opensearch_engine_version = "OpenSearch_1.0"
+    client.create_domain(
+        DomainName=opensearch_domain_name, EngineVersion=opensearch_engine_version
+    )
+
+    with pytest.raises(ClientError) as exc:
+        client.list_domain_names(EngineType="unknown")
+    err = exc.value.response["Error"]
+    assert err["Code"] == "EngineTypeNotFoundException"
+    assert err["Message"] == "Engine Type not found: testdn"