Add pytest for async REST API tests

Author: sethmlarson

elasticsearch/_async/client/__init__.py

@@ -1297,6 +1297,7 @@ async def rank_eval(self, body, index=None, params=None, headers=None):
 
     @query_params(
         "max_docs",
+        "prefer_v2_templates",
         "refresh",
         "requests_per_second",
         "scroll",
@@ -1316,6 +1317,8 @@ async def reindex(self, body, params=None, headers=None):
             prototype for the index request.
         :arg max_docs: Maximum number of documents to process (default:
             all documents)
+        :arg prefer_v2_templates: favor V2 templates instead of V1
+            templates during index creation
         :arg refresh: Should the affected indexes be refreshed?
         :arg requests_per_second: The throttle to set on this request in
             sub-requests per second. -1 means no throttle.

elasticsearch/_async/client/cat.py

@@ -326,7 +326,7 @@ async def pending_tasks(self, params=None, headers=None):
             "GET", "/_cat/pending_tasks", params=params, headers=headers
         )
 
-    @query_params("format", "h", "help", "local", "master_timeout", "s", "size", "v")
+    @query_params("format", "h", "help", "local", "master_timeout", "s", "time", "v")
     async def thread_pool(self, thread_pool_patterns=None, params=None, headers=None):
         """
         Returns cluster-wide thread pool statistics per node. By default the active,
@@ -345,8 +345,8 @@ async def thread_pool(self, thread_pool_patterns=None, params=None, headers=None
             to master node
         :arg s: Comma-separated list of column names or column aliases
             to sort by
-        :arg size: The multiplier in which to display values  Valid
-            choices: , k, m, g, t, p
+        :arg time: The unit in which to display time values  Valid
+            choices: d, h, m, s, ms, micros, nanos
         :arg v: Verbose mode. Display column headers
         """
         return await self.transport.perform_request(

elasticsearch/_async/client/cluster.py

@@ -137,7 +137,7 @@ async def stats(self, node_id=None, params=None, headers=None):
             "GET",
             "/_cluster/stats"
             if node_id in SKIP_IN_PATH
-            else _make_path("_cluster/stats/nodes", node_id),
+            else _make_path("_cluster", "stats", "nodes", node_id),
             params=params,
             headers=headers,
         )

elasticsearch/_async/client/indices.py

@@ -1301,20 +1301,19 @@ async def get_index_template(self, name=None, params=None, headers=None):
             "GET", _make_path("_index_template", name), params=params, headers=headers
         )
 
-    @query_params("create", "master_timeout", "order")
+    @query_params("cause", "create", "master_timeout")
     async def put_index_template(self, name, body, params=None, headers=None):
         """
         Creates or updates an index template.
         `<https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-templates.html>`_
 
         :arg name: The name of the template
         :arg body: The template definition
+        :arg cause: User defined reason for creating/updating the index
+            template
         :arg create: Whether the index template should only be added if
             new or can also replace an existing one
         :arg master_timeout: Specify timeout for connection to master
-        :arg order: The order for this template when merging multiple
-            matching ones (higher numbers are merged later, overriding the lower
-            numbers)
         """
         for param in (name, body):
             if param in SKIP_IN_PATH:
@@ -1349,7 +1348,7 @@ async def exists_index_template(self, name, params=None, headers=None):
             "HEAD", _make_path("_index_template", name), params=params, headers=headers
         )
 
-    @query_params("master_timeout")
+    @query_params("cause", "create", "master_timeout")
     async def simulate_index_template(self, name, body=None, params=None, headers=None):
         """
         Simulate matching the given index name against the index templates in the
@@ -1360,6 +1359,11 @@ async def simulate_index_template(self, name, body=None, params=None, headers=No
             name)
         :arg body: New index template definition, which will be included
             in the simulation, as if it already exists in the system
+        :arg cause: User defined reason for dry-run creating the new
+            template for simulation purposes
+        :arg create: Whether the index template we optionally defined in
+            the body should only be dry-run added if new or can also replace an
+            existing one
         :arg master_timeout: Specify timeout for connection to master
         """
         if name in SKIP_IN_PATH:

elasticsearch/client/__init__.py

@@ -1297,6 +1297,7 @@ def rank_eval(self, body, index=None, params=None, headers=None):
 
     @query_params(
         "max_docs",
+        "prefer_v2_templates",
         "refresh",
         "requests_per_second",
         "scroll",
@@ -1316,6 +1317,8 @@ def reindex(self, body, params=None, headers=None):
             prototype for the index request.
         :arg max_docs: Maximum number of documents to process (default:
             all documents)
+        :arg prefer_v2_templates: favor V2 templates instead of V1
+            templates during index creation
         :arg refresh: Should the affected indexes be refreshed?
         :arg requests_per_second: The throttle to set on this request in
             sub-requests per second. -1 means no throttle.

elasticsearch/client/cat.py

@@ -326,7 +326,7 @@ def pending_tasks(self, params=None, headers=None):
             "GET", "/_cat/pending_tasks", params=params, headers=headers
         )
 
-    @query_params("format", "h", "help", "local", "master_timeout", "s", "size", "v")
+    @query_params("format", "h", "help", "local", "master_timeout", "s", "time", "v")
     def thread_pool(self, thread_pool_patterns=None, params=None, headers=None):
         """
         Returns cluster-wide thread pool statistics per node. By default the active,
@@ -345,8 +345,8 @@ def thread_pool(self, thread_pool_patterns=None, params=None, headers=None):
             to master node
         :arg s: Comma-separated list of column names or column aliases
             to sort by
-        :arg size: The multiplier in which to display values  Valid
-            choices: , k, m, g, t, p
+        :arg time: The unit in which to display time values  Valid
+            choices: d, h, m, s, ms, micros, nanos
         :arg v: Verbose mode. Display column headers
         """
         return self.transport.perform_request(

elasticsearch/client/indices.py

@@ -1299,20 +1299,19 @@ def get_index_template(self, name=None, params=None, headers=None):
             "GET", _make_path("_index_template", name), params=params, headers=headers
         )
 
-    @query_params("create", "master_timeout", "order")
+    @query_params("cause", "create", "master_timeout")
     def put_index_template(self, name, body, params=None, headers=None):
         """
         Creates or updates an index template.
         `<https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-templates.html>`_
 
         :arg name: The name of the template
         :arg body: The template definition
+        :arg cause: User defined reason for creating/updating the index
+            template
         :arg create: Whether the index template should only be added if
             new or can also replace an existing one
         :arg master_timeout: Specify timeout for connection to master
-        :arg order: The order for this template when merging multiple
-            matching ones (higher numbers are merged later, overriding the lower
-            numbers)
         """
         for param in (name, body):
             if param in SKIP_IN_PATH:
@@ -1347,7 +1346,7 @@ def exists_index_template(self, name, params=None, headers=None):
             "HEAD", _make_path("_index_template", name), params=params, headers=headers
         )
 
-    @query_params("master_timeout")
+    @query_params("cause", "create", "master_timeout")
     def simulate_index_template(self, name, body=None, params=None, headers=None):
         """
         Simulate matching the given index name against the index templates in the
@@ -1358,6 +1357,11 @@ def simulate_index_template(self, name, body=None, params=None, headers=None):
             name)
         :arg body: New index template definition, which will be included
             in the simulation, as if it already exists in the system
+        :arg cause: User defined reason for dry-run creating the new
+            template for simulation purposes
+        :arg create: Whether the index template we optionally defined in
+            the body should only be dry-run added if new or can also replace an
+            existing one
         :arg master_timeout: Specify timeout for connection to master
         """
         if name in SKIP_IN_PATH:

setup.py

@@ -62,6 +62,7 @@
     ],
     python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, <4",
     install_requires=install_requires,
+    test_suite="test_elasticsearch.run_tests.run_all",
     tests_require=tests_require,
     extras_require={
         "develop": tests_require + docs_require + generate_require,

test_elasticsearch/run_tests.py

@@ -78,9 +78,14 @@ def run_all(argv=None):
             "--log-level=DEBUG",
             "--cache-clear",
             "-vv",
-            abspath(dirname(__file__)),
         ]
 
+        # Skip all async tests unless Python 3.6+
+        if sys.version_info < (3, 6):
+            argv.append("--ignore=test_elasticsearch/test_async/")
+
+        argv.append(abspath(dirname(__file__)))
+
     exit_code = 0
     try:
         subprocess.check_call(argv, stdout=sys.stdout, stderr=sys.stderr)

test_elasticsearch/test_async/conftest.py

@@ -0,0 +1,10 @@
+# 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
+
+import sys
+import pytest
+
+pytestmark = pytest.mark.skipif(
+    sys.version_info < (3, 6), reason="'test_async' is only run on Python 3.6+"
+)

test_elasticsearch/test_async/test_connection.py

@@ -4,14 +4,25 @@
 # See the LICENSE file in the project root for more information
 
 import ssl
+import gzip
+import io
 from mock import Mock, patch
 import warnings
 from platform import python_version
+import aiohttp
+import pytest
 
 from elasticsearch import AIOHttpConnection
 from elasticsearch import __versionstr__
 from ..test_cases import TestCase, SkipTest
 
+pytestmark = pytest.mark.asyncio
+
+
+def gzip_decompress(data):
+    buf = gzip.GzipFile(fileobj=io.BytesIO(data), mode="rb")
+    return buf.read()
+
 
 class TestAIOHttpConnection(TestCase):
     async def _get_mock_connection(self, connection_params={}, response_body=b"{}"):
@@ -232,14 +243,15 @@ def test_uses_https_if_verify_certs_is_off(self):
         self.assertEqual(con.scheme, "https")
         self.assertEqual(con.host, "https://localhost:9200")
 
-    def nowarn_when_test_uses_https_if_verify_certs_is_off(self):
+    async def test_nowarn_when_test_uses_https_if_verify_certs_is_off(self):
         with warnings.catch_warnings(record=True) as w:
-            con = Urllib3HttpConnection(
+            con = AIOHttpConnection(
                 use_ssl=True, verify_certs=False, ssl_show_warn=False
             )
+            con._create_aiohttp_session()
             self.assertEqual(0, len(w))
 
-        self.assertIsInstance(con.pool, urllib3.HTTPSConnectionPool)
+        self.assertIsInstance(con.session, aiohttp.ClientSession)
 
     def test_doesnt_use_https_if_not_specified(self):
         con = AIOHttpConnection()

test_elasticsearch/test_async/test_connection_pool.py

@@ -3,6 +3,7 @@
 # See the LICENSE file in the project root for more information
 
 import time
+import pytest
 
 from elasticsearch import (
     AsyncConnectionPool,
@@ -15,6 +16,9 @@
 from ..test_cases import TestCase
 
 
+pytestmark = pytest.mark.asyncio
+
+
 class TestConnectionPool(TestCase):
     def test_dummy_cp_raises_exception_on_more_connections(self):
         self.assertRaises(ImproperlyConfigured, AsyncDummyConnectionPool, [])