diff --git a/test_elasticsearch/test_async/test_server/__init__.py b/test_elasticsearch/test_async/test_server/__init__.py index 47633799..1a3c439e 100644 --- a/test_elasticsearch/test_async/test_server/__init__.py +++ b/test_elasticsearch/test_async/test_server/__init__.py @@ -1,4 +1,3 @@ # Licensed to Elasticsearch B.V under one or more agreements. # Elasticsearch B.V licenses this file to you under the Apache 2.0 License. # See the LICENSE file in the project root for more information - diff --git a/test_elasticsearch/test_async/test_server/conftest.py b/test_elasticsearch/test_async/test_server/conftest.py index 47633799..f97b1762 100644 --- a/test_elasticsearch/test_async/test_server/conftest.py +++ b/test_elasticsearch/test_async/test_server/conftest.py @@ -2,3 +2,60 @@ # Elasticsearch B.V licenses this file to you under the Apache 2.0 License. # See the LICENSE file in the project root for more information +import os +import pytest +import asyncio +import elasticsearch + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture(scope="function") +async def async_client(): + client = None + try: + if not hasattr(elasticsearch, "AsyncElasticsearch"): + pytest.skip("test requires 'AsyncElasticsearch'") + + kw = { + "timeout": 30, + "ca_certs": ".ci/certs/ca.pem", + "connection_class": elasticsearch.AIOHttpConnection, + } + + client = elasticsearch.AsyncElasticsearch( + [os.environ.get("ELASTICSEARCH_HOST", {})], **kw + ) + + # wait for yellow status + for _ in range(100): + try: + await client.cluster.health(wait_for_status="yellow") + break + except ConnectionError: + await asyncio.sleep(0.1) + else: + # timeout + pytest.skip("Elasticsearch failed to start.") + + yield client + + finally: + if client: + version = tuple( + [ + int(x) if x.isdigit() else 999 + for x in (await client.info())["version"]["number"].split(".") + ] + ) + + expand_wildcards = ["open", "closed"] + if version >= (7, 7): + expand_wildcards.append("hidden") + + await client.indices.delete( + index="*", ignore=404, expand_wildcards=expand_wildcards + ) + await client.indices.delete_template(name="*", ignore=404) + await client.indices.delete_index_template(name="*", ignore=404) + await client.close() diff --git a/test_elasticsearch/test_async/test_server/test_clients.py b/test_elasticsearch/test_async/test_server/test_clients.py index 47633799..c7b4279a 100644 --- a/test_elasticsearch/test_async/test_server/test_clients.py +++ b/test_elasticsearch/test_async/test_server/test_clients.py @@ -1,4 +1,30 @@ +# -*- coding: utf-8 -*- # Licensed to Elasticsearch B.V under one or more agreements. # Elasticsearch B.V licenses this file to you under the Apache 2.0 License. # See the LICENSE file in the project root for more information +from __future__ import unicode_literals +import pytest + +pytestmark = pytest.mark.asyncio + + +class TestUnicode: + async def test_indices_analyze(self, async_client): + await async_client.indices.analyze(body='{"text": "привет"}') + + +class TestBulk: + async def test_bulk_works_with_string_body(self, async_client): + docs = '{ "index" : { "_index" : "bulk_test_index", "_id" : "1" } }\n{"answer": 42}' + response = await async_client.bulk(body=docs) + + assert response["errors"] is False + assert len(response["items"]) == 1 + + async def test_bulk_works_with_bytestring_body(self, async_client): + docs = b'{ "index" : { "_index" : "bulk_test_index", "_id" : "2" } }\n{"answer": 42}' + response = await async_client.bulk(body=docs) + + assert response["errors"] is False + assert len(response["items"]) == 1 diff --git a/test_elasticsearch/test_async/test_server/test_rest_api_spec.py b/test_elasticsearch/test_async/test_server/test_rest_api_spec.py index 47633799..d849f46d 100644 --- a/test_elasticsearch/test_async/test_server/test_rest_api_spec.py +++ b/test_elasticsearch/test_async/test_server/test_rest_api_spec.py @@ -2,3 +2,209 @@ # Elasticsearch B.V licenses this file to you under the Apache 2.0 License. # See the LICENSE file in the project root for more information +""" +Dynamically generated set of TestCases based on set of yaml files decribing +some integration tests. These files are shared among all official Elasticsearch +clients. +""" +import pytest +from shutil import rmtree +import warnings +import inspect + +from elasticsearch import RequestError, ElasticsearchDeprecationWarning +from elasticsearch.helpers.test import _get_version +from ...test_server.test_rest_api_spec import ( + YamlRunner, + YAML_TEST_SPECS, + InvalidActionType, + RUN_ASYNC_REST_API_TESTS, + PARAMS_RENAMES, + IMPLEMENTED_FEATURES, +) + +pytestmark = pytest.mark.asyncio + +XPACK_FEATURES = None +ES_VERSION = None + + +async def await_if_coro(x): + if inspect.iscoroutine(x): + return await x + return x + + +class AsyncYamlRunner(YamlRunner): + async def setup(self): + if self._setup_code: + await self.run_code(self._setup_code) + + async def teardown(self): + if self._teardown_code: + await self.run_code(self._teardown_code) + + for repo, definition in ( + await self.client.snapshot.get_repository(repository="_all") + ).items(): + await self.client.snapshot.delete_repository(repository=repo) + if definition["type"] == "fs": + rmtree( + "/tmp/%s" % definition["settings"]["location"], ignore_errors=True + ) + + # stop and remove all ML stuff + if await self._feature_enabled("ml"): + await self.client.ml.stop_datafeed(datafeed_id="*", force=True) + for feed in (await self.client.ml.get_datafeeds(datafeed_id="*"))[ + "datafeeds" + ]: + await self.client.ml.delete_datafeed(datafeed_id=feed["datafeed_id"]) + + await self.client.ml.close_job(job_id="*", force=True) + for job in (await self.client.ml.get_jobs(job_id="*"))["jobs"]: + await self.client.ml.delete_job( + job_id=job["job_id"], wait_for_completion=True, force=True + ) + + # stop and remove all Rollup jobs + if await self._feature_enabled("rollup"): + for rollup in (await self.client.rollup.get_jobs(id="*"))["jobs"]: + await self.client.rollup.stop_job( + id=rollup["config"]["id"], wait_for_completion=True + ) + await self.client.rollup.delete_job(id=rollup["config"]["id"]) + + async def es_version(self): + global ES_VERSION + if ES_VERSION is None: + version_string = (await self.client.info())["version"]["number"] + if "." not in version_string: + return () + version = version_string.strip().split(".") + ES_VERSION = tuple(int(v) if v.isdigit() else 999 for v in version) + return ES_VERSION + + async def run(self): + try: + await self.setup() + await self.run_code(self._run_code) + finally: + await self.teardown() + + async def run_code(self, test): + """ Execute an instruction based on it's type. """ + print(test) + for action in test: + assert len(action) == 1 + action_type, action = list(action.items())[0] + + if hasattr(self, "run_" + action_type): + await await_if_coro(getattr(self, "run_" + action_type)(action)) + else: + raise InvalidActionType(action_type) + + async def run_do(self, action): + api = self.client + headers = action.pop("headers", None) + catch = action.pop("catch", None) + warn = action.pop("warnings", ()) + allowed_warnings = action.pop("allowed_warnings", ()) + assert len(action) == 1 + + method, args = list(action.items())[0] + args["headers"] = headers + + # locate api endpoint + for m in method.split("."): + assert hasattr(api, m) + api = getattr(api, m) + + # some parameters had to be renamed to not clash with python builtins, + # compensate + for k in PARAMS_RENAMES: + if k in args: + args[PARAMS_RENAMES[k]] = args.pop(k) + + # resolve vars + for k in args: + args[k] = self._resolve(args[k]) + + warnings.simplefilter("always", category=ElasticsearchDeprecationWarning) + with warnings.catch_warnings(record=True) as caught_warnings: + try: + self.last_response = await api(**args) + except Exception as e: + if not catch: + raise + self.run_catch(catch, e) + else: + if catch: + raise AssertionError( + "Failed to catch %r in %r." % (catch, self.last_response) + ) + + # Filter out warnings raised by other components. + caught_warnings = [ + str(w.message) + for w in caught_warnings + if w.category == ElasticsearchDeprecationWarning + and str(w.message) not in allowed_warnings + ] + + # Sorting removes the issue with order raised. We only care about + # if all warnings are raised in the single API call. + if warn and sorted(warn) != sorted(caught_warnings): + raise AssertionError( + "Expected warnings not equal to actual warnings: expected=%r actual=%r" + % (warn, caught_warnings) + ) + + async def run_skip(self, skip): + if "features" in skip: + features = skip["features"] + if not isinstance(features, (tuple, list)): + features = [features] + for feature in features: + if feature in IMPLEMENTED_FEATURES: + continue + pytest.skip("feature '%s' is not supported" % feature) + + if "version" in skip: + version, reason = skip["version"], skip["reason"] + if version == "all": + pytest.skip(reason) + min_version, max_version = version.split("-") + min_version = _get_version(min_version) or (0,) + max_version = _get_version(max_version) or (999,) + if min_version <= (await self.es_version()) <= max_version: + pytest.skip(reason) + + async def _feature_enabled(self, name): + global XPACK_FEATURES + if XPACK_FEATURES is None: + try: + xinfo = await self.client.xpack.info() + XPACK_FEATURES = set( + f for f in xinfo["features"] if xinfo["features"][f]["enabled"] + ) + IMPLEMENTED_FEATURES.add("xpack") + except RequestError: + XPACK_FEATURES = set() + IMPLEMENTED_FEATURES.add("no_xpack") + return name in XPACK_FEATURES + + +@pytest.fixture(scope="function") +def async_runner(async_client): + return AsyncYamlRunner(async_client) + + +@pytest.mark.parametrize("test_spec", YAML_TEST_SPECS) +async def test_rest_api_spec(test_spec, async_runner): + if not RUN_ASYNC_REST_API_TESTS: + pytest.skip("Skipped running async REST API tests") + if test_spec.get("skip", False): + pytest.skip("Manually skipped in 'SKIP_TESTS'") + async_runner.use_spec(test_spec) + await async_runner.run() diff --git a/test_elasticsearch/test_connection.py b/test_elasticsearch/test_connection.py index dbdc15f7..a324ee23 100644 --- a/test_elasticsearch/test_connection.py +++ b/test_elasticsearch/test_connection.py @@ -167,6 +167,10 @@ class TestUrllib3Connection(TestCase): self.assertIsInstance(con.pool.conn_kw["ssl_context"], ssl.SSLContext) self.assertTrue(con.use_ssl) + def test_opaque_id(self): + con = Urllib3HttpConnection(opaque_id="app-1") + self.assertEqual(con.headers["x-opaque-id"], "app-1") + def test_http_cloud_id(self): con = Urllib3HttpConnection( cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==" @@ -394,7 +398,7 @@ class TestUrllib3Connection(TestCase): self.assertEqual(2, logger.debug.call_count) req, resp = logger.debug.call_args_list - print(req, resp) + self.assertEqual('> {"example": "body"}', req[0][0] % req[0][1:]) self.assertEqual("< {}", resp[0][0] % resp[0][1:]) @@ -448,6 +452,10 @@ class TestRequestsConnection(TestCase): con = RequestsHttpConnection(timeout=42) self.assertEqual(42, con.timeout) + def test_opaque_id(self): + con = RequestsHttpConnection(opaque_id="app-1") + self.assertEqual(con.headers["x-opaque-id"], "app-1") + def test_http_cloud_id(self): con = RequestsHttpConnection( cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==" diff --git a/test_elasticsearch/test_server/test_rest_api_spec.py b/test_elasticsearch/test_server/test_rest_api_spec.py index cd3d2432..4e296a15 100644 --- a/test_elasticsearch/test_server/test_rest_api_spec.py +++ b/test_elasticsearch/test_server/test_rest_api_spec.py @@ -7,7 +7,9 @@ Dynamically generated set of TestCases based on set of yaml files describing some integration tests. These files are shared among all official Elasticsearch clients. """ +import sys import re +import os from os import walk, environ from os.path import exists, join, dirname, pardir, relpath import yaml @@ -44,6 +46,7 @@ SKIP_TESTS = { # [interval] on [date_histogram] is deprecated, use [fixed_interval] or [calendar_interval] in the future. "search/aggregation/230_composite[6]", "search/aggregation/250_moving_fn[1]", + "search/aggregation/250_moving_fn[2]", # fails by not returning 'search'? "search/320_disallow_queries[2]", "search/40_indices_boost[1]", @@ -58,11 +61,17 @@ SKIP_TESTS = { "indices/put_template/10_basic[4]", # depends on order of response JSON which is random "indices/simulate_index_template/10_basic[1]", + # body: null? body is {} + "indices/simulate_index_template/10_basic[2]", } XPACK_FEATURES = None ES_VERSION = None +RUN_ASYNC_REST_API_TESTS = ( + sys.version_info >= (3, 6) + and os.environ.get("PYTHON_CONNECTION_CLASS") == "RequestsHttpConnection" +) class YamlRunner: @@ -78,7 +87,7 @@ class YamlRunner: def use_spec(self, test_spec): self._setup_code = test_spec.pop("setup", None) self._run_code = test_spec.pop("run", None) - self._teardown_code = test_spec.pop("teardown") + self._teardown_code = test_spec.pop("teardown", None) def setup(self): if self._setup_code: @@ -417,6 +426,8 @@ def sync_runner(sync_client): @pytest.mark.parametrize("test_spec", YAML_TEST_SPECS) def test_rest_api_spec(test_spec, sync_runner): + if RUN_ASYNC_REST_API_TESTS: + pytest.skip("Skipped running sync REST API tests") if test_spec.get("skip", False): pytest.skip("Manually skipped in 'SKIP_TESTS'") sync_runner.use_spec(test_spec)