diff --git a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx index b18a0839..c863997d 100644 --- a/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx +++ b/docs-site/src/content/docs/lexical-graph/vector-store-opensearch-serverless.mdx @@ -6,9 +6,10 @@ title: OpenSearch Serverless Vector Store - [Overview](#overview) - [Install dependencies](#install-dependencies) - - [Creating an OpenSearch Serverless vector store](#creating-an-opensearch-serverless-vector-store) + - [Creating an Amazon OpenSearch Serverless vector store](#creating-an-opensearch-serverless-vector-store) - [Amazon OpenSearch Serverless and custom document IDs](#amazon-opensearch-serverless-and-custom-document-ids) - [Verify and repair an Amazon OpenSearch Serverless vector store](#verify-and-repair-an-amazon-opensearch-serverless-vector-store) + - [Connecting to an OpenSearch vector store](#connecting-to-an-opensearch-vector-store) ### Overview @@ -130,3 +131,40 @@ Field descriptions: | `num_docs` | Number of documents in a specific tenant vector index | | `num_deleted` | Number of documents deleted from a specific tenant vector index (indicative number only if `dry_run` is `true`) | | `num_unindexed` | Number of nodes that have not been indexed in a specific tenant vector index | + +### Connecting to an OpenSearch vector store + +You can use [OpenSearch](https://opensearch.org) directly as a vector store — in a Docker or Finch container, on an EC2 instance, or on your own infrastructure. graphrag-toolkit connects to it with the [opensearch-py](https://pypi.org/project/opensearch-py/) client. Pass the endpoint as an `http://` or `https://` URL: + +```python +from graphrag_toolkit.lexical_graph.storage import VectorStoreFactory + +opensearch_connection_info = 'https://localhost:9200' + +with VectorStoreFactory.for_vector_store( + opensearch_connection_info, + index_names=['chunk'] +) as vector_store: + ... +``` + +Any `https://`/`http://` endpoint that isn't an Amazon OpenSearch Serverless domain is treated as a self-managed OpenSearch cluster (use the `aoss://` prefix for Serverless). Use `https://` to connect over SSL, the toolkit's default (matching OpenSearch, which ships with SSL enabled), or `http://` for a plain-HTTP instance. An endpoint without a port defaults to 9200. + +A default local OpenSearch (Docker or Finch) uses a self-signed certificate, so the example above fails certificate verification until you either trust its CA or disable verification with `client_kwargs={'verify_certs': False}` (see the self-signed example below). Certificate verification stays on by default. + +The client authenticates with HTTP basic auth from the `OPENSEARCH_USERNAME` and `OPENSEARCH_PASSWORD` environment variables (or `GraphRAGConfig.opensearch_username` / `GraphRAGConfig.opensearch_password`), or with no auth if neither is set. + +Any `client_kwargs` you pass to `VectorStoreFactory.for_vector_store` are forwarded to the opensearch-py client, so all of its [connection options](https://opensearch-project.github.io/opensearch-py/api-ref/clients/opensearch_client.html) — certificate verification, CA bundles, timeouts — are available. For example, for an endpoint with a self-signed TLS certificate: + +```python +with VectorStoreFactory.for_vector_store( + 'https://localhost:9200', + index_names=['chunk'], + client_kwargs={'verify_certs': False} +) as vector_store: + ... +``` + +#### Limitations + +Authentication methods other than basic auth, including SigV4 for IAM-authenticated domains, are not supported yet ([#407](https://github.com/awslabs/graphrag-toolkit/issues/407)). diff --git a/lexical-graph/README.md b/lexical-graph/README.md index b80382a1..813c5ad7 100644 --- a/lexical-graph/README.md +++ b/lexical-graph/README.md @@ -41,7 +41,7 @@ If you're running on AWS, you must run your application in an AWS region contain You will need to install additional dependencies for specific graph and vector store backends: -#### Amazon OpenSearch Serverless +#### OpenSearch and Amazon OpenSearch Serverless (AOSS) ```bash $ pip install opensearch-py llama-index-vector-stores-opensearch @@ -68,12 +68,15 @@ Pass a connection string to `GraphStoreFactory.for_graph_store()` or `VectorStor | Neptune Analytics (graph) | `neptune-graph://` | | Neptune Database (graph) | `neptune-db://` or any hostname ending `.neptune.amazonaws.com` | | Neo4j (graph) | `bolt://`, `bolt+ssc://`, `bolt+s://`, `neo4j://`, `neo4j+ssc://`, or `neo4j+s://` URLs | -| OpenSearch Serverless (vector) | `aoss://` | +| Amazon OpenSearch Serverless / AOSS (vector) | `aoss://` | +| OpenSearch (vector) | `https://` or `http://` | | Neptune Analytics (vector) | `neptune-graph://` | | pgvector (vector) | constructed via `PGVectorIndexFactory` | | S3 Vectors (vector) | constructed via `S3VectorIndexFactory` | | Dummy / no-op | `None` or any unrecognised string — falls back to `DummyGraphStore` / `DummyVectorIndex` | +An `opensearch://` endpoint connects with HTTP basic auth from the `OPENSEARCH_USERNAME` / `OPENSEARCH_PASSWORD` environment variables (no auth if unset), over SSL unless the endpoint has an explicit `http://` scheme. `client_kwargs` passed to `VectorStoreFactory.for_vector_store` are forwarded to the [opensearch-py](https://pypi.org/project/opensearch-py/) client. See [the vector store documentation](https://awslabs.github.io/graphrag-toolkit/lexical-graph/vector-store-opensearch-serverless/) for details and limitations. + ## Example of use ### Indexing diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py index b93e304f..12dc6e95 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/config.py @@ -320,6 +320,8 @@ class _GraphRAGConfig: _enable_cache: Optional[bool] = None _metadata_datetime_suffixes: Optional[List[str]] = None _opensearch_engine: Optional[str] = None + _opensearch_username: Optional[str] = None + _opensearch_password: Optional[str] = None _opensearch_serverless_generation: Optional['OpenSearchServerlessGeneration'] = None _opensearch_serverless_nextgen_compression: Optional[str] = None _enable_versioning = None @@ -1203,6 +1205,31 @@ def opensearch_engine(self) -> str: def opensearch_engine(self, opensearch_engine: str) -> None: self._opensearch_engine = opensearch_engine + @property + def opensearch_username(self) -> Optional[str]: + """Basic-auth username for an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS). Unset (None) means no basic auth is applied -- used with an + OpenSearch instance that has no security plugin configured.""" + if self._opensearch_username is None: + self._opensearch_username = os.environ.get('OPENSEARCH_USERNAME') + return self._opensearch_username + + @opensearch_username.setter + def opensearch_username(self, opensearch_username: Optional[str]) -> None: + self._opensearch_username = opensearch_username + + @property + def opensearch_password(self) -> Optional[str]: + """Basic-auth password for an OpenSearch endpoint that isn't Amazon OpenSearch + Serverless (AOSS). See opensearch_username.""" + if self._opensearch_password is None: + self._opensearch_password = os.environ.get('OPENSEARCH_PASSWORD') + return self._opensearch_password + + @opensearch_password.setter + def opensearch_password(self, opensearch_password: Optional[str]) -> None: + self._opensearch_password = opensearch_password + @property def opensearch_serverless_generation(self) -> Optional['OpenSearchServerlessGeneration']: """Unset (None) auto-detects Classic vs NextGen: index_exists() tries the Classic diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py index bf620d44..62254c32 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_index_factory.py @@ -43,19 +43,30 @@ def try_create(self, index_names:List[str], vector_index_info:str, **kwargs) -> names and endpoint configuration, or None if no suitable endpoint is found. """ endpoint = None + is_sigv4_auth = True if vector_index_info.startswith(OPENSEARCH_SERVERLESS): endpoint = vector_index_info[len(OPENSEARCH_SERVERLESS):] + if not endpoint: + raise ValueError(f'Empty endpoint in OpenSearch Serverless vector store connection info: {vector_index_info}') if not endpoint.startswith('https://') and not endpoint.startswith('http://'): endpoint = f'https://{endpoint}' elif vector_index_info.startswith('https://') and vector_index_info.endswith(OPENSEARCH_SERVERLESS_DNS): endpoint = vector_index_info + elif vector_index_info.startswith('https://') or vector_index_info.startswith('http://'): + # Any other http(s) endpoint is a self-managed (non-AOSS) OpenSearch cluster, + # so no prefix is required; aoss:// and bare AOSS domains are handled above. + endpoint = vector_index_info + is_sigv4_auth = False if endpoint: + # is_sigv4_auth is derived from the connection-string scheme above; drop any + # caller-supplied duplicate so it doesn't collide with the keyword below. + kwargs.pop('is_sigv4_auth', None) try: from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes import OpenSearchIndex - logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}]') - return [OpenSearchIndex.for_index(index_name, endpoint, **kwargs) for index_name in index_names] + logger.debug(f'Opening OpenSearch vector indexes [index_names: {index_names}, endpoint: {endpoint}, is_sigv4_auth: {is_sigv4_auth}]') + return [OpenSearchIndex.for_index(index_name, endpoint, is_sigv4_auth=is_sigv4_auth, **kwargs) for index_name in index_names] except ImportError as e: raise e - + else: - return None \ No newline at end of file + return None diff --git a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py index a0903000..81301f52 100644 --- a/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py +++ b/lexical-graph/src/graphrag_toolkit/lexical_graph/storage/vector/opensearch_vector_indexes.py @@ -15,6 +15,7 @@ from llama_index.core.vector_stores.types import VectorStoreQueryResult, VectorStoreQueryMode, MetadataFilters, MetadataFilter from llama_index.core.indices.utils import embed_nodes from llama_index.core.vector_stores.types import MetadataFilters +from llama_index.core.async_utils import asyncio_run from graphrag_toolkit.lexical_graph.metadata import FilterConfig, is_datetime_key, format_datetime from graphrag_toolkit.lexical_graph.versioning import VALID_FROM, VALID_TO, TIMESTAMP_LOWER_BOUND, TIMESTAMP_UPPER_BOUND @@ -116,20 +117,35 @@ class DummyAuth: DEFAULT_POOL_MAXSIZE = 32 -def create_os_client(endpoint, **kwargs): +def _basic_http_auth(): + """HTTP basic-auth tuple for an OpenSearch endpoint (not Amazon OpenSearch + Serverless) from GraphRAGConfig.opensearch_username/opensearch_password, or None if + either is unset (an instance with no security plugin configured).""" + username = GraphRAGConfig.opensearch_username + password = GraphRAGConfig.opensearch_password + if username and password: + return (username, password) + if username or password: + logger.warning('Only one of opensearch_username/opensearch_password is set; connecting without basic auth') + return None + +def create_os_client(endpoint, is_sigv4_auth=True, **kwargs): """ - Creates an OpenSearch client configured to use AWS Signature Version 4 - authentication. + Creates an OpenSearch client. - The function utilizes AWS credentials from a pre-configured session and region - information. It signs requests using Urllib3-based AWS Signature V4 - authentication. The client is initialized with specific connection properties - like SSL usage, certificate verification, timeout, retry settings, and any - additional keyword arguments provided. + By default (is_sigv4_auth True) the client targets Amazon OpenSearch Serverless + (AOSS) and authenticates with AWS Signature Version 4, signed with credentials from + a pre-configured session and region. When is_sigv4_auth is False, this instead + connects to an OpenSearch endpoint that isn't AOSS: HTTP basic auth from + GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise + no auth. The AWS session and region are never touched on that path. Args: endpoint: str The OpenSearch endpoint URL to connect to. + is_sigv4_auth: bool + True (default) for Amazon OpenSearch Serverless (AOSS), signed with SigV4. + False for an OpenSearch endpoint that isn't AOSS; skips SigV4. **kwargs: Any Additional keyword arguments passed to the OpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. @@ -139,15 +155,25 @@ def create_os_client(endpoint, **kwargs): A configured client instance for interacting with OpenSearch. """ - session = GraphRAGConfig.session - region = GraphRAGConfig.aws_region - credentials = session.get_credentials() - service = 'aoss' - - auth = Urllib3AWSV4SignerAuth(credentials, region, service) + # TODO(#407): support auth methods beyond SigV4 and basic auth + if not is_sigv4_auth: + auth = _basic_http_auth() + else: + session = GraphRAGConfig.session + region = GraphRAGConfig.aws_region + credentials = session.get_credentials() + service = 'aoss' + + auth = Urllib3AWSV4SignerAuth(credentials, region, service) + + # SigV4 requests must always go over TLS. On the non-AOSS path honor an explicit + # http:// scheme and default to TLS otherwise. A caller can still override via client_kwargs. + use_ssl = kwargs.get('use_ssl', True if is_sigv4_auth else not endpoint.lower().startswith('http://')) + if auth and not use_ssl: + logger.warning('Connecting with authentication over plain HTTP; credentials will be sent unencrypted') defaults = { - 'use_ssl': True, + 'use_ssl': use_ssl, 'verify_certs': True, 'connection_class': Urllib3HttpConnection, 'timeout': 300, @@ -162,31 +188,44 @@ def create_os_client(endpoint, **kwargs): **{**defaults, **kwargs}, ) -def create_os_async_client(endpoint, **kwargs): +def create_os_async_client(endpoint, is_sigv4_auth=True, **kwargs): """ Creates an asynchronous OpenSearch client. - This function configures and returns an instance of the AsyncOpenSearch client, - configured for secure communication with the specified endpoint. The client - leverages AWS Signature Version 4 for authentication. + By default (is_sigv4_auth True) this targets Amazon OpenSearch Serverless (AOSS) and + authenticates with AWS Signature Version 4. When is_sigv4_auth is False, it connects + to an OpenSearch endpoint that isn't AOSS instead: HTTP basic auth from + GraphRAGConfig.opensearch_username/opensearch_password if both are set, otherwise no + auth. The AWS session and region are never touched on that path. Args: endpoint: The URL of the OpenSearch cluster endpoint. + is_sigv4_auth: bool + True (default) for Amazon OpenSearch Serverless (AOSS), signed with SigV4. + False for an OpenSearch endpoint that isn't AOSS; skips SigV4. **kwargs: Optional parameters for customizing the AsyncOpenSearch client. ``pool_maxsize`` defaults to :data:`DEFAULT_POOL_MAXSIZE` when unset. Returns: AsyncOpenSearch: An instantiated asynchronous OpenSearch client. """ - session = GraphRAGConfig.session - region = GraphRAGConfig.aws_region - credentials = session.get_credentials() - service = 'aoss' + # TODO(#407): support auth methods beyond SigV4 and basic auth + if not is_sigv4_auth: + auth = _basic_http_auth() + else: + session = GraphRAGConfig.session + region = GraphRAGConfig.aws_region + credentials = session.get_credentials() + service = 'aoss' + + auth = AWSV4SignerAsyncAuth(credentials, region, service) - auth = AWSV4SignerAsyncAuth(credentials, region, service) + use_ssl = kwargs.get('use_ssl', True if is_sigv4_auth else not endpoint.lower().startswith('http://')) + if auth and not use_ssl: + logger.warning('Connecting with authentication over plain HTTP; credentials will be sent unencrypted') defaults = { - 'use_ssl': True, + 'use_ssl': use_ssl, 'verify_certs': True, 'connection_class': AsyncHttpConnection, 'timeout': 300, @@ -367,7 +406,7 @@ def _try_create_index(client, index_name, dimensions, writeable, endpoint, nextg return False -def index_exists(endpoint, index_name, dimensions, writeable) -> bool: +def index_exists(endpoint, index_name, dimensions, writeable, client_kwargs=None, *, is_sigv4_auth=True) -> bool: """ Checks if an OpenSearch index exists, and optionally creates it if it does not exist. @@ -388,6 +427,11 @@ def index_exists(endpoint, index_name, dimensions, writeable) -> bool: index_name: The name of the index to check for existence. dimensions: The number of dimensions for the knn_vector in the index. writeable: Flag indicating whether to create the index if it does not exist. + is_sigv4_auth: True (default) for Amazon OpenSearch Serverless (AOSS). False for + an OpenSearch endpoint that isn't AOSS; skips SigV4. + client_kwargs: Optional dict of keyword arguments passed through to the + OpenSearch client (e.g. ``{"use_ssl": False}`` for a plain-HTTP endpoint). + ``pool_maxsize`` defaults to 1 unless overridden here. Returns: bool: True if the index exists (or is created successfully), False otherwise. @@ -397,7 +441,8 @@ def index_exists(endpoint, index_name, dimensions, writeable) -> bool: AOSS NextGen collection and GraphRAGConfig.opensearch_serverless_generation was explicitly set to CLASSIC, so auto-detection didn't get a chance to retry. """ - client = create_os_client(endpoint, pool_maxsize=1) + client_kwargs = {'pool_maxsize': 1, **(client_kwargs or {})} + client = create_os_client(endpoint, is_sigv4_auth=is_sigv4_auth, **client_kwargs) generation = GraphRAGConfig.opensearch_serverless_generation # When unset, generation is detected reactively: try Classic, retry NextGen on # rejection (see _try_create_index). A deterministic alternative is the AOSS @@ -415,7 +460,7 @@ def index_exists(endpoint, index_name, dimensions, writeable) -> bool: client.close() -def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, client_kwargs=None): +def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_model, client_kwargs=None, *, is_sigv4_auth=True): """ Creates an OpenSearch vector client for interacting with an OpenSearch cluster. @@ -428,6 +473,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode index_name: Name of the index to be used for storing vectors. dimensions: Dimensions of the vector space used for embeddings. embed_model: Embedding model associated with the vectors. + is_sigv4_auth: True (default) for Amazon OpenSearch Serverless (AOSS). False for + an OpenSearch endpoint that isn't AOSS; skips SigV4 and marks the underlying + client as non-AOSS (refresh() is called after bulk ingest). client_kwargs: Optional dict of keyword arguments passed through to both the synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). Defaults that are already set @@ -446,7 +494,7 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode client_kwargs = client_kwargs or {} - logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}]') + logger.debug(f'Creating OpenSearch vector client [index_name={index_name}, endpoint: {endpoint}, embed_model={embed_model}, dimensions={dimensions}, client_kwargs={client_kwargs}, is_sigv4_auth={is_sigv4_auth}]') client = None retry_count = 0 @@ -458,9 +506,9 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode dimensions, embedding_field=embedding_field, text_field=text_field, - os_client=create_os_client(endpoint, **client_kwargs), - os_async_client=create_os_async_client(endpoint, **client_kwargs), - http_auth=DummyAuth(service='aoss') + os_client=create_os_client(endpoint, is_sigv4_auth=is_sigv4_auth, **client_kwargs), + os_async_client=create_os_async_client(endpoint, is_sigv4_auth=is_sigv4_auth, **client_kwargs), + http_auth=DummyAuth(service='aoss') if is_sigv4_auth else DummyAuth(service='opensearch') ) start = time.time() @@ -479,7 +527,11 @@ def create_opensearch_vector_client(endpoint, index_name, dimensions, embed_mode logger.debug(f'Index online [index_name: {index_name}, endpoint: {endpoint}, elapsed_seconds: {elapsed}]') else: logger.debug(f'Index not yet available, recreating client [index_name: {index_name}, endpoint: {endpoint}, elapsed_seconds: {elapsed}]') + # Close both clients before recreating; the async client holds its own + # connection pool, so dropping it without closing would leak on every + # retry while the index finishes provisioning. client._os_client.close() + asyncio_run(client._os_async_client.close()) client = None except NotFoundError as err: @@ -579,7 +631,7 @@ class OpenSearchIndex(VectorIndex): client instance for interacting with OpenSearch, initialized on demand. """ @staticmethod - def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kwargs=None): + def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kwargs=None, *, is_sigv4_auth=True): """ Creates and returns an instance of OpenSearchIndex using the provided parameters. @@ -596,6 +648,9 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw the configuration specified in GraphRAGConfig. dimensions (Optional[int]): The dimensions to be used for the embeddings. Defaults to the configuration specified in GraphRAGConfig. + is_sigv4_auth (bool): True (default) for Amazon OpenSearch Serverless + (AOSS). False for an OpenSearch endpoint that isn't AOSS; skips AWS SigV4 + signing in favor of HTTP basic auth or no auth. client_kwargs (Optional[Dict[str, Any]]): Optional dict of keyword arguments forwarded to the underlying synchronous and asynchronous OpenSearch clients (e.g. ``{"pool_maxsize": 10, "timeout": 60}``). @@ -607,13 +662,14 @@ def for_index(index_name, endpoint, embed_model=None, dimensions=None, client_kw embed_model = embed_model or GraphRAGConfig.embed_model dimensions = dimensions or GraphRAGConfig.embed_dimensions - return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, client_kwargs=client_kwargs) + return OpenSearchIndex(index_name=index_name, endpoint=endpoint, dimensions=dimensions, embed_model=embed_model, is_sigv4_auth=is_sigv4_auth, client_kwargs=client_kwargs) endpoint:str index_name:str dimensions:int embed_model:EmbeddingType model_config = ConfigDict(arbitrary_types_allowed=True) + is_sigv4_auth:bool=True client_kwargs:Optional[Dict[str, Any]]=None _client: OpensearchVectorClient = PrivateAttr(default=None) @@ -660,20 +716,22 @@ def client(self) -> OpensearchVectorClient: self._client = None if not self._client: - if index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable): + does_index_exist = index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs) + if does_index_exist: self._client = create_opensearch_vector_client( self.endpoint, self.underlying_index_name(), self.dimensions, self.embed_model, + is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs, ) else: self._client = DummyOpensearchVectorClient() return self._client - + def index_exists(self): - return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable) + return index_exists(self.endpoint, self.underlying_index_name(), self.dimensions, self.writeable, is_sigv4_auth=self.is_sigv4_auth, client_kwargs=self.client_kwargs) def _clean_id(self, s): """ diff --git a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py index a73a4cf8..51bfeb5e 100644 --- a/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py +++ b/lexical-graph/tests/unit/storage/vector/test_opensearch_clients.py @@ -8,6 +8,7 @@ - create_os_async_client (returns AsyncOpenSearch instance) - create_opensearch_vector_client (success, NotFoundError retry) - OpenSearchIndex.client property (index exists, index does not exist) + - create_os_client/create_os_async_client is_sigv4_auth=False path (basic auth, no auth, no SigV4) """ import pytest @@ -139,6 +140,49 @@ def side_effect(*args, **kwargs): assert result is mock_vector_client assert call_count["n"] == 2 + def test_closes_async_client_when_not_available_then_recreates(self): + # Index unavailable on the first attempt, available on the second. The first + # attempt's clients must both be closed before recreating, so the AsyncOpenSearch + # connection pool is not leaked while the index finishes provisioning. + first_client = MagicMock() + second_client = MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", side_effect=[first_client, second_client]): + with patch.object(ovi, "create_os_client", return_value=MagicMock()): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", side_effect=[False, True]): + # (start, elapsed-init, elapsed-in-loop) per attempt; the 60 forces + # the first attempt's inner wait to exit as not-available. + with patch.object(ovi.time, "time", side_effect=[0, 0, 60, 0, 0, 0]): + with patch.object(ovi.time, "sleep"): + with patch.object(ovi, "asyncio_run") as mock_async_run: + result = ovi.create_opensearch_vector_client( + "https://endpoint", "my-index", 1024, MagicMock() + ) + assert result is second_client + # first attempt's sync client closed directly, async client + # closed via asyncio_run(...close()) — no leak + assert first_client._os_client.close.called + assert mock_async_run.called + + def test_client_kwargs_positional_binds_to_client_kwargs_not_sigv4(self): + captured = {} + + def fake_create_os_client(endpoint, is_sigv4_auth=True, **kwargs): + captured.update(kwargs) + captured["is_sigv4_auth"] = is_sigv4_auth + return MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", return_value=MagicMock()): + with patch.object(ovi, "create_os_client", side_effect=fake_create_os_client): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://endpoint", "my-index", 1024, MagicMock(), {"timeout": 99} + ) + assert captured.get("timeout") == 99 + assert captured.get("is_sigv4_auth") is True + # --------------------------------------------------------------------------- # OpenSearchIndex.client property @@ -185,3 +229,354 @@ def test_client_creates_real_client_when_index_exists(self): with patch.object(ovi, "create_opensearch_vector_client", return_value=mock_vector_client): result = index.client assert result is mock_vector_client + + +# --------------------------------------------------------------------------- +# create_os_client / create_os_async_client: is_sigv4_auth=False path +# --------------------------------------------------------------------------- + +class TestCreateOsClientLocal: + + def test_local_with_credentials_uses_basic_auth(self): + mock_os_instance = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: + result = ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + assert result is mock_os_instance + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] == ("admin", "secret") + + def test_local_without_credentials_uses_no_auth(self): + mock_os_instance = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch", return_value=mock_os_instance) as mock_os: + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] is None + + def test_local_with_only_username_warns_and_uses_no_auth(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] is None + assert any("Only one of" in r.message for r in caplog.records) + + def test_local_with_only_password_warns_and_uses_no_auth(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch") as mock_os: + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["http_auth"] is None + assert any("Only one of" in r.message for r in caplog.records) + + def test_schemeless_endpoint_defaults_to_ssl(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is True + + def test_http_scheme_disables_ssl(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("http://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is False + + def test_uppercase_http_scheme_disables_ssl(self): + # opensearch-py lowercases the scheme, so an uppercase http:// must also + # be detected as plain HTTP rather than defaulting to TLS. + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("HTTP://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is False + + def test_sigv4_auth_forces_ssl_even_for_http_endpoint(self): + # A SigV4-signed request must never go over plaintext, regardless of an + # explicit http:// scheme on the endpoint. + mock_session = MagicMock() + mock_session.get_credentials.return_value = MagicMock() + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.session = mock_session + mock_cfg.aws_region = "us-east-1" + with patch.object(ovi, "OpenSearch") as mock_os: + with patch.object(ovi, "Urllib3AWSV4SignerAuth"): + ovi.create_os_client("http://my-collection.aoss.amazonaws.com", is_sigv4_auth=True) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is True + + def test_basic_auth_over_plain_http_warns(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch"): + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("http://localhost:9200", is_sigv4_auth=False) + assert any("unencrypted" in r.message for r in caplog.records) + + def test_basic_auth_over_https_does_not_warn(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "OpenSearch"): + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + assert not any("unencrypted" in r.message for r in caplog.records) + + def test_http_scheme_disables_ssl_async(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch") as mock_os: + ovi.create_os_async_client("http://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is False + + def test_use_ssl_kwarg_overrides_scheme_default(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch") as mock_os: + ovi.create_os_client("http://localhost:9200", is_sigv4_auth=False, use_ssl=True) + _, kwargs = mock_os.call_args + assert kwargs["use_ssl"] is True + + def test_local_never_touches_aws_session_or_sigv4(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "OpenSearch"): + with patch.object(ovi, "Urllib3AWSV4SignerAuth") as mock_sigv4: + ovi.create_os_client("https://localhost:9200", is_sigv4_auth=False) + mock_sigv4.assert_not_called() + mock_cfg.session.assert_not_called() + + def test_default_is_sigv4_auth_true_preserves_sigv4(self): + mock_session = MagicMock() + mock_session.get_credentials.return_value = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.session = mock_session + mock_cfg.aws_region = "us-east-1" + with patch.object(ovi, "OpenSearch"): + with patch.object(ovi, "Urllib3AWSV4SignerAuth") as mock_sigv4: + ovi.create_os_client("https://my-endpoint.aoss.amazonaws.com") + mock_sigv4.assert_called_once() + + +class TestCreateOsAsyncClientLocal: + + def test_local_with_credentials_uses_basic_auth(self): + mock_async_instance = MagicMock() + + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "AsyncOpenSearch", return_value=mock_async_instance) as mock_async: + result = ovi.create_os_async_client("https://localhost:9200", is_sigv4_auth=False) + assert result is mock_async_instance + _, kwargs = mock_async.call_args + assert kwargs["http_auth"] == ("admin", "secret") + + def test_local_without_credentials_uses_no_auth(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch") as mock_async: + ovi.create_os_async_client("https://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_async.call_args + assert kwargs["http_auth"] is None + + def test_local_never_touches_aws_session_or_sigv4(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch"): + with patch.object(ovi, "AWSV4SignerAsyncAuth") as mock_sigv4: + ovi.create_os_async_client("https://localhost:9200", is_sigv4_auth=False) + mock_sigv4.assert_not_called() + mock_cfg.session.assert_not_called() + + def test_uppercase_http_scheme_disables_ssl_async(self): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = None + mock_cfg.opensearch_password = None + with patch.object(ovi, "AsyncOpenSearch") as mock_async: + ovi.create_os_async_client("HTTP://localhost:9200", is_sigv4_auth=False) + _, kwargs = mock_async.call_args + assert kwargs["use_ssl"] is False + + def test_sigv4_auth_forces_ssl_even_for_http_endpoint_async(self): + mock_session = MagicMock() + mock_session.get_credentials.return_value = MagicMock() + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.session = mock_session + mock_cfg.aws_region = "us-east-1" + with patch.object(ovi, "AsyncOpenSearch") as mock_async: + with patch.object(ovi, "AWSV4SignerAsyncAuth"): + ovi.create_os_async_client("http://my-collection.aoss.amazonaws.com", is_sigv4_auth=True) + _, kwargs = mock_async.call_args + assert kwargs["use_ssl"] is True + + def test_basic_auth_over_plain_http_warns_async(self, caplog): + with patch.object(ovi, "GraphRAGConfig") as mock_cfg: + mock_cfg.opensearch_username = "admin" + mock_cfg.opensearch_password = "secret" + with patch.object(ovi, "AsyncOpenSearch"): + with caplog.at_level("WARNING", logger="graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes"): + ovi.create_os_async_client("http://localhost:9200", is_sigv4_auth=False) + assert any("unencrypted" in r.message for r in caplog.records) + + +class TestCreateOpensearchVectorClientLocal: + + def test_is_sigv4_auth_threaded_to_both_clients(self): + with patch.object(ovi, "OpensearchVectorClient", return_value=MagicMock()): + with patch.object(ovi, "create_os_client", return_value=MagicMock()) as mock_sync: + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()) as mock_async: + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://localhost:9200", "my-index", 1024, MagicMock(), is_sigv4_auth=False + ) + assert mock_sync.call_args.kwargs.get("is_sigv4_auth") is False + assert mock_async.call_args.kwargs.get("is_sigv4_auth") is False + + def test_non_sigv4_auth_sets_non_aoss_dummy_auth_service(self): + captured = {} + + def fake_vector_client(*args, **kwargs): + captured["http_auth"] = kwargs.get("http_auth") + return MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", side_effect=fake_vector_client): + with patch.object(ovi, "create_os_client", return_value=MagicMock()): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://localhost:9200", "my-index", 1024, MagicMock(), is_sigv4_auth=False + ) + assert captured["http_auth"].service != "aoss" + + def test_default_is_sigv4_auth_true_keeps_aoss_dummy_auth_service(self): + captured = {} + + def fake_vector_client(*args, **kwargs): + captured["http_auth"] = kwargs.get("http_auth") + return MagicMock() + + with patch.object(ovi, "OpensearchVectorClient", side_effect=fake_vector_client): + with patch.object(ovi, "create_os_client", return_value=MagicMock()): + with patch.object(ovi, "create_os_async_client", return_value=MagicMock()): + with patch.object(ovi, "index_is_available", return_value=True): + ovi.create_opensearch_vector_client( + "https://endpoint", "my-index", 1024, MagicMock() + ) + assert captured["http_auth"].service == "aoss" + + +class TestOpenSearchIndexClientPropertyLocal: + + def _make_local_index(self): + from graphrag_toolkit.lexical_graph.tenant_id import TenantId + return ovi.OpenSearchIndex.model_construct( + index_name="chunk", + endpoint="https://localhost:9200", + dimensions=1024, + embed_model=MagicMock(), + tenant_id=TenantId(), + writeable=False, + is_sigv4_auth=False, + _client=None, + ) + + def test_client_property_threads_is_sigv4_auth_to_index_exists(self): + index = self._make_local_index() + + with patch.object(ovi, "index_exists", return_value=False) as mock_index_exists: + index.client + assert mock_index_exists.call_args.kwargs.get("is_sigv4_auth") is False + + def test_client_property_threads_is_sigv4_auth_to_vector_client(self): + index = self._make_local_index() + mock_vector_client = MagicMock() + + with patch.object(ovi, "index_exists", return_value=True): + with patch.object(ovi, "create_opensearch_vector_client", return_value=mock_vector_client) as mock_create: + index.client + assert mock_create.call_args.kwargs.get("is_sigv4_auth") is False + + def test_client_property_threads_client_kwargs_to_index_exists(self): + from graphrag_toolkit.lexical_graph.tenant_id import TenantId + index = ovi.OpenSearchIndex.model_construct( + index_name="chunk", + endpoint="http://localhost:9200", + dimensions=1024, + embed_model=MagicMock(), + tenant_id=TenantId(), + writeable=False, + is_sigv4_auth=False, + client_kwargs={"use_ssl": False, "verify_certs": False}, + _client=None, + ) + + with patch.object(ovi, "index_exists", return_value=False) as mock_index_exists: + index.client + assert mock_index_exists.call_args.kwargs.get("client_kwargs") == {"use_ssl": False, "verify_certs": False} + + +# --------------------------------------------------------------------------- +# index_exists: client_kwargs passthrough (plain-HTTP / self-signed local endpoints) +# --------------------------------------------------------------------------- + +class TestIndexExistsClientKwargs: + + def _mock_client(self): + client = MagicMock() + client.indices.exists.return_value = True + return client + + def test_default_client_kwargs_none_still_sets_pool_maxsize(self): + with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: + ovi.index_exists("https://endpoint", "my-index", 1024, writeable=False) + _, kwargs = mock_create.call_args + assert kwargs["pool_maxsize"] == 1 + + def test_client_kwargs_forwarded_to_create_os_client(self): + with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: + ovi.index_exists( + "http://localhost:9200", "my-index", 1024, writeable=False, is_sigv4_auth=False, + client_kwargs={"use_ssl": False, "verify_certs": False}, + ) + _, kwargs = mock_create.call_args + assert kwargs["use_ssl"] is False + assert kwargs["verify_certs"] is False + assert kwargs["is_sigv4_auth"] is False + + def test_client_kwargs_can_override_default_pool_maxsize(self): + with patch.object(ovi, "create_os_client", return_value=self._mock_client()) as mock_create: + ovi.index_exists( + "https://endpoint", "my-index", 1024, writeable=False, + client_kwargs={"pool_maxsize": 10}, + ) + _, kwargs = mock_create.call_args + assert kwargs["pool_maxsize"] == 10 diff --git a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py index 4d0b7822..f9495a14 100644 --- a/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py +++ b/lexical-graph/tests/unit/storage/vector/test_vector_index_factories.py @@ -4,7 +4,7 @@ """Tests for vector index factory try_create methods. Covers: - - OpenSearchVectorIndexFactory.try_create (aoss://, https://...aoss, no match) + - OpenSearchVectorIndexFactory.try_create (aoss://, https://...aoss, http(s):// endpoint, no match) - PGVectorIndexFactory.try_create (postgres://, postgresql://, no match) - S3VectorIndexFactory.try_create (s3vectors://, no match) """ @@ -66,6 +66,107 @@ def test_try_create_https_aoss_dns_returns_indexes(self): "https://abc.us-east-1.aoss.amazonaws.com" ) assert result is not None + # a bare AOSS domain must stay on SigV4 — this branch is matched ahead of + # the generic https:// branch, so the ordering is load-bearing + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_sigv4_auth") is True + + def test_try_create_https_endpoint_returns_indexes(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_index = MagicMock() + mock_cls = MagicMock() + mock_cls.for_index.return_value = mock_index + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + result = factory.try_create( + ["chunk", "statement"], + "https://localhost:9200" + ) + assert result is not None + assert len(result) == 2 + args, kwargs = mock_cls.for_index.call_args + # endpoint passed through verbatim, no prefix stripping, non-AOSS auth + assert args[1] == "https://localhost:9200" + assert kwargs.get("is_sigv4_auth") is False + + def test_try_create_http_endpoint_passes_through_and_sets_non_sigv4(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "http://localhost:9200") + args, kwargs = mock_cls.for_index.call_args + assert args[1] == "http://localhost:9200" + assert kwargs.get("is_sigv4_auth") is False + + def test_try_create_schemeless_endpoint_returns_none(self): + # Without the opensearch:// prefix, a schemeless endpoint is no longer claimed + # by this factory; the caller must supply an http:// or https:// scheme. + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + factory = OpenSearchVectorIndexFactory() + assert factory.try_create(["chunk"], "localhost:9200") is None + + def test_try_create_opensearch_prefix_no_longer_matched(self): + # opensearch:// was removed; such a string should fall through (return None) + # so the dispatch reports it as unrecognized rather than mis-routing. + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + factory = OpenSearchVectorIndexFactory() + assert factory.try_create(["chunk"], "opensearch://localhost:9200") is None + + def test_try_create_aoss_prefix_sets_is_sigv4_auth_true(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "aoss://https://abc.us-east-1.aoss.amazonaws.com") + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_sigv4_auth") is True + + def test_try_create_aoss_prefix_empty_endpoint_raises(self): + import pytest + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + factory = OpenSearchVectorIndexFactory() + with pytest.raises(ValueError, match="Empty endpoint"): + factory.try_create(["chunk"], "aoss://") + + def test_try_create_caller_supplied_is_sigv4_auth_does_not_collide(self): + from graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_index_factory import OpenSearchVectorIndexFactory + + mock_cls = MagicMock() + mock_cls.for_index.return_value = MagicMock() + + with patch.dict("sys.modules", { + "graphrag_toolkit.lexical_graph.storage.vector.opensearch_vector_indexes": MagicMock( + OpenSearchIndex=mock_cls + ) + }): + factory = OpenSearchVectorIndexFactory() + factory.try_create(["chunk"], "https://localhost:9200", is_sigv4_auth=True) + _, kwargs = mock_cls.for_index.call_args + assert kwargs.get("is_sigv4_auth") is False # --------------------------------------------------------------------------- diff --git a/lexical-graph/tests/unit/test_config.py b/lexical-graph/tests/unit/test_config.py index f398d327..1000823e 100644 --- a/lexical-graph/tests/unit/test_config.py +++ b/lexical-graph/tests/unit/test_config.py @@ -206,6 +206,65 @@ def test_build_batch_size_from_env(self): finally: GraphRAGConfig._build_batch_size = original + def test_opensearch_username_defaults_none(self): + """Verify opensearch_username defaults to None when unset.""" + original = GraphRAGConfig._opensearch_username + try: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop('OPENSEARCH_USERNAME', None) + GraphRAGConfig._opensearch_username = None + assert GraphRAGConfig.opensearch_username is None + finally: + GraphRAGConfig._opensearch_username = original + + def test_opensearch_username_from_env(self): + """Verify opensearch_username reads from OPENSEARCH_USERNAME env var.""" + original = GraphRAGConfig._opensearch_username + try: + with patch.dict(os.environ, {'OPENSEARCH_USERNAME': 'admin'}): + GraphRAGConfig._opensearch_username = None + assert GraphRAGConfig.opensearch_username == 'admin' + finally: + GraphRAGConfig._opensearch_username = original + + def test_opensearch_username_setter(self): + """Verify opensearch_username can be set directly.""" + original = GraphRAGConfig._opensearch_username + try: + GraphRAGConfig.opensearch_username = 'custom-user' + assert GraphRAGConfig.opensearch_username == 'custom-user' + finally: + GraphRAGConfig._opensearch_username = original + + def test_opensearch_password_defaults_none(self): + """Verify opensearch_password defaults to None when unset.""" + original = GraphRAGConfig._opensearch_password + try: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop('OPENSEARCH_PASSWORD', None) + GraphRAGConfig._opensearch_password = None + assert GraphRAGConfig.opensearch_password is None + finally: + GraphRAGConfig._opensearch_password = original + + def test_opensearch_password_from_env(self): + """Verify opensearch_password reads from OPENSEARCH_PASSWORD env var.""" + original = GraphRAGConfig._opensearch_password + try: + with patch.dict(os.environ, {'OPENSEARCH_PASSWORD': 'secret'}): + GraphRAGConfig._opensearch_password = None + assert GraphRAGConfig.opensearch_password == 'secret' + finally: + GraphRAGConfig._opensearch_password = original + + def test_opensearch_password_setter(self): + """Verify opensearch_password can be set directly.""" + original = GraphRAGConfig._opensearch_password + try: + GraphRAGConfig.opensearch_password = 'custom-pass' + assert GraphRAGConfig.opensearch_password == 'custom-pass' + finally: + GraphRAGConfig._opensearch_password = original def test_opensearch_serverless_generation_defaults_none_for_auto_detect(self): """opensearch_serverless_generation defaults to None (auto-detect) when unset.""" original = GraphRAGConfig._opensearch_serverless_generation