[7.x] Add test suite for async API

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
-

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()

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

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()

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=="

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)