From 934ea8cc5ed110f7032aad909ce629a2beb2af1b Mon Sep 17 00:00:00 2001 From: Shephali Mittal Date: Thu, 5 Aug 2021 15:23:50 +0530 Subject: [PATCH] Revert product checker commits: 023cf5380f900d3a21b6155bd963caf6d10edf61: "[7.x] Document 'UnsupportedProductError'" d381491482c0e9a8b09a5c63520d7eb1ea8a5ef8: "[7.x] Rename product error to 'UnsupportedProductError'" b56fbcb155c2e552df8b698a243ef592e2aacde2: "Document the Elasticsearch product check" b63d00513802ddefa14720043b9dadb85a9e556e: "Don't swallow unexpected errors during Elasticsearch verification" 801a8390937a6ef909158c87a6a2e0e39300dcfa: "Verify we're connected to Elasticsearch before requests" Signed-off-by: Shephali Mittal --- docs/sphinx/exceptions.rst | 1 - docs/sphinx/transports.rst | 21 - elasticsearch/__init__.py | 2 - elasticsearch/__init__.pyi | 1 - elasticsearch/_async/transport.py | 99 +--- elasticsearch/compat.py | 11 - elasticsearch/exceptions.py | 6 - elasticsearch/exceptions.pyi | 1 - elasticsearch/transport.py | 191 ------- .../test_async/test_connection.py | 8 +- .../test_async/test_transport.py | 330 +---------- test_elasticsearch/test_connection.py | 8 +- test_elasticsearch/test_transport.py | 519 +----------------- 13 files changed, 6 insertions(+), 1192 deletions(-) diff --git a/docs/sphinx/exceptions.rst b/docs/sphinx/exceptions.rst index 2ec9c6ce..03bb60e1 100644 --- a/docs/sphinx/exceptions.rst +++ b/docs/sphinx/exceptions.rst @@ -23,4 +23,3 @@ Exceptions .. autoclass:: RequestError(TransportError) .. autoclass:: AuthenticationException(TransportError) .. autoclass:: AuthorizationException(TransportError) -.. autoclass:: UnsupportedProductError diff --git a/docs/sphinx/transports.rst b/docs/sphinx/transports.rst index 6daa15db..1a0d902b 100644 --- a/docs/sphinx/transports.rst +++ b/docs/sphinx/transports.rst @@ -21,27 +21,6 @@ and lightweight than the optional ``requests``-based class. Only use ``RequestsHttpConnection`` if you have need of any of ``requests`` advanced features like custom auth plugins etc. -Product check on first request ------------------------------- - -Starting in v7.14.0 the client performs a required product check before -the first API call is executed. This product check allows the client to -establish that it's communicating with a supported Elasticsearch cluster. - -The product check requires a single HTTP request to the ``info`` API. In -most cases this request will succeed quickly and then no further product -check HTTP requests will be sent. - -The product check will verify that the ``X-Elastic-Product: Elasticsearch`` -HTTP header is being sent or if the ``info`` API indicates a supported -distribution of Elasticsearch. - -If the client detects that it's not connected to a supported distribution of -Elasticsearch the ``UnsupportedProductError`` exception will be raised. -In previous versions of Elasticsearch the ``info`` API required additional -permissions so if an authentication or authorization error is raised during -the product check then an ``ElasticsearchWarning`` is raised and the client -proceeds normally. .. py:module:: elasticsearch.connection diff --git a/elasticsearch/__init__.py b/elasticsearch/__init__.py index ea05b5b2..05f16398 100644 --- a/elasticsearch/__init__.py +++ b/elasticsearch/__init__.py @@ -51,7 +51,6 @@ from .exceptions import ( SerializationError, SSLError, TransportError, - UnsupportedProductError, ) from .serializer import JSONSerializer from .transport import Transport @@ -82,7 +81,6 @@ __all__ = [ "ConnectionTimeout", "AuthenticationException", "AuthorizationException", - "UnsupportedProductError", "ElasticsearchWarning", "ElasticsearchDeprecationWarning", ] diff --git a/elasticsearch/__init__.pyi b/elasticsearch/__init__.pyi index 373f623c..20a60aa2 100644 --- a/elasticsearch/__init__.pyi +++ b/elasticsearch/__init__.pyi @@ -40,7 +40,6 @@ from .exceptions import RequestError as RequestError from .exceptions import SerializationError as SerializationError from .exceptions import SSLError as SSLError from .exceptions import TransportError as TransportError -from .exceptions import UnsupportedProductError as UnsupportedProductError from .serializer import JSONSerializer as JSONSerializer from .transport import Transport as Transport diff --git a/elasticsearch/_async/transport.py b/elasticsearch/_async/transport.py index 9fa94107..3fd63733 100644 --- a/elasticsearch/_async/transport.py +++ b/elasticsearch/_async/transport.py @@ -18,19 +18,15 @@ import asyncio import logging import sys -import warnings from itertools import chain from ..exceptions import ( - AuthenticationException, - AuthorizationException, ConnectionError, ConnectionTimeout, - ElasticsearchWarning, SerializationError, TransportError, ) -from ..transport import Transport, _ProductChecker +from ..transport import Transport from .compat import get_running_loop from .http_aiohttp import AIOHttpConnection @@ -117,10 +113,6 @@ class AsyncTransport(Transport): self.loop = get_running_loop() self.kwargs["loop"] = self.loop - # Set our 'verified_once' implementation to one that - # works with 'asyncio' instead of 'threading' - self._verify_elasticsearch_lock = asyncio.Lock() - # Now that we have a loop we can create all our HTTP connections... self.set_connections(self.hosts) self.seed_connections = list(self.connection_pool.connections[:]) @@ -335,14 +327,6 @@ class AsyncTransport(Transport): method, headers, params, body ) - # Before we make the actual API call we verify the Elasticsearch instance. - if self._verified_elasticsearch is None: - await self._do_verify_elasticsearch(headers=headers, timeout=timeout) - - # If '_verified_elasticsearch' isn't 'True' then we raise an error. - if self._verified_elasticsearch is not True: - _ProductChecker.raise_error(self._verified_elasticsearch) - for attempt in range(self.max_retries + 1): connection = self.get_connection() @@ -414,84 +398,3 @@ class AsyncTransport(Transport): for connection in self.connection_pool.connections: await connection.close() - - async def _do_verify_elasticsearch(self, headers, timeout): - """Verifies that we're connected to an Elasticsearch cluster. - This is done at least once before the first actual API call - and makes a single request to the 'GET /' API endpoint and - check version along with other details of the response. - - If we're unable to verify we're talking to Elasticsearch - but we're also unable to rule it out due to a permission - error we instead emit an 'ElasticsearchWarning'. - """ - # Ensure that there's only one async exec within this section - # at a time to not emit unnecessary index API calls. - async with self._verify_elasticsearch_lock: - - # Product check has already been completed while we were - # waiting our turn, no need to do again. - if self._verified_elasticsearch is not None: - return - - headers = { - header.lower(): value for header, value in (headers or {}).items() - } - # We know we definitely want JSON so request it via 'accept' - headers.setdefault("accept", "application/json") - - info_headers = {} - info_response = {} - error = None - - attempted_conns = [] - for conn in chain(self.connection_pool.connections, self.seed_connections): - # Only attempt once per connection max. - if conn in attempted_conns: - continue - attempted_conns.append(conn) - - try: - _, info_headers, info_response = await conn.perform_request( - "GET", "/", headers=headers, timeout=timeout - ) - - # Lowercase all the header names for consistency in accessing them. - info_headers = { - header.lower(): value for header, value in info_headers.items() - } - - info_response = self.deserializer.loads( - info_response, mimetype="application/json" - ) - break - - # Previous versions of 7.x Elasticsearch required a specific - # permission so if we receive HTTP 401/403 we should warn - # instead of erroring out. - except (AuthenticationException, AuthorizationException): - warnings.warn( - ( - "The client is unable to verify that the server is " - "Elasticsearch due security privileges on the server side" - ), - ElasticsearchWarning, - stacklevel=4, - ) - self._verified_elasticsearch = True - return - - # This connection didn't work, we'll try another. - except (ConnectionError, SerializationError, TransportError) as err: - if error is None: - error = err - - # If we received a connection error and weren't successful - # anywhere then we re-raise the more appropriate error. - if error and not info_response: - raise error - - # Check the information we got back from the index request. - self._verified_elasticsearch = _ProductChecker.check_product( - info_headers, info_response - ) diff --git a/elasticsearch/compat.py b/elasticsearch/compat.py index 912d2c72..99425ce6 100644 --- a/elasticsearch/compat.py +++ b/elasticsearch/compat.py @@ -70,17 +70,6 @@ try: except (ImportError, AttributeError): pass -try: - from threading import Lock -except ImportError: # Python <3.7 isn't guaranteed to have threading support. - - class Lock: - def __enter__(self): - pass - - def __exit__(self, *_): - pass - __all__ = [ "string_types", diff --git a/elasticsearch/exceptions.py b/elasticsearch/exceptions.py index 9ff5c8b2..1a7a438d 100644 --- a/elasticsearch/exceptions.py +++ b/elasticsearch/exceptions.py @@ -51,12 +51,6 @@ class SerializationError(ElasticsearchException): """ -class UnsupportedProductError(ElasticsearchException): - """Error which is raised when the client detects - it's not connected to a supported product. - """ - - class TransportError(ElasticsearchException): """ Exception raised when ES returns a non-OK (>=400) HTTP status code. Or when diff --git a/elasticsearch/exceptions.pyi b/elasticsearch/exceptions.pyi index af39f2d5..fd28fd80 100644 --- a/elasticsearch/exceptions.pyi +++ b/elasticsearch/exceptions.pyi @@ -20,7 +20,6 @@ from typing import Any, Dict, Union class ImproperlyConfigured(Exception): ... class ElasticsearchException(Exception): ... class SerializationError(ElasticsearchException): ... -class UnsupportedProductError(ElasticsearchException): ... class TransportError(ElasticsearchException): @property diff --git a/elasticsearch/transport.py b/elasticsearch/transport.py index 5e9d80e2..cf46e9b2 100644 --- a/elasticsearch/transport.py +++ b/elasticsearch/transport.py @@ -15,25 +15,18 @@ # specific language governing permissions and limitations # under the License. -import re import time -import warnings from itertools import chain from platform import python_version from ._version import __versionstr__ -from .compat import Lock from .connection import Urllib3HttpConnection from .connection_pool import ConnectionPool, DummyConnectionPool, EmptyConnectionPool from .exceptions import ( - AuthenticationException, - AuthorizationException, ConnectionError, ConnectionTimeout, - ElasticsearchWarning, SerializationError, TransportError, - UnsupportedProductError, ) from .serializer import DEFAULT_SERIALIZERS, Deserializer, JSONSerializer from .utils import _client_meta_version @@ -205,23 +198,6 @@ class Transport(object): if http_client_meta: self._client_meta += (http_client_meta,) - # Tri-state flag that describes what state the verification - # of whether we're connected to an Elasticsearch cluster or not. - # The three states are: - # - 'None': Means we've either not started the verification process - # or that the verification is in progress. '_verified_once' ensures - # that multiple requests don't kick off multiple verification processes. - # - 'True': Means we've verified that we're talking to Elasticsearch or - # that we can't rule out Elasticsearch due to auth issues. A warning - # will be raised if we receive 401/403. - # - 'int': Means we're talking to an unsupported product, should raise - # the corresponding error. - self._verified_elasticsearch = None - - # Ensures that the ES verification request only fires once and that - # all requests block until this request returns back. - self._verify_elasticsearch_lock = Lock() - def add_connection(self, host): """ Create a new :class:`~elasticsearch.Connection` instance and add it to the pool. @@ -404,14 +380,6 @@ class Transport(object): method, headers, params, body ) - # Before we make the actual API call we verify the Elasticsearch instance. - if self._verified_elasticsearch is None: - self._do_verify_elasticsearch(headers=headers, timeout=timeout) - - # If '_verified_elasticsearch' isn't 'True' then we raise an error. - if self._verified_elasticsearch is not True: - _ProductChecker.raise_error(self._verified_elasticsearch) - for attempt in range(self.max_retries + 1): connection = self.get_connection() @@ -520,162 +488,3 @@ class Transport(object): ) return method, headers, params, body, ignore, timeout - - def _do_verify_elasticsearch(self, headers, timeout): - """Verifies that we're connected to an Elasticsearch cluster. - This is done at least once before the first actual API call - and makes a single request to the 'GET /' API endpoint to - check the version along with other details of the response. - - If we're unable to verify we're talking to Elasticsearch - but we're also unable to rule it out due to a permission - error we instead emit an 'ElasticsearchWarning'. - """ - # Ensure that there's only one thread within this section - # at a time to not emit unnecessary index API calls. - with self._verify_elasticsearch_lock: - - # Product check has already been completed while we were - # waiting our turn, no need to do again. - if self._verified_elasticsearch is not None: - return - - headers = { - header.lower(): value for header, value in (headers or {}).items() - } - # We know we definitely want JSON so request it via 'accept' - headers.setdefault("accept", "application/json") - - info_headers = {} - info_response = {} - error = None - - attempted_conns = [] - for conn in chain(self.connection_pool.connections, self.seed_connections): - # Only attempt once per connection max. - if conn in attempted_conns: - continue - attempted_conns.append(conn) - - try: - _, info_headers, info_response = conn.perform_request( - "GET", "/", headers=headers, timeout=timeout - ) - - # Lowercase all the header names for consistency in accessing them. - info_headers = { - header.lower(): value for header, value in info_headers.items() - } - - info_response = self.deserializer.loads( - info_response, mimetype="application/json" - ) - break - - # Previous versions of 7.x Elasticsearch required a specific - # permission so if we receive HTTP 401/403 we should warn - # instead of erroring out. - except (AuthenticationException, AuthorizationException): - warnings.warn( - ( - "The client is unable to verify that the server is " - "Elasticsearch due security privileges on the server side" - ), - ElasticsearchWarning, - stacklevel=5, - ) - self._verified_elasticsearch = True - return - - # This connection didn't work, we'll try another. - except (ConnectionError, SerializationError, TransportError) as err: - if error is None: - error = err - - # If we received a connection error and weren't successful - # anywhere then we re-raise the more appropriate error. - if error and not info_response: - raise error - - # Check the information we got back from the index request. - self._verified_elasticsearch = _ProductChecker.check_product( - info_headers, info_response - ) - - -class _ProductChecker: - """Class which verifies we're connected to a supported product""" - - # States that can be returned from 'check_product' - SUCCESS = True - UNSUPPORTED_PRODUCT = 2 - UNSUPPORTED_DISTRIBUTION = 3 - - @classmethod - def raise_error(cls, state): - # These states mean the product_check() didn't fail so do nothing. - if state in (None, True): - return - - if state == cls.UNSUPPORTED_DISTRIBUTION: - message = ( - "The client noticed that the server is not " - "a supported distribution of Elasticsearch" - ) - else: # UNSUPPORTED_PRODUCT - message = ( - "The client noticed that the server is not Elasticsearch " - "and we do not support this unknown product" - ) - raise UnsupportedProductError(message) - - @classmethod - def check_product(cls, headers, response): - # type: (dict[str, str], dict[str, str]) -> int - """Verifies that the server we're talking to is Elasticsearch. - Does this by checking HTTP headers and the deserialized - response to the 'info' API. Returns one of the states above. - """ - try: - version = response.get("version", {}) - version_number = tuple( - int(x) if x is not None else 999 - for x in re.search( - r"^([0-9]+)\.([0-9]+)(?:\.([0-9]+))?", version["number"] - ).groups() - ) - except (KeyError, TypeError, ValueError, AttributeError): - # No valid 'version.number' field, effectively 0.0.0 - version = {} - version_number = (0, 0, 0) - - # Check all of the fields and headers for missing/valid values. - try: - bad_tagline = response.get("tagline", None) != "You Know, for Search" - bad_build_flavor = version.get("build_flavor", None) != "default" - bad_product_header = ( - headers.get("x-elastic-product", None) != "Elasticsearch" - ) - except (AttributeError, TypeError): - bad_tagline = True - bad_build_flavor = True - bad_product_header = True - - # 7.0-7.13 and there's a bad 'tagline' or unsupported 'build_flavor' - if (7, 0, 0) <= version_number < (7, 14, 0): - if bad_tagline: - return cls.UNSUPPORTED_PRODUCT - elif bad_build_flavor: - return cls.UNSUPPORTED_DISTRIBUTION - - elif ( - # No version or version less than 6.x - version_number < (6, 0, 0) - # 6.x and there's a bad 'tagline' - or ((6, 0, 0) <= version_number < (7, 0, 0) and bad_tagline) - # 7.14+ and there's a bad 'X-Elastic-Product' HTTP header - or ((7, 14, 0) <= version_number and bad_product_header) - ): - return cls.UNSUPPORTED_PRODUCT - - return True diff --git a/test_elasticsearch/test_async/test_connection.py b/test_elasticsearch/test_async/test_connection.py index 1d7d1ae6..d0646e1c 100644 --- a/test_elasticsearch/test_async/test_connection.py +++ b/test_elasticsearch/test_async/test_connection.py @@ -28,7 +28,7 @@ import pytest from mock import patch from multidict import CIMultiDict -from elasticsearch import AIOHttpConnection, AsyncElasticsearch, __versionstr__ +from elasticsearch import AIOHttpConnection, __versionstr__ from elasticsearch.compat import reraise_exceptions from elasticsearch.exceptions import ConnectionError @@ -410,9 +410,3 @@ class TestConnectionHttpbin: conn = AIOHttpConnection("not.a.host.name") with pytest.raises(ConnectionError): await conn.perform_request("GET", "/") - - async def test_elasticsearch_connection_error(self): - es = AsyncElasticsearch("http://not.a.host.name") - - with pytest.raises(ConnectionError): - await es.search() diff --git a/test_elasticsearch/test_async/test_transport.py b/test_elasticsearch/test_async/test_transport.py index 9b3743d7..13bc492f 100644 --- a/test_elasticsearch/test_async/test_transport.py +++ b/test_elasticsearch/test_async/test_transport.py @@ -28,16 +28,7 @@ from mock import patch from elasticsearch import AsyncTransport from elasticsearch.connection import Connection from elasticsearch.connection_pool import DummyConnectionPool -from elasticsearch.exceptions import ( - AuthenticationException, - AuthorizationException, - ConnectionError, - ElasticsearchWarning, - NotFoundError, - TransportError, - UnsupportedProductError, -) -from elasticsearch.transport import _ProductChecker +from elasticsearch.exceptions import ConnectionError, TransportError pytestmark = pytest.mark.asyncio @@ -130,7 +121,6 @@ class TestTransport: async def test_request_timeout_extracted_from_params_and_passed(self): t = AsyncTransport([{}], connection_class=DummyConnection, meta_header=False) - t._verified_elasticsearch = True await t.perform_request("GET", "/", params={"request_timeout": 42}) assert 1 == len(t.get_connection().calls) @@ -145,7 +135,6 @@ class TestTransport: t = AsyncTransport( [{}], opaque_id="app-1", connection_class=DummyConnection, meta_header=False ) - t._verified_elasticsearch = True await t.perform_request("GET", "/") assert 1 == len(t.get_connection().calls) @@ -168,7 +157,6 @@ class TestTransport: async def test_request_with_custom_user_agent_header(self): t = AsyncTransport([{}], connection_class=DummyConnection, meta_header=False) - t._verified_elasticsearch = True await t.perform_request( "GET", "/", headers={"user-agent": "my-custom-value/1.2.3"} @@ -184,7 +172,6 @@ class TestTransport: t = AsyncTransport( [{}], send_get_body_as="source", connection_class=DummyConnection ) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body={}) assert 1 == len(t.get_connection().calls) @@ -194,7 +181,6 @@ class TestTransport: t = AsyncTransport( [{}], send_get_body_as="POST", connection_class=DummyConnection ) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body={}) assert 1 == len(t.get_connection().calls) @@ -202,7 +188,6 @@ class TestTransport: async def test_client_meta_header(self): t = AsyncTransport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body={}) assert len(t.get_connection().calls) == 1 @@ -216,7 +201,6 @@ class TestTransport: HTTP_CLIENT_META = ("dm", "1.2.3") t = AsyncTransport([{}], connection_class=DummyConnectionWithMeta) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body={}, headers={"Custom": "header"}) assert len(t.get_connection().calls) == 1 @@ -229,7 +213,6 @@ class TestTransport: async def test_client_meta_header_not_sent(self): t = AsyncTransport([{}], meta_header=False, connection_class=DummyConnection) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body={}) assert len(t.get_connection().calls) == 1 @@ -238,7 +221,6 @@ class TestTransport: async def test_body_gets_encoded_into_bytes(self): t = AsyncTransport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body="你好") assert 1 == len(t.get_connection().calls) @@ -251,7 +233,6 @@ class TestTransport: async def test_body_bytes_get_passed_untouched(self): t = AsyncTransport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True body = b"\xe4\xbd\xa0\xe5\xa5\xbd" await t.perform_request("GET", "/", body=body) @@ -260,7 +241,6 @@ class TestTransport: async def test_body_surrogates_replaced_encoded_into_bytes(self): t = AsyncTransport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True await t.perform_request("GET", "/", body="你好\uda6a") assert 1 == len(t.get_connection().calls) @@ -273,8 +253,6 @@ class TestTransport: async def test_kwargs_passed_on_to_connections(self): t = AsyncTransport([{"host": "google.com"}], port=123) - t._verified_elasticsearch = True - await t._async_call() assert 1 == len(t.connection_pool.connections) assert "http://google.com:123" == t.connection_pool.connections[0].host @@ -282,8 +260,6 @@ class TestTransport: async def test_kwargs_passed_on_to_connection_pool(self): dt = object() t = AsyncTransport([{}, {}], dead_timeout=dt) - t._verified_elasticsearch = True - await t._async_call() assert dt is t.connection_pool.dead_timeout @@ -293,15 +269,12 @@ class TestTransport: self.kwargs = kwargs t = AsyncTransport([{}], connection_class=MyConnection) - t._verified_elasticsearch = True - await t._async_call() assert 1 == len(t.connection_pool.connections) assert isinstance(t.connection_pool.connections[0], MyConnection) def test_add_connection(self): t = AsyncTransport([{}], randomize_hosts=False) - t._verified_elasticsearch = True t.add_connection({"host": "google.com", "port": 1234}) assert 2 == len(t.connection_pool.connections) @@ -312,7 +285,6 @@ class TestTransport: [{"exception": ConnectionError("abandon ship")}], connection_class=DummyConnection, ) - t._verified_elasticsearch = True connection_error = False try: @@ -328,7 +300,6 @@ class TestTransport: [{"exception": ConnectionError("abandon ship")}] * 2, connection_class=DummyConnection, ) - t._verified_elasticsearch = True connection_error = False try: @@ -342,8 +313,6 @@ class TestTransport: async def test_resurrected_connection_will_be_marked_as_live_on_success(self): for method in ("GET", "HEAD"): t = AsyncTransport([{}, {}], connection_class=DummyConnection) - t._verified_elasticsearch = True - await t._async_call() con1 = t.connection_pool.get_connection() con2 = t.connection_pool.get_connection() @@ -369,7 +338,6 @@ class TestTransport: connection_class=DummyConnection, sniff_on_start=True, ) - await t._async_call() await t.sniffing_task # Need to wait for the sniffing task to complete @@ -383,7 +351,6 @@ class TestTransport: sniff_on_start=True, sniff_timeout=12, ) - await t._async_call() await t.sniffing_task # Need to wait for the sniffing task to complete @@ -426,8 +393,6 @@ class TestTransport: max_retries=0, randomize_hosts=False, ) - t._verified_elasticsearch = True - await t._async_call() connection_error = False @@ -452,7 +417,6 @@ class TestTransport: max_retries=3, randomize_hosts=False, ) - t._verified_elasticsearch = True await t._async_init() conn_err, conn_data = t.connection_pool.connections @@ -468,7 +432,6 @@ class TestTransport: connection_class=DummyConnection, sniffer_timeout=5, ) - t._verified_elasticsearch = True await t._async_call() for _ in range(4): @@ -492,9 +455,7 @@ class TestTransport: connection_class=DummyConnection, sniff_timeout=42, ) - t._verified_elasticsearch = True await t._async_call() - await t.sniff_hosts() # Ensure we parsed out the fqdn and port from the fqdn/ip:port string. assert t.connection_pool.connection_opts[0][1] == { @@ -511,7 +472,6 @@ class TestTransport: connection_class=DummyConnection, cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==", ) - t._verified_elasticsearch = True await t._async_call() assert not t.sniff_on_connection_fail @@ -522,7 +482,6 @@ class TestTransport: async def test_transport_close_closes_all_pool_connections(self): t = AsyncTransport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True await t._async_call() assert not any([conn.closed for conn in t.connection_pool.connections]) @@ -530,7 +489,6 @@ class TestTransport: assert all([conn.closed for conn in t.connection_pool.connections]) t = AsyncTransport([{}, {}], connection_class=DummyConnection) - t._verified_elasticsearch = True await t._async_call() assert not any([conn.closed for conn in t.connection_pool.connections]) @@ -547,7 +505,6 @@ class TestTransport: connection_class=DummyConnection, sniff_on_start=True, ) - t._verified_elasticsearch = True # If our initial sniffing attempt comes back # empty then we raise an error. @@ -565,7 +522,6 @@ class TestTransport: connection_class=DummyConnection, sniff_on_start=True, ) - t._verified_elasticsearch = True # Start the timer right before the first task # and have a bunch of tasks come in immediately. @@ -600,7 +556,6 @@ class TestTransport: connection_class=DummyConnection, sniff_on_start=True, ) - t._verified_elasticsearch = True # Start making _async_calls() before we cancel tasks = [] @@ -619,286 +574,3 @@ class TestTransport: # A lot quicker than 10 seconds defined in 'delay' assert duration < 1 - - @pytest.mark.parametrize( - ["headers", "data"], - [ - ( - {}, - '{"version":{"number":"6.99.0"},"tagline":"You Know, for Search"}', - ), - ( - {}, - '{"version":{"number":"7.13.0","build_flavor":"default"},"tagline":"You Know, for Search"}', - ), - ( - {"X-elastic-product": "Elasticsearch"}, - '{"version":{"number":"7.14.0","build_flavor":"default"},"tagline":"You Know, for Search"}', - ), - ], - ) - async def test_verify_elasticsearch(self, headers, data): - t = AsyncTransport( - [{"data": data, "headers": headers}], connection_class=DummyConnection - ) - await t.perform_request("GET", "/_search") - assert t._verified_elasticsearch is True - - calls = t.connection_pool.connections[0].calls - _ = [call[1]["headers"].pop("x-elastic-client-meta") for call in calls] - - assert calls == [ - ( - ("GET", "/"), - { - "headers": { - "accept": "application/json", - }, - "timeout": None, - }, - ), - ( - ("GET", "/_search", None, None), - { - "headers": {}, - "ignore": (), - "timeout": None, - }, - ), - ] - - @pytest.mark.parametrize( - "exception_cls", [AuthorizationException, AuthenticationException] - ) - async def test_verify_elasticsearch_skips_on_auth_errors(self, exception_cls): - t = AsyncTransport( - [{"exception": exception_cls(exception_cls.status_code)}], - connection_class=DummyConnection, - ) - - with pytest.warns(ElasticsearchWarning) as warns: - with pytest.raises(exception_cls): - await t.perform_request( - "GET", "/_search", headers={"Authorization": "testme"} - ) - - # Assert that a warning was raised due to security privileges - assert [str(w.message) for w in warns] == [ - "The client is unable to verify that the server is " - "Elasticsearch due security privileges on the server side" - ] - - # Assert that the cluster is "verified" - assert t._verified_elasticsearch is True - - # See that the headers were passed along to the "info" request made - calls = t.connection_pool.connections[0].calls - _ = [call[1]["headers"].pop("x-elastic-client-meta") for call in calls] - - assert calls == [ - ( - ("GET", "/"), - { - "headers": { - "accept": "application/json", - "authorization": "testme", - }, - "timeout": None, - }, - ), - ( - ("GET", "/_search", None, None), - { - "headers": { - "Authorization": "testme", - }, - "ignore": (), - "timeout": None, - }, - ), - ] - - async def test_multiple_requests_verify_elasticsearch_success(self, event_loop): - t = AsyncTransport( - [ - { - "data": '{"version":{"number":"7.13.0","build_flavor":"default"},"tagline":"You Know, for Search"}', - "delay": 1, - } - ], - connection_class=DummyConnection, - ) - - results = [] - completed_at = [] - - async def request_task(): - try: - results.append(await t.perform_request("GET", "/_search")) - except Exception as e: - results.append(e) - completed_at.append(event_loop.time()) - - # Execute a bunch of requests concurrently. - tasks = [] - start_time = event_loop.time() - for _ in range(10): - tasks.append(event_loop.create_task(request_task())) - await asyncio.gather(*tasks) - end_time = event_loop.time() - - # Exactly 10 results completed - assert len(results) == 10 - - # No errors in the results - assert all(isinstance(result, dict) for result in results) - - # Assert that this took longer than 2 seconds but less than 2.1 seconds - duration = end_time - start_time - assert 2 <= duration <= 2.1 - - # Assert that every result came after ~2 seconds, no fast completions. - assert all( - 2 <= completed_time - start_time <= 2.1 for completed_time in completed_at - ) - - # Assert that the cluster is "verified" - assert t._verified_elasticsearch is True - - # See that the first request is always 'GET /' for ES check - calls = t.connection_pool.connections[0].calls - assert calls[0][0] == ("GET", "/") - - # The rest of the requests are 'GET /_search' afterwards - assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:]) - - @pytest.mark.parametrize( - ["build_flavor", "tagline", "product_error", "error_message"], - [ - ( - "default", - "BAD TAGLINE", - _ProductChecker.UNSUPPORTED_PRODUCT, - "The client noticed that the server is not Elasticsearch and we do not support this unknown product", - ), - ( - "BAD BUILD FLAVOR", - "BAD TAGLINE", - _ProductChecker.UNSUPPORTED_PRODUCT, - "The client noticed that the server is not Elasticsearch and we do not support this unknown product", - ), - ( - "BAD BUILD FLAVOR", - "You Know, for Search", - _ProductChecker.UNSUPPORTED_DISTRIBUTION, - "The client noticed that the server is not a supported distribution of Elasticsearch", - ), - ], - ) - async def test_multiple_requests_verify_elasticsearch_product_error( - self, event_loop, build_flavor, tagline, product_error, error_message - ): - t = AsyncTransport( - [ - { - "data": '{"version":{"number":"7.13.0","build_flavor":"%s"},"tagline":"%s"}' - % (build_flavor, tagline), - "delay": 1, - } - ], - connection_class=DummyConnection, - ) - - results = [] - completed_at = [] - - async def request_task(): - try: - results.append(await t.perform_request("GET", "/_search")) - except Exception as e: - results.append(e) - completed_at.append(event_loop.time()) - - # Execute a bunch of requests concurrently. - tasks = [] - start_time = event_loop.time() - for _ in range(10): - tasks.append(event_loop.create_task(request_task())) - await asyncio.gather(*tasks) - end_time = event_loop.time() - - # Exactly 10 results completed - assert len(results) == 10 - - # All results were errors - assert all(isinstance(result, UnsupportedProductError) for result in results) - assert all(str(result) == error_message for result in results) - - # Assert that one request was made but not 2 requests. - duration = end_time - start_time - assert 1 <= duration <= 1.1 - - # Assert that every result came after ~1 seconds, no fast completions. - assert all( - 1 <= completed_time - start_time <= 1.1 for completed_time in completed_at - ) - - # Assert that the cluster is definitely not Elasticsearch - assert t._verified_elasticsearch == product_error - - # See that the first request is always 'GET /' for ES check - calls = t.connection_pool.connections[0].calls - assert calls[0][0] == ("GET", "/") - - # The rest of the requests are 'GET /_search' afterwards - assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:]) - - @pytest.mark.parametrize("error_cls", [ConnectionError, NotFoundError]) - async def test_multiple_requests_verify_elasticsearch_retry_on_errors( - self, event_loop, error_cls - ): - t = AsyncTransport( - [ - { - "exception": error_cls(), - "delay": 0.1, - } - ], - connection_class=DummyConnection, - ) - - results = [] - completed_at = [] - - async def request_task(): - try: - results.append(await t.perform_request("GET", "/_search")) - except Exception as e: - results.append(e) - completed_at.append(event_loop.time()) - - # Execute a bunch of requests concurrently. - tasks = [] - start_time = event_loop.time() - for _ in range(5): - tasks.append(event_loop.create_task(request_task())) - await asyncio.gather(*tasks) - end_time = event_loop.time() - - # Exactly 5 results completed - assert len(results) == 5 - - # All results were errors and not wrapped in 'NotElasticsearchError' - assert all(isinstance(result, error_cls) for result in results) - - # Assert that 5 requests were made in total (5 transport requests per x 0.1s/conn request) - duration = end_time - start_time - assert 0.5 <= duration <= 0.6 - - # Assert that the cluster is still in the unknown/unverified stage. - assert t._verified_elasticsearch is None - - # See that the API isn't hit, instead it's the index requests that are failing. - calls = t.connection_pool.connections[0].calls - assert len(calls) == 5 - assert all(call[0] == ("GET", "/") for call in calls) diff --git a/test_elasticsearch/test_connection.py b/test_elasticsearch/test_connection.py index 5a1f5449..2ade950b 100644 --- a/test_elasticsearch/test_connection.py +++ b/test_elasticsearch/test_connection.py @@ -31,7 +31,7 @@ from mock import Mock, patch from requests.auth import AuthBase from urllib3._collections import HTTPHeaderDict -from elasticsearch import Elasticsearch, __versionstr__ +from elasticsearch import __versionstr__ from elasticsearch.compat import reraise_exceptions from elasticsearch.connection import ( Connection, @@ -1045,9 +1045,3 @@ class TestConnectionHttpbin: conn = RequestsHttpConnection("not.a.host.name") with pytest.raises(ConnectionError): conn.perform_request("GET", "/") - - def test_elasticsearch_connection_error(self): - es = Elasticsearch("http://not.a.host.name") - - with pytest.raises(ConnectionError): - es.search() diff --git a/test_elasticsearch/test_transport.py b/test_elasticsearch/test_transport.py index a9a92401..c5cd3f0c 100644 --- a/test_elasticsearch/test_transport.py +++ b/test_elasticsearch/test_transport.py @@ -26,16 +26,8 @@ from mock import patch from elasticsearch.connection import Connection from elasticsearch.connection_pool import DummyConnectionPool -from elasticsearch.exceptions import ( - AuthenticationException, - AuthorizationException, - ConnectionError, - ElasticsearchWarning, - NotFoundError, - TransportError, - UnsupportedProductError, -) -from elasticsearch.transport import Transport, _ProductChecker, get_host_info +from elasticsearch.exceptions import ConnectionError, TransportError +from elasticsearch.transport import Transport, get_host_info from .test_cases import TestCase @@ -45,14 +37,11 @@ class DummyConnection(Connection): self.exception = kwargs.pop("exception", None) self.status, self.data = kwargs.pop("status", 200), kwargs.pop("data", "{}") self.headers = kwargs.pop("headers", {}) - self.delay = kwargs.pop("delay", None) self.calls = [] super(DummyConnection, self).__init__(**kwargs) def perform_request(self, *args, **kwargs): self.calls.append((args, kwargs)) - if self.delay is not None: - time.sleep(self.delay) if self.exception: raise self.exception return self.status, self.headers, self.data @@ -129,16 +118,12 @@ class TestHostsInfoCallback(TestCase): class TestTransport(TestCase): def test_single_connection_uses_dummy_connection_pool(self): t = Transport([{}]) - t._verified_elasticsearch = True self.assertIsInstance(t.connection_pool, DummyConnectionPool) - t = Transport([{"host": "localhost"}]) - t._verified_elasticsearch = True self.assertIsInstance(t.connection_pool, DummyConnectionPool) def test_request_timeout_extracted_from_params_and_passed(self): t = Transport([{}], meta_header=False, connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", params={"request_timeout": 42}) self.assertEqual(1, len(t.get_connection().calls)) @@ -152,7 +137,6 @@ class TestTransport(TestCase): t = Transport( [{}], opaque_id="app-1", meta_header=False, connection_class=DummyConnection ) - t._verified_elasticsearch = True t.perform_request("GET", "/") self.assertEqual(1, len(t.get_connection().calls)) @@ -173,7 +157,6 @@ class TestTransport(TestCase): def test_request_with_custom_user_agent_header(self): t = Transport([{}], meta_header=False, connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", headers={"user-agent": "my-custom-value/1.2.3"}) self.assertEqual(1, len(t.get_connection().calls)) @@ -188,7 +171,6 @@ class TestTransport(TestCase): def test_send_get_body_as_source(self): t = Transport([{}], send_get_body_as="source", connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", body={}) self.assertEqual(1, len(t.get_connection().calls)) @@ -198,7 +180,6 @@ class TestTransport(TestCase): def test_send_get_body_as_post(self): t = Transport([{}], send_get_body_as="POST", connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", body={}) self.assertEqual(1, len(t.get_connection().calls)) @@ -206,7 +187,6 @@ class TestTransport(TestCase): def test_client_meta_header(self): t = Transport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", body={}) self.assertEqual(1, len(t.get_connection().calls)) @@ -219,7 +199,6 @@ class TestTransport(TestCase): HTTP_CLIENT_META = ("dm", "1.2.3") t = Transport([{}], connection_class=DummyConnectionWithMeta) - t._verified_elasticsearch = True t.perform_request("GET", "/", body={}, headers={"Custom": "header"}) self.assertEqual(1, len(t.get_connection().calls)) @@ -232,7 +211,6 @@ class TestTransport(TestCase): def test_client_meta_header_not_sent(self): t = Transport([{}], meta_header=False, connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", body={}) self.assertEqual(1, len(t.get_connection().calls)) @@ -246,7 +224,6 @@ class TestTransport(TestCase): def test_body_gets_encoded_into_bytes(self): t = Transport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", body="你好") self.assertEqual(1, len(t.get_connection().calls)) @@ -257,7 +234,6 @@ class TestTransport(TestCase): def test_body_bytes_get_passed_untouched(self): t = Transport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True body = b"\xe4\xbd\xa0\xe5\xa5\xbd" t.perform_request("GET", "/", body=body) @@ -266,7 +242,6 @@ class TestTransport(TestCase): def test_body_surrogates_replaced_encoded_into_bytes(self): t = Transport([{}], connection_class=DummyConnection) - t._verified_elasticsearch = True t.perform_request("GET", "/", body="你好\uda6a") self.assertEqual(1, len(t.get_connection().calls)) @@ -277,14 +252,12 @@ class TestTransport(TestCase): def test_kwargs_passed_on_to_connections(self): t = Transport([{"host": "google.com"}], port=123) - t._verified_elasticsearch = True self.assertEqual(1, len(t.connection_pool.connections)) self.assertEqual("http://google.com:123", t.connection_pool.connections[0].host) def test_kwargs_passed_on_to_connection_pool(self): dt = object() t = Transport([{}, {}], dead_timeout=dt) - t._verified_elasticsearch = True self.assertIs(dt, t.connection_pool.dead_timeout) def test_custom_connection_class(self): @@ -293,13 +266,11 @@ class TestTransport(TestCase): self.kwargs = kwargs t = Transport([{}], connection_class=MyConnection) - t._verified_elasticsearch = True self.assertEqual(1, len(t.connection_pool.connections)) self.assertIsInstance(t.connection_pool.connections[0], MyConnection) def test_add_connection(self): t = Transport([{}], randomize_hosts=False) - t._verified_elasticsearch = True t.add_connection({"host": "google.com", "port": 1234}) self.assertEqual(2, len(t.connection_pool.connections)) @@ -312,7 +283,6 @@ class TestTransport(TestCase): [{"exception": ConnectionError("abandon ship")}], connection_class=DummyConnection, ) - t._verified_elasticsearch = True self.assertRaises(ConnectionError, t.perform_request, "GET", "/") self.assertEqual(4, len(t.get_connection().calls)) @@ -322,7 +292,6 @@ class TestTransport(TestCase): [{"exception": ConnectionError("abandon ship")}] * 2, connection_class=DummyConnection, ) - t._verified_elasticsearch = True self.assertRaises(ConnectionError, t.perform_request, "GET", "/") self.assertEqual(0, len(t.connection_pool.connections)) @@ -330,7 +299,6 @@ class TestTransport(TestCase): def test_resurrected_connection_will_be_marked_as_live_on_success(self): for method in ("GET", "HEAD"): t = Transport([{}, {}], connection_class=DummyConnection) - t._verified_elasticsearch = True con1 = t.connection_pool.get_connection() con2 = t.connection_pool.get_connection() t.connection_pool.mark_dead(con1) @@ -342,7 +310,6 @@ class TestTransport(TestCase): def test_sniff_will_use_seed_connections(self): t = Transport([{"data": CLUSTER_NODES}], connection_class=DummyConnection) - t._verified_elasticsearch = True t.set_connections([{"data": "invalid"}]) t.sniff_hosts() @@ -355,7 +322,6 @@ class TestTransport(TestCase): connection_class=DummyConnection, sniff_on_start=True, ) - t._verified_elasticsearch = True self.assertEqual(1, len(t.connection_pool.connections)) self.assertEqual("http://1.1.1.1:123", t.get_connection().host) @@ -366,7 +332,6 @@ class TestTransport(TestCase): sniff_on_start=True, sniff_timeout=12, ) - t._verified_elasticsearch = True self.assertEqual( (("GET", "/_nodes/_all/http"), {"timeout": None}), t.seed_connections[0].calls[0], @@ -378,7 +343,6 @@ class TestTransport(TestCase): connection_class=DummyConnection, sniff_timeout=42, ) - t._verified_elasticsearch = True t.sniff_hosts() self.assertEqual( (("GET", "/_nodes/_all/http"), {"timeout": 42}), @@ -391,7 +355,6 @@ class TestTransport(TestCase): connection_class=DummyConnection, randomize_hosts=False, ) - t._verified_elasticsearch = True connection = t.connection_pool.connections[1] t.sniff_hosts() @@ -406,7 +369,6 @@ class TestTransport(TestCase): max_retries=0, randomize_hosts=False, ) - t._verified_elasticsearch = True self.assertRaises(ConnectionError, t.perform_request, "GET", "/") self.assertEqual(1, len(t.connection_pool.connections)) @@ -422,7 +384,6 @@ class TestTransport(TestCase): max_retries=3, randomize_hosts=False, ) - t._verified_elasticsearch = True conn_err, conn_data = t.connection_pool.connections response = t.perform_request("GET", "/") @@ -437,7 +398,6 @@ class TestTransport(TestCase): connection_class=DummyConnection, sniffer_timeout=5, ) - t._verified_elasticsearch = True for _ in range(4): t.perform_request("GET", "/") @@ -458,7 +418,6 @@ class TestTransport(TestCase): connection_class=DummyConnection, sniff_timeout=42, ) - t._verified_elasticsearch = True t.sniff_hosts() # Ensure we parsed out the fqdn and port from the fqdn/ip:port string. self.assertEqual( @@ -474,480 +433,6 @@ class TestTransport(TestCase): sniff_on_connection_fail=True, cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==", ) - t._verified_elasticsearch = True self.assertFalse(t.sniff_on_connection_fail) self.assertIs(sniff_hosts.call_args, None) # Assert not called. - - -TAGLINE = "You Know, for Search" - - -@pytest.mark.parametrize( - ["headers", "response", "product_error"], - [ - # All empty. - ({}, {}, _ProductChecker.UNSUPPORTED_PRODUCT), - # Don't check the product header immediately, need to check version first. - ( - {"x-elastic-product": "Elasticsearch"}, - {}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Version not there. - ({}, {"tagline": TAGLINE}, _ProductChecker.UNSUPPORTED_PRODUCT), - # Version is nonsense - ( - {}, - {"version": "1.0.0", "tagline": TAGLINE}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Version number not there - ({}, {"version": {}, "tagline": TAGLINE}, _ProductChecker.UNSUPPORTED_PRODUCT), - # Version number is nonsense - ( - {}, - {"version": {"number": "nonsense"}, "tagline": TAGLINE}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Version number way in the past - ( - {}, - {"version": {"number": "1.0.0"}, "tagline": TAGLINE}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Version number way in the future - ( - {}, - {"version": {"number": "999.0.0"}, "tagline": TAGLINE}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Build flavor not supposed to be missing - ( - {}, - {"version": {"number": "7.13.0"}, "tagline": TAGLINE}, - _ProductChecker.UNSUPPORTED_DISTRIBUTION, - ), - # Build flavor is 'oss' - ( - {}, - { - "version": {"number": "7.10.0", "build_flavor": "oss"}, - "tagline": TAGLINE, - }, - _ProductChecker.UNSUPPORTED_DISTRIBUTION, - ), - # Build flavor is nonsense - ( - {}, - { - "version": {"number": "7.13.0", "build_flavor": "nonsense"}, - "tagline": TAGLINE, - }, - _ProductChecker.UNSUPPORTED_DISTRIBUTION, - ), - # Tagline is nonsense - ( - {}, - {"version": {"number": "7.1.0-SNAPSHOT"}, "tagline": "nonsense"}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Product header is not supposed to be missing - ( - {}, - {"version": {"number": "7.14.0"}, "tagline": "You Know, for Search"}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - # Product header is nonsense - ( - {"x-elastic-product": "nonsense"}, - {"version": {"number": "7.15.0"}, "tagline": TAGLINE}, - _ProductChecker.UNSUPPORTED_PRODUCT, - ), - ], -) -def test_verify_elasticsearch_errors(headers, response, product_error): - assert _ProductChecker.check_product(headers, response) == product_error - - -@pytest.mark.parametrize( - ["headers", "response"], - [ - ({}, {"version": {"number": "6.0.0"}, "tagline": TAGLINE}), - ({}, {"version": {"number": "6.99.99"}, "tagline": TAGLINE}), - ( - {}, - { - "version": {"number": "7.0.0", "build_flavor": "default"}, - "tagline": TAGLINE, - }, - ), - ( - {}, - { - "version": {"number": "7.13.99", "build_flavor": "default"}, - "tagline": TAGLINE, - }, - ), - ( - {"x-elastic-product": "Elasticsearch"}, - { - "version": {"number": "7.14.0", "build_flavor": "default"}, - "tagline": TAGLINE, - }, - ), - ( - {"x-elastic-product": "Elasticsearch"}, - { - "version": {"number": "7.99.99", "build_flavor": "default"}, - "tagline": TAGLINE, - }, - ), - ( - {"x-elastic-product": "Elasticsearch"}, - { - "version": {"number": "8.0.0"}, - }, - ), - ], -) -def test_verify_elasticsearch_passes(headers, response): - result = _ProductChecker.check_product(headers, response) - assert result == _ProductChecker.SUCCESS - assert result is True - - -@pytest.mark.parametrize( - ["headers", "data"], - [ - ( - {}, - '{"version":{"number":"6.99.0"},"tagline":"You Know, for Search"}', - ), - ( - {}, - """{ - "name" : "io", - "cluster_name" : "elasticsearch", - "cluster_uuid" : "HaMHUswUSGGnzla8B17Iqw", - "version" : { - "number" : "7.6.0", - "build_flavor" : "default", - "build_type" : "tar", - "build_hash" : "7f634e9f44834fbc12724506cc1da681b0c3b1e3", - "build_date" : "2020-02-06T00:09:00.449973Z", - "build_snapshot" : false, - "lucene_version" : "8.4.0", - "minimum_wire_compatibility_version" : "6.8.0", - "minimum_index_compatibility_version" : "6.0.0-beta1" - }, - "tagline" : "You Know, for Search" -}""", - ), - ( - {}, - '{"version":{"number":"7.13.0","build_flavor":"default"},"tagline":"You Know, for Search"}', - ), - ( - {"X-elastic-product": "Elasticsearch"}, - '{"version":{"number":"7.14.0","build_flavor":"default"},"tagline":"You Know, for Search"}', - ), - ], -) -def test_verify_elasticsearch(headers, data): - t = Transport( - [{"data": data, "headers": headers}], connection_class=DummyConnection - ) - t.perform_request("GET", "/_search") - assert t._verified_elasticsearch is True - - calls = t.connection_pool.connections[0].calls - _ = [call[1]["headers"].pop("x-elastic-client-meta") for call in calls] - - assert calls == [ - ( - ("GET", "/"), - { - "headers": { - "accept": "application/json", - }, - "timeout": None, - }, - ), - ( - ("GET", "/_search", None, None), - { - "headers": {}, - "ignore": (), - "timeout": None, - }, - ), - ] - - -@pytest.mark.parametrize( - "exception_cls", [AuthorizationException, AuthenticationException] -) -def test_verify_elasticsearch_skips_on_auth_errors(exception_cls): - t = Transport( - [{"exception": exception_cls(exception_cls.status_code)}], - connection_class=DummyConnection, - ) - - with pytest.warns(ElasticsearchWarning) as warns: - with pytest.raises(exception_cls): - t.perform_request( - "GET", - "/_search", - headers={"Authorization": "testme"}, - params={"request_timeout": 3}, - ) - - # Assert that a warning was raised due to security privileges - assert [str(w.message) for w in warns] == [ - "The client is unable to verify that the server is " - "Elasticsearch due security privileges on the server side" - ] - - # Assert that the cluster is "verified" - assert t._verified_elasticsearch is True - - # See that the headers were passed along to the "info" request made - calls = t.connection_pool.connections[0].calls - _ = [call[1]["headers"].pop("x-elastic-client-meta") for call in calls] - - assert calls == [ - ( - ("GET", "/"), - { - "headers": { - "accept": "application/json", - "authorization": "testme", - }, - "timeout": 3, - }, - ), - ( - ("GET", "/_search", {}, None), - { - "headers": { - "Authorization": "testme", - }, - "ignore": (), - "timeout": 3, - }, - ), - ] - - -def test_multiple_requests_verify_elasticsearch_success(): - try: - import threading - except ImportError: - return pytest.skip("Requires the 'threading' module") - - t = Transport( - [ - { - "data": '{"version":{"number":"7.13.0","build_flavor":"default"},"tagline":"You Know, for Search"}', - "delay": 1, - } - ], - connection_class=DummyConnection, - ) - - results = [] - completed_at = [] - - class RequestThread(threading.Thread): - def run(self): - try: - results.append(t.perform_request("GET", "/_search")) - except Exception as e: - results.append(e) - completed_at.append(time.time()) - - # Execute a bunch of requests concurrently. - threads = [] - start_time = time.time() - for _ in range(10): - thread = RequestThread() - thread.start() - threads.append(thread) - for thread in threads: - thread.join() - end_time = time.time() - - # Exactly 10 results completed - assert len(results) == 10 - - # No errors in the results - assert all(isinstance(result, dict) for result in results) - - # Assert that this took longer than 2 seconds but less than 2.1 seconds - duration = end_time - start_time - assert 2 <= duration <= 2.1 - - # Assert that every result came after ~2 seconds, no fast completions. - assert all( - 2 <= completed_time - start_time <= 2.1 for completed_time in completed_at - ) - - # Assert that the cluster is "verified" - assert t._verified_elasticsearch is True - - # See that the first request is always 'GET /' for ES check - calls = t.connection_pool.connections[0].calls - assert calls[0][0] == ("GET", "/") - - # The rest of the requests are 'GET /_search' afterwards - assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:]) - - -@pytest.mark.parametrize( - ["build_flavor", "tagline", "product_error", "error_message"], - [ - ( - "default", - "BAD TAGLINE", - _ProductChecker.UNSUPPORTED_PRODUCT, - "The client noticed that the server is not Elasticsearch and we do not support this unknown product", - ), - ( - "BAD BUILD FLAVOR", - "BAD TAGLINE", - _ProductChecker.UNSUPPORTED_PRODUCT, - "The client noticed that the server is not Elasticsearch and we do not support this unknown product", - ), - ( - "BAD BUILD FLAVOR", - "You Know, for Search", - _ProductChecker.UNSUPPORTED_DISTRIBUTION, - "The client noticed that the server is not a supported distribution of Elasticsearch", - ), - ], -) -def test_multiple_requests_verify_elasticsearch_product_error( - build_flavor, tagline, product_error, error_message -): - try: - import threading - except ImportError: - return pytest.skip("Requires the 'threading' module") - - t = Transport( - [ - { - "data": '{"version":{"number":"7.13.0","build_flavor":"%s"},"tagline":"%s"}' - % (build_flavor, tagline), - "delay": 1, - } - ], - connection_class=DummyConnection, - ) - - results = [] - completed_at = [] - - class RequestThread(threading.Thread): - def run(self): - try: - results.append(t.perform_request("GET", "/_search")) - except Exception as e: - results.append(e) - completed_at.append(time.time()) - - # Execute a bunch of requests concurrently. - threads = [] - start_time = time.time() - for _ in range(10): - thread = RequestThread() - thread.start() - threads.append(thread) - for thread in threads: - thread.join() - end_time = time.time() - - # Exactly 10 results completed - assert len(results) == 10 - - # All results were errors - assert all(isinstance(result, UnsupportedProductError) for result in results) - assert all(str(result) == error_message for result in results) - - # Assert that one request was made but not 2 requests. - duration = end_time - start_time - assert 1 <= duration <= 1.1 - - # Assert that every result came after ~1 seconds, no fast completions. - assert all( - 1 <= completed_time - start_time <= 1.1 for completed_time in completed_at - ) - - # Assert that the cluster is definitely not Elasticsearch - assert t._verified_elasticsearch == product_error - - # See that the first request is always 'GET /' for ES check - calls = t.connection_pool.connections[0].calls - assert calls[0][0] == ("GET", "/") - - # The rest of the requests are 'GET /_search' afterwards - assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:]) - - -@pytest.mark.parametrize("error_cls", [ConnectionError, NotFoundError]) -def test_multiple_requests_verify_elasticsearch_retry_on_errors(error_cls): - try: - import threading - except ImportError: - return pytest.skip("Requires the 'threading' module") - - t = Transport( - [ - { - "exception": error_cls(), - "delay": 0.1, - } - ], - connection_class=DummyConnection, - ) - - results = [] - completed_at = [] - - class RequestThread(threading.Thread): - def run(self): - try: - results.append(t.perform_request("GET", "/_search")) - except Exception as e: - results.append(e) - completed_at.append(time.time()) - - # Execute a bunch of requests concurrently. - threads = [] - start_time = time.time() - for _ in range(5): - thread = RequestThread() - thread.start() - threads.append(thread) - for thread in threads: - thread.join() - end_time = time.time() - - # Exactly 5 results completed - assert len(results) == 5 - - # All results were errors and not wrapped in 'UnsupportedProductError' - assert all(isinstance(result, error_cls) for result in results) - - # Assert that 5 requests were made in total (5 transport requests per x 0.1s/conn request) - duration = end_time - start_time - assert 0.5 <= duration <= 0.6 - - # Assert that the cluster is still in the unknown/unverified stage. - assert t._verified_elasticsearch is None - - # See that the API isn't hit, instead it's the index requests that are failing. - calls = t.connection_pool.connections[0].calls - assert len(calls) == 5 - assert all(call[0] == ("GET", "/") for call in calls)