FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

fix(opensearch): add support for admin_password in >= 2.12 by Rahul-Anil · Pull Request #697 · testcontainers/testcontainers-python · GitHub

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (2) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from contextlib import suppress

from opensearchpy import OpenSearch
from opensearchpy.exceptions import ConnectionError, TransportError
from urllib3.exceptions import ProtocolError
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand Down
23 changes: 23 additions & 0 deletions modules/opensearch/tests/test_opensearch.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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:
Expand All @@ -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()
Expand Down

Back | FazBrowse Home | New Git URL