Allow integration tests to be run from IDE

.ci/run-elasticsearch.sh

@@ -7,7 +7,7 @@
 # Export the TEST_SUITE variable, eg. 'free' or 'platinum' defaults to 'free'.
 # Export the NUMBER_OF_NODES variable to start more than 1 node
 
-# Version 1.3.0
+# Version 1.4.0
 # - Initial version of the run-elasticsearch.sh script
 # - Deleting the volume should not dependent on the container still running
 # - Fixed `ES_JAVA_OPTS` config
@@ -17,6 +17,7 @@
 # - Added 5 retries on docker pull for fixing transient network errors
 # - Added flags to make local CCR configurations work
 # - Added action.destructive_requires_name=false as the default will be true in v8
+# - Added ingest.geoip.downloader.enabled=false as it causes false positives in testing
 
 script_path=$(dirname $(realpath -s $0))
 source $script_path/functions/imports.sh
@@ -40,6 +41,7 @@ environment=($(cat <<-END
   --env path.repo=/tmp
   --env repositories.url.allowed_urls=http://snapshot.test*
   --env action.destructive_requires_name=false
+  --env ingest.geoip.downloader.enabled=false
 END
 ))
 if [[ "$TEST_SUITE" == "platinum" ]]; then
@@ -48,6 +50,7 @@ if [[ "$TEST_SUITE" == "platinum" ]]; then
     --env xpack.license.self_generated.type=trial
     --env xpack.security.enabled=true
     --env xpack.security.http.ssl.enabled=true
+    --env xpack.security.http.ssl.verification_mode=certificate
     --env xpack.security.http.ssl.key=certs/testnode.key
     --env xpack.security.http.ssl.certificate=certs/testnode.crt
     --env xpack.security.http.ssl.certificate_authorities=certs/ca.crt

elasticsearch/_async/helpers.py

@@ -190,7 +190,7 @@ async def map_actions():
                         raise_on_error,
                         ignore_status,
                         *args,
-                        **kwargs
+                        **kwargs,
                     ),
                 ):
 
@@ -458,5 +458,5 @@ async def _change_doc_index(hits, index):
         target_client,
         _change_doc_index(docs, target_index),
         chunk_size=chunk_size,
-        **kwargs
+        **kwargs,
     )

elasticsearch/helpers/test.py

@@ -19,6 +19,7 @@
 
 import os
 import time
+from os.path import abspath, dirname, join
 from unittest import SkipTest, TestCase
 
 from elasticsearch import Elasticsearch
@@ -29,10 +30,12 @@
 else:
     ELASTICSEARCH_URL = "https://elastic:changeme@localhost:9200"
 
+CA_CERTS = join(dirname(dirname(dirname(abspath(__file__)))), ".ci/certs/ca.pem")
+
 
 def get_test_client(nowait=False, **kwargs):
     # construct kwargs from the environment
-    kw = {"timeout": 30, "ca_certs": ".ci/certs/ca.pem"}
+    kw = {"timeout": 30, "ca_certs": CA_CERTS}
     if "PYTHON_CONNECTION_CLASS" in os.environ:
         from elasticsearch import connection
 
@@ -55,13 +58,6 @@ def get_test_client(nowait=False, **kwargs):
         raise SkipTest("Elasticsearch failed to start.")
 
 
-def _get_version(version_string):
-    if "." not in version_string:
-        return ()
-    version = version_string.strip().split(".")
-    return tuple(int(v) if v.isdigit() else 999 for v in version)
-
-
 class ElasticsearchTestCase(TestCase):
     @staticmethod
     def _get_client():
@@ -85,6 +81,16 @@ def teardown_method(self, _):
 
     def es_version(self):
         if not hasattr(self, "_es_version"):
-            version_string = self.client.info()["version"]["number"]
-            self._es_version = _get_version(version_string)
+            self._es_version = es_version(self.client)
         return self._es_version
+
+
+def _get_version(version_string):
+    if "." not in version_string:
+        return ()
+    version = version_string.strip().split(".")
+    return tuple(int(v) if v.isdigit() else 999 for v in version)
+
+
+def es_version(client):
+    return _get_version(client.info()["version"]["number"])

elasticsearch/helpers/test.pyi

@@ -21,6 +21,7 @@ from unittest import TestCase
 from ..client import Elasticsearch
 
 ELASTICSEARCH_URL: str
+CA_CERTS: str
 
 def get_test_client(nowait: bool = ..., **kwargs: Any) -> Elasticsearch: ...
 def _get_version(version_string: str) -> Tuple[int, ...]: ...

test_elasticsearch/test_async/test_server/conftest.py

@@ -20,7 +20,7 @@
 import pytest
 
 import elasticsearch
-from elasticsearch.helpers.test import ELASTICSEARCH_URL
+from elasticsearch.helpers.test import CA_CERTS, ELASTICSEARCH_URL
 
 from ...utils import wipe_cluster
 
@@ -34,7 +34,7 @@ async def async_client():
         if not hasattr(elasticsearch, "AsyncElasticsearch"):
             pytest.skip("test requires 'AsyncElasticsearch'")
 
-        kw = {"timeout": 3, "ca_certs": ".ci/certs/ca.pem"}
+        kw = {"timeout": 3, "ca_certs": CA_CERTS}
         client = elasticsearch.AsyncElasticsearch(ELASTICSEARCH_URL, **kw)
 
         # wait for yellow status

test_elasticsearch/test_async/test_server/test_rest_api_spec.py

@@ -50,11 +50,26 @@ async def await_if_coro(x):
 
 class AsyncYamlRunner(YamlRunner):
     async def setup(self):
+        # Pull skips from individual tests to not do unnecessary setup.
+        skip_code = []
+        for action in self._run_code:
+            assert len(action) == 1
+            action_type, _ = list(action.items())[0]
+            if action_type == "skip":
+                skip_code.append(action)
+            else:
+                break
+
+        if self._setup_code or skip_code:
+            self.section("setup")
+        if skip_code:
+            await self.run_code(skip_code)
         if self._setup_code:
             await self.run_code(self._setup_code)
 
     async def teardown(self):
         if self._teardown_code:
+            self.section("teardown")
             await self.run_code(self._teardown_code)
 
     async def es_version(self):
@@ -67,19 +82,26 @@ async def es_version(self):
             ES_VERSION = tuple(int(v) if v.isdigit() else 999 for v in version)
         return ES_VERSION
 
+    def section(self, name):
+        print(("=" * 10) + " " + name + " " + ("=" * 10))
+
     async def run(self):
         try:
             await self.setup()
+            self.section("test")
             await self.run_code(self._run_code)
         finally:
-            await self.teardown()
+            try:
+                await self.teardown()
+            except Exception:
+                pass
 
     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]
+            print(action_type, action)
 
             if hasattr(self, "run_" + action_type):
                 await await_if_coro(getattr(self, "run_" + action_type)(action))

test_elasticsearch/test_server/conftest.py

@@ -21,7 +21,7 @@
 import pytest
 
 import elasticsearch
-from elasticsearch.helpers.test import ELASTICSEARCH_URL
+from elasticsearch.helpers.test import CA_CERTS, ELASTICSEARCH_URL
 
 from ..utils import wipe_cluster
 
@@ -38,15 +38,23 @@ def sync_client_factory():
     try:
         # Configure the client with certificates and optionally
         # an HTTP conn class depending on 'PYTHON_CONNECTION_CLASS' envvar
-        kw = {"timeout": 3, "ca_certs": ".ci/certs/ca.pem"}
+        kw = {
+            "timeout": 3,
+            "ca_certs": CA_CERTS,
+            "headers": {"Authorization": "Basic ZWxhc3RpYzpjaGFuZ2VtZQ=="},
+        }
         if "PYTHON_CONNECTION_CLASS" in os.environ:
             from elasticsearch import connection
 
             kw["connection_class"] = getattr(
                 connection, os.environ["PYTHON_CONNECTION_CLASS"]
             )
 
-        client = elasticsearch.Elasticsearch(ELASTICSEARCH_URL, **kw)
+        # We do this little dance with the URL to force
+        # Requests to respect 'headers: None' within rest API spec tests.
+        client = elasticsearch.Elasticsearch(
+            ELASTICSEARCH_URL.replace("elastic:changeme@", ""), **kw
+        )
 
         # Wait for the cluster to report a status of 'yellow'
         for _ in range(100):