fix(opensearch): add support for admin_password in >= 2.12 (#697)

Opensearch has made an initial admin password mandatory since version
2.12 as can be read in the
[changelogs](https://github.com/opensearch-project/opensearch-build/blob/main/release-notes/opensearch-release-notes-2.12.0.md).

The open search module in test containers-python does not pass in this
additional flag and current compatibility with open search >= 2.12 is
broken .

This MR adds compatibility with open search >= 2.12 by adding the
inidial_admin_password flag to the docker environment

---------

Co-authored-by: David Ankin <daveankin@gmail.com>

modules/opensearch/testcontainers/opensearch/__init__.py

@@ -1,3 +1,5 @@
+from contextlib import suppress
+
 from opensearchpy import OpenSearch
 from opensearchpy.exceptions import ConnectionError, TransportError
 from urllib3.exceptions import ProtocolError
@@ -35,25 +37,37 @@ def __init__(
         image: str = "opensearchproject/opensearch:2.4.0",
         port: int = 9200,
         security_enabled: bool = False,
+        initial_admin_password: str = "admin",
         **kwargs,
     ) -> None:
         """
         Args:
             image: Docker image to use for the container.
             port: Port to expose on the container.
             security_enabled: :code:`False` disables the security plugin in OpenSearch.
+            initial_admin_password: set the password for opensearch, For OpenSearch versions 2.12 and
+                later, you must set the initial admin password as seen in the documentation,
+                https://opensearch.org/docs/latest/security/configuration/demo-configuration/#setting-up-a-custom-admin-password
         """
         raise_for_deprecated_parameter(kwargs, "port_to_expose", "port")
         super().__init__(image, **kwargs)
         self.port = port
         self.security_enabled = security_enabled
+        self.initial_admin_password = initial_admin_password
 
         self.with_exposed_ports(self.port)
         self.with_env("discovery.type", "single-node")
         self.with_env("plugins.security.disabled", "false" if security_enabled else "true")
+        if self._supports_initial_admin_password(str(image)):
+            self.with_env("OPENSEARCH_INITIAL_ADMIN_PASSWORD", self.initial_admin_password)
         if security_enabled:
             self.with_env("plugins.security.allow_default_init_securityindex", "true")
 
+    def _supports_initial_admin_password(self, image: str) -> bool:
+        with suppress(Exception):
+            return [int(n) for n in image.split(":")[-1].split(".")] >= [int(n) for n in "2.12.0".split(".")]
+        return False
+
     def get_config(self) -> dict:
         """This method returns the configuration of the OpenSearch container,
         including the host, port, username, and password.
@@ -66,7 +80,7 @@ def get_config(self) -> dict:
             "host": self.get_container_host_ip(),
             "port": self.get_exposed_port(self.port),
             "username": "admin",
-            "password": "admin",
+            "password": self.initial_admin_password,
         }
 
     def get_client(self, verify_certs: bool = False, **kwargs) -> OpenSearch:

modules/opensearch/tests/test_opensearch.py

@@ -1,5 +1,20 @@
 from testcontainers.opensearch import OpenSearchContainer
 
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def disable_logging():
+    import logging
+    import warnings
+
+    warnings.filterwarnings("ignore")
+    logging.getLogger("opensearch").setLevel(logging.CRITICAL)
+
+    yield
+    warnings.resetwarnings()
+    logging.getLogger("opensearch").setLevel(logging.NOTSET)
+
 
 def test_docker_run_opensearch():
     with OpenSearchContainer() as opensearch:
@@ -25,6 +40,14 @@ def test_docker_run_opensearch_v1_with_security():
         assert client.cluster.health()["status"] == "green"
 
 
+def test_docker_run_opensearch_v2_12():
+    with OpenSearchContainer(
+        image="opensearchproject/opensearch:2.12.0", initial_admin_password="Testing!#345"
+    ) as opensearch:
+        client = opensearch.get_client()
+        assert client.cluster.health()["status"] == "green"
+
+
 def test_search():
     with OpenSearchContainer() as opensearch:
         client = opensearch.get_client()