actions/opensearch-pyd
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Revert product checker commits:

Browse Source 
023cf5380f: "[7.x] Document 'UnsupportedProductError'"
	d381491482: "[7.x] Rename product error to 'UnsupportedProductError'"
	b56fbcb155: "Document the Elasticsearch product check"
	b63d005138: "Don't swallow unexpected errors during Elasticsearch verification"
	801a839093: "Verify we're connected to Elasticsearch before requests"

Signed-off-by: Shephali Mittal <shephalm@amazon.com>
This commit is contained in:
Shephali Mittal
2021-08-06 12:17:51 +05:30
parent 18b39afcbf
commit 934ea8cc5e
13 changed files with 6 additions and 1192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/sphinx/exceptions.rst
-1
View File
@@ -23,4 +23,3 @@ Exceptions
.. autoclass:: RequestError(TransportError)
.. autoclass:: AuthenticationException(TransportError)
.. autoclass:: AuthorizationException(TransportError)
.. autoclass:: UnsupportedProductError
docs/sphinx/transports.rst
-21
View File
@@ -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
elasticsearch/__init__.py
-2
View File
@@ -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",
]
elasticsearch/__init__.pyi
-1
View File
@@ -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
elasticsearch/_async/transport.py
+1 -98
View File
@@ -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
)
elasticsearch/compat.py
-11
View File
@@ -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",
elasticsearch/exceptions.py
-6
View File
@@ -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
elasticsearch/exceptions.pyi
-1
View File
@@ -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
elasticsearch/transport.py
-191
View File
@@ -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
test_elasticsearch/test_async/test_connection.py
+1 -7
View File
@@ -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()
test_elasticsearch/test_async/test_transport.py
+1 -329
View File
@@ -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)
test_elasticsearch/test_connection.py
+1 -7
View File
@@ -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()
test_elasticsearch/test_transport.py
+2 -517
View File
@@ -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)