[7.x] Start using the Elastic Artifacts API for YAML REST tests

.ci/functions/imports.sh

@@ -18,7 +18,7 @@ require_stack_version
 if [[ -z $es_node_name ]]; then
   # only set these once
   set -euo pipefail
-  export TEST_SUITE=${TEST_SUITE-free}
+  export TEST_SUITE=${TEST_SUITE-platinum}
   export RUNSCRIPTS=${RUNSCRIPTS-}
   export DETACH=${DETACH-false}
   export CLEANUP=${CLEANUP-false}

.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.2.0
+# Version 1.3.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
@@ -15,6 +15,8 @@
 # - Refactored into functions and imports
 # - Support NUMBER_OF_NODES
 # - 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
 
 script_path=$(dirname $(realpath -s $0))
 source $script_path/functions/imports.sh
@@ -37,6 +39,7 @@ environment=($(cat <<-END
   --env node.attr.testattr=test
   --env path.repo=/tmp
   --env repositories.url.allowed_urls=http://snapshot.test*
+  --env action.destructive_requires_name=false
 END
 ))
 if [[ "$TEST_SUITE" == "platinum" ]]; then
@@ -45,11 +48,11 @@ 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
     --env xpack.security.transport.ssl.enabled=true
+    --env xpack.security.transport.ssl.verification_mode=certificate
     --env xpack.security.transport.ssl.key=certs/testnode.key
     --env xpack.security.transport.ssl.certificate=certs/testnode.crt
     --env xpack.security.transport.ssl.certificate_authorities=certs/ca.crt
@@ -104,7 +107,7 @@ END
   docker run \
     --name "$node_name" \
     --network "$network_name" \
-    --env "ES_JAVA_OPTS=-Xms1g -Xmx1g" \
+    --env "ES_JAVA_OPTS=-Xms1g -Xmx1g -da:org.elasticsearch.xpack.ccr.index.engine.FollowingEngineAssertions" \
     "${environment[@]}" \
     "${volumes[@]}" \
     --publish "$http_port":9200 \

.ci/run-repository.sh

@@ -34,6 +34,7 @@ echo -e "\033[1m>>>>> Run [elastic/elasticsearch-py container] >>>>>>>>>>>>>>>>>
 mkdir -p junit
 docker run \
   --network=${network_name} \
+  --env "STACK_VERSION=${STACK_VERSION}" \
   --env "ELASTICSEARCH_URL=${elasticsearch_url}" \
   --env "TEST_SUITE=${TEST_SUITE}" \
   --env "PYTHON_CONNECTION_CLASS=${PYTHON_CONNECTION_CLASS}" \

.ci/run-tests

@@ -6,7 +6,7 @@
 
 # Default environment variables
 export STACK_VERSION="${STACK_VERSION:=8.0.0-SNAPSHOT}"
-export TEST_SUITE="${TEST_SUITE:=free}"
+export TEST_SUITE="${TEST_SUITE:=platinum}"
 export PYTHON_VERSION="${PYTHON_VERSION:=3.9}"
 export PYTHON_CONNECTION_CLASS="${PYTHON_CONNECTION_CLASS:=Urllib3HttpConnection}"
 

.ci/test-matrix.yml

@@ -2,7 +2,6 @@ STACK_VERSION:
   - 7.x-SNAPSHOT
 
 TEST_SUITE:
-  - free
   - platinum
 
 PYTHON_VERSION:

CONTRIBUTING.md

@@ -23,9 +23,6 @@ There are several environment variables that control integration tests:
   the same as tags of `docker.elastic.co/elasticsearch/elasticsearch`
   such as `8.0.0-SNAPSHOT`, `7.x-SNAPSHOT`, etc. Defaults to the
   same `*-SNAPSHOT` version as the branch.
-- `TEST_SUITE`: Determines how to configure Elasticsearch either by running
-  without any non-free features or by beginning a Platinum license. Possible options
-  are `free` and `platinum`. Defaults to `free` as there are fewer test cases.
 
 **NOTE: You don't need to run the live integration tests for all changes. If
 you don't have Elasticsearch running locally the integration tests will be skipped.**

elasticsearch/_async/helpers.py

@@ -267,7 +267,9 @@ async def async_bulk(
 
     # make streaming_bulk yield successful results so we can count them
     kwargs["yield_ok"] = True
-    async for ok, item in async_streaming_bulk(client, actions, *args, **kwargs):
+    async for ok, item in async_streaming_bulk(
+        client, actions, ignore_status=ignore_status, *args, **kwargs
+    ):
         # go through request-response pairs and detect failures
         if not ok:
             if not stats_only:

elasticsearch/helpers/test.py

@@ -26,10 +26,8 @@ from elasticsearch.exceptions import ConnectionError
 
 if "ELASTICSEARCH_URL" in os.environ:
     ELASTICSEARCH_URL = os.environ["ELASTICSEARCH_URL"]
-elif os.environ.get("TEST_SUITE") == "platinum":
-    ELASTICSEARCH_URL = "https://elastic:changeme@localhost:9200"
 else:
-    ELASTICSEARCH_URL = "http://localhost:9200"
+    ELASTICSEARCH_URL = "https://elastic:changeme@localhost:9200"
 
 
 def get_test_client(nowait=False, **kwargs):

test_elasticsearch/test_async/test_server/test_rest_api_spec.py

@@ -33,7 +33,6 @@ from ...test_server.test_rest_api_spec import (
     PARAMS_RENAMES,
     RUN_ASYNC_REST_API_TESTS,
     YAML_TEST_SPECS,
-    InvalidActionType,
     YamlRunner,
 )
 
@@ -85,7 +84,7 @@ class AsyncYamlRunner(YamlRunner):
             if hasattr(self, "run_" + action_type):
                 await await_if_coro(getattr(self, "run_" + action_type)(action))
             else:
-                raise InvalidActionType(action_type)
+                raise RuntimeError("Invalid action type %r" % (action_type,))
 
     async def run_do(self, action):
         api = self.client
@@ -95,6 +94,16 @@ class AsyncYamlRunner(YamlRunner):
         allowed_warnings = action.pop("allowed_warnings", ())
         assert len(action) == 1
 
+        # Remove the x_pack_rest_user authentication
+        # if it's given via headers. We're already authenticated
+        # via the 'elastic' user.
+        if (
+            headers
+            and headers.get("Authorization", None)
+            == "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA=="
+        ):
+            headers.pop("Authorization")
+
         method, args = list(action.items())[0]
         args["headers"] = headers
 
@@ -135,6 +144,18 @@ class AsyncYamlRunner(YamlRunner):
             and str(w.message) not in allowed_warnings
         ]
 
+        # This warning can show up in many places but isn't accounted for
+        # in tests, so we remove it to make sure things pass.
+        include_type_name_warning = (
+            "[types removal] Using include_type_name in create index requests is deprecated. "
+            "The parameter will be removed in the next major version."
+        )
+        if (
+            include_type_name_warning in caught_warnings
+            and include_type_name_warning not in warn
+        ):
+            caught_warnings.remove(include_type_name_warning)
+
         # 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):

test_elasticsearch/test_server/conftest.py

@@ -25,11 +25,19 @@ from elasticsearch.helpers.test import ELASTICSEARCH_URL
 
 from ..utils import wipe_cluster
 
+# Information about the Elasticsearch instance running, if any
+# Used for
+ELASTICSEARCH_VERSION = ""
+ELASTICSEARCH_BUILD_HASH = ""
+ELASTICSEARCH_REST_API_TESTS = []
 
-@pytest.fixture(scope="function")
-def sync_client():
+
+@pytest.fixture(scope="session")
+def sync_client_factory():
     client = None
     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"}
         if "PYTHON_CONNECTION_CLASS" in os.environ:
             from elasticsearch import connection
@@ -40,7 +48,7 @@ def sync_client():
 
         client = elasticsearch.Elasticsearch(ELASTICSEARCH_URL, **kw)
 
-        # wait for yellow status
+        # Wait for the cluster to report a status of 'yellow'
         for _ in range(100):
             try:
                 client.cluster.health(wait_for_status="yellow")
@@ -48,12 +56,18 @@ def sync_client():
             except ConnectionError:
                 time.sleep(0.1)
         else:
-            # timeout
-            pytest.skip("Elasticsearch failed to start.")
+            pytest.skip("Elasticsearch wasn't running at %r" % (ELASTICSEARCH_URL,))
 
+        wipe_cluster(client)
         yield client
-
     finally:
         if client:
-            wipe_cluster(client)
             client.close()
+
+
+@pytest.fixture(scope="function")
+def sync_client(sync_client_factory):
+    try:
+        yield sync_client_factory
+    finally:
+        wipe_cluster(sync_client_factory)

test_elasticsearch/test_server/test_rest_api_spec.py

@@ -20,20 +20,24 @@ 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 io
+import json
 import os
 import re
 import sys
 import warnings
-from os import environ, walk
-from os.path import dirname, exists, join, pardir, relpath
+import zipfile
 
 import pytest
+import urllib3
 import yaml
 
 from elasticsearch import ElasticsearchWarning, RequestError, TransportError
 from elasticsearch.compat import string_types
 from elasticsearch.helpers.test import _get_version
 
+from . import get_client
+
 # some params had to be changed in python, keep track of them so we can rename
 # those in the tests accordingly
 PARAMS_RENAMES = {"type": "doc_type", "from": "from_"}
@@ -56,11 +60,32 @@ IMPLEMENTED_FEATURES = {
 
 # broken YAML tests on some releases
 SKIP_TESTS = {
-    "indices/get_alias/10_basic[23]",
-    "indices/simulate_index_template/10_basic[2]",
-    "search/aggregation/250_moving_fn[1]",
-    "search/aggregation/250_moving_fn[2]",
-    "search/highlight/20_fvh[3]",
+    "ml/post_data[1]",
+    "ml/post_data[2]",
+    "ml/post_data[3]",
+    "ml/post_data[4]",
+    "ml/post_data[5]",
+    "ml/post_data[6]",
+    "ml/get_trained_model_stats[1]",
+    "ml/get_trained_model_stats[2]",
+    "ml/get_trained_model_stats[3]",
+    "ml/set_upgrade_mode[1]",
+    "ml/set_upgrade_mode[2]",
+    "ml/set_upgrade_mode[3]",
+    "ml/jobs_get_stats[0]",
+    "ml/jobs_get_stats[1]",
+    "ml/jobs_get_stats[2]",
+    "ml/jobs_get_stats[3]",
+    "ml/jobs_get_stats[4]",
+    "ml/jobs_get_stats[5]",
+    "ml/jobs_get_stats[6]",
+    "ml/jobs_get_stats[7]",
+    "ml/jobs_get_stats[8]",
+    "ml/jobs_get_stats[9]",
+    "ml/jobs_get_stats[10]",
+    "service_accounts/10_basic[0]",
+    "service_accounts/10_basic[1]",
+    "snapshot/20_operator_privileges_disabled[0]",
 }
 
 
@@ -122,7 +147,7 @@ class YamlRunner:
             if hasattr(self, "run_" + action_type):
                 getattr(self, "run_" + action_type)(action)
             else:
-                raise InvalidActionType(action_type)
+                raise RuntimeError("Invalid action type %r" % (action_type,))
 
     def run_do(self, action):
         api = self.client
@@ -132,6 +157,16 @@ class YamlRunner:
         allowed_warnings = action.pop("allowed_warnings", ())
         assert len(action) == 1
 
+        # Remove the x_pack_rest_user authentication
+        # if it's given via headers. We're already authenticated
+        # via the 'elastic' user.
+        if (
+            headers
+            and headers.get("Authorization", None)
+            == "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA=="
+        ):
+            headers.pop("Authorization")
+
         method, args = list(action.items())[0]
         args["headers"] = headers
 
@@ -172,6 +207,18 @@ class YamlRunner:
             and str(w.message) not in allowed_warnings
         ]
 
+        # This warning can show up in many places but isn't accounted for
+        # in tests, so we remove it to make sure things pass.
+        include_type_name_warning = (
+            "[types removal] Using include_type_name in create index requests is deprecated. "
+            "The parameter will be removed in the next major version."
+        )
+        if (
+            include_type_name_warning in caught_warnings
+            and include_type_name_warning not in warn
+        ):
+            caught_warnings.remove(include_type_name_warning)
+
         # 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):
@@ -337,70 +384,102 @@ class YamlRunner:
         return name in XPACK_FEATURES
 
 
-class InvalidActionType(Exception):
-    pass
-
-
-YAML_DIR = environ.get(
-    "TEST_ES_YAML_DIR",
-    join(
-        dirname(__file__),
-        pardir,
-        pardir,
-        pardir,
-        "elasticsearch",
-        "rest-api-spec",
-        "src",
-        "main",
-        "resources",
-        "rest-api-spec",
-        "test",
-    ),
-)
+@pytest.fixture(scope="function")
+def sync_runner(sync_client):
+    return YamlRunner(sync_client)
 
 
 YAML_TEST_SPECS = []
 
-if exists(YAML_DIR):
-    # find all the test definitions in yaml files ...
-    for path, _, files in walk(YAML_DIR):
-        for filename in files:
-            if not filename.endswith((".yaml", ".yml")):
-                continue
+# Try loading the REST API test specs from the Elastic Artifacts API
+try:
+    # Construct the HTTP and Elasticsearch client
+    http = urllib3.PoolManager(retries=10)
+    client = get_client()
 
-            filepath = join(path, filename)
-            with open(filepath) as f:
-                tests = list(yaml.load_all(f, Loader=yaml.SafeLoader))
+    # Make a request to Elasticsearch for the build hash, we'll be looking for
+    # an artifact with this same hash to download test specs for.
+    build_hash = client.info()["version"]["build_hash"]
 
-            setup_code = None
-            teardown_code = None
-            run_codes = []
-            for i, test in enumerate(tests):
-                for test_name, definition in test.items():
-                    if test_name == "setup":
-                        setup_code = definition
-                    elif test_name == "teardown":
-                        teardown_code = definition
-                    else:
-                        run_codes.append((i, definition))
+    # Now talk to the artifacts API with the 'STACK_VERSION' environment variable
+    resp = http.request(
+        "GET",
+        "https://artifacts-api.elastic.co/v1/versions/%s"
+        % (os.environ["STACK_VERSION"],),
+    )
+    resp = json.loads(resp.data.decode("utf-8"))
 
-            for i, run_code in run_codes:
-                src = {"setup": setup_code, "run": run_code, "teardown": teardown_code}
-                # Pytest already replaces '.' and '_' with '/' so we do
-                # it ourselves so UI and 'SKIP_TESTS' match.
-                pytest_param_id = (
-                    "%s[%d]" % (relpath(filepath, YAML_DIR).rpartition(".")[0], i)
-                ).replace(".", "/")
+    # Look through every build and see if one matches the commit hash
+    # we're looking for. If not it's okay, we'll just use the latest and
+    # hope for the best!
+    builds = resp["version"]["builds"]
+    for build in builds:
+        if build["projects"]["elasticsearch"]["commit_hash"] == build_hash:
+            break
+    else:
+        build = builds[0]  # Use the latest
 
-                if pytest_param_id in SKIP_TESTS:
-                    src["skip"] = True
+    # Now we're looking for the 'rest-api-spec-<VERSION>-sources.jar' file
+    # to download and extract in-memory.
+    packages = build["projects"]["elasticsearch"]["packages"]
+    for package in packages:
+        if re.match(r"rest-resources-zip-.*\.zip", package):
+            package_url = packages[package]["url"]
+            break
+    else:
+        raise RuntimeError(
+            "Could not find the package 'rest-resources-zip-*.zip' in build %r" % build
+        )
 
-                YAML_TEST_SPECS.append(pytest.param(src, id=pytest_param_id))
+    # Download the zip and start reading YAML from the files in memory
+    package_zip = zipfile.ZipFile(io.BytesIO(http.request("GET", package_url).data))
+    for yaml_file in package_zip.namelist():
+        if not re.match(r"^rest-api-spec/test/.*\.ya?ml$", yaml_file):
+            continue
+        yaml_tests = list(yaml.safe_load_all(package_zip.read(yaml_file)))
 
+        # Each file may have a "test" named 'setup' or 'teardown',
+        # these sets of steps should be run at the beginning and end
+        # of every other test within the file so we do one pass to capture those.
+        setup_steps = teardown_steps = None
+        test_numbers_and_steps = []
+        test_number = 0
 
-@pytest.fixture(scope="function")
-def sync_runner(sync_client):
-    return YamlRunner(sync_client)
+        for yaml_test in yaml_tests:
+            test_name, test_step = yaml_test.popitem()
+            if test_name == "setup":
+                setup_steps = test_step
+            elif test_name == "teardown":
+                teardown_steps = test_step
+            else:
+                test_numbers_and_steps.append((test_number, yaml_test))
+                test_number += 1
+
+        # Now we combine setup, teardown, and test_steps into
+        # a set of pytest.param() instances
+        for test_number, test_step in test_numbers_and_steps:
+            # Build the id from the name of the YAML file and
+            # the number within that file. Most important step
+            # is to remove most of the file path prefixes and
+            # the .yml suffix.
+            pytest_test_name = yaml_file.rpartition(".")[0].replace(".", "/")
+            for prefix in ("rest-api-spec/", "test/", "free/", "platinum/"):
+                if pytest_test_name.startswith(prefix):
+                    pytest_test_name = pytest_test_name[len(prefix) :]
+            pytest_param_id = "%s[%d]" % (pytest_test_name, test_number)
+
+            pytest_param = {
+                "setup": setup_steps,
+                "run": test_step,
+                "teardown": teardown_steps,
+            }
+            if pytest_param_id in SKIP_TESTS:
+                pytest_param["skip"] = True
+
+            YAML_TEST_SPECS.append(pytest.param(pytest_param, id=pytest_param_id))
+
+except Exception as e:
+    warnings.warn("Could not load REST API tests: %s" % (str(e),))
 
 
 if not RUN_ASYNC_REST_API_TESTS:

test_elasticsearch/utils.py

@@ -15,7 +15,6 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
-import os
 import time
 
 from elasticsearch import Elasticsearch, NotFoundError, RequestError
@@ -35,7 +34,7 @@ def wipe_cluster(client):
     except ImportError:
         pass
 
-    is_xpack = os.getenv("TEST_SUITE") == "platinum"
+    is_xpack = True
     if is_xpack:
         wipe_rollup_jobs(client)
         wait_for_pending_tasks(client, filter="xpack/rollup/job")
@@ -105,7 +104,7 @@ def wipe_data_streams(client):
 
 def wipe_indices(client):
     client.indices.delete(
-        index="*",
+        index="*,-.ds-ilm-history-*",
         expand_wildcards="all",
         ignore=404,
     )
@@ -133,11 +132,9 @@ def wipe_xpack_templates(client):
     for _ in range(3):
         for template in list(templates_to_delete):
             try:
-
                 client.cluster.delete_component_template(
                     name=template["name"],
                 )
-
             except RequestError:
                 pass
             else:
@@ -211,6 +208,7 @@ def is_xpack_template(name):
         "security_audit_log",
         ".slm-history",
         ".async-search",
+        ".geoip_databases",
         "saml-service-provider",
         "ilm-history",
         "logs",

utils/generate-api.py

@@ -325,7 +325,7 @@ class API:
 @contextlib.contextmanager
 def download_artifact(version):
     # Download the list of all artifacts for a version
-    # and find the latest build URL for 'rest-api-spec-X.jar'
+    # and find the latest build URL for 'rest-resources-zip-*.zip'
     resp = http.request(
         "GET", f"https://artifacts-api.elastic.co/v1/versions/{version}"
     )
@@ -333,11 +333,13 @@ def download_artifact(version):
         "elasticsearch"
     ]["packages"]
     for package in packages:
-        if re.match(r"^rest-api-spec-.*-sources\.jar$", package):
+        if re.match(r"^rest-resources-zip-.*\.zip$", package):
             zip_url = packages[package]["url"]
             break
     else:
-        raise ValueError("Could not find 'rest-api-spec'")
+        raise RuntimeError(
+            "Could not find the package 'rest-resources-zip-*.zip' in build"
+        )
 
     # Download the .jar file and unzip only the API
     # .json files into a temporary directory