Surface deprecation warnings from Elasticsearch

elasticsearch/__init__.py

@@ -6,6 +6,7 @@
 __versionstr__ = ".".join(map(str, VERSION))
 
 import logging
+import warnings
 
 logger = logging.getLogger("elasticsearch")
 logger.addHandler(logging.NullHandler())
@@ -28,8 +29,13 @@
     ConnectionTimeout,
     AuthenticationException,
     AuthorizationException,
+    ElasticsearchDeprecationWarning,
 )
 
+# Only raise one warning per deprecation message so as not
+# to spam up the user if the same action is done multiple times.
+warnings.simplefilter("default", category=ElasticsearchDeprecationWarning, append=True)
+
 __all__ = [
     "Elasticsearch",
     "Transport",
@@ -52,4 +58,5 @@
     "ConnectionTimeout",
     "AuthenticationException",
     "AuthorizationException",
+    "ElasticsearchDeprecationWarning",
 ]

elasticsearch/connection/base.py

@@ -2,14 +2,21 @@
 import binascii
 import gzip
 import io
+import re
 from platform import python_version
+import warnings
 
 try:
     import simplejson as json
 except ImportError:
     import json
 
-from ..exceptions import TransportError, ImproperlyConfigured, HTTP_EXCEPTIONS
+from ..exceptions import (
+    TransportError,
+    ImproperlyConfigured,
+    ElasticsearchDeprecationWarning,
+    HTTP_EXCEPTIONS,
+)
 from .. import __versionstr__
 
 logger = logging.getLogger("elasticsearch")
@@ -21,6 +28,8 @@
 if not _tracer_already_configured:
     tracer.propagate = False
 
+_WARNING_RE = re.compile(r"\"([^\"]*)\"")
+
 
 class Connection(object):
     """
@@ -135,6 +144,33 @@ def _gzip_compress(self, body):
             f.write(body)
         return buf.getvalue()
 
+    def _raise_warnings(self, warning_headers):
+        """If 'headers' contains a 'Warning' header raise
+        the warnings to be seen by the user. Takes an iterable
+        of string values from any number of 'Warning' headers.
+        """
+        if not warning_headers:
+            return
+
+        # Grab only the message from each header, the rest is discarded.
+        # Format is: '(number) Elasticsearch-(version)-(instance) "(message)"'
+        warning_messages = []
+        for header in warning_headers:
+            # Because 'Requests' does it's own folding of multiple HTTP headers
+            # into one header delimited by commas (totally standard compliant, just
+            # annoying for cases like this) we need to expect there may be
+            # more than one message per 'Warning' header.
+            matches = _WARNING_RE.findall(header)
+            if matches:
+                warning_messages.extend(matches)
+            else:
+                # Don't want to throw away any warnings, even if they
+                # don't follow the format we have now. Use the whole header.
+                warning_messages.append(header)
+
+        for message in warning_messages:
+            warnings.warn(message, category=ElasticsearchDeprecationWarning)
+
     def _pretty_json(self, data):
         # pretty JSON in tracer curl logs
         try:

elasticsearch/connection/http_requests.py

@@ -156,6 +156,12 @@ def perform_request(
                 raise ConnectionTimeout("TIMEOUT", str(e), e)
             raise ConnectionError("N/A", str(e), e)
 
+        # raise warnings if any from the 'Warnings' header.
+        warnings_headers = (
+            (response.headers["warning"],) if "warning" in response.headers else ()
+        )
+        self._raise_warnings(warnings_headers)
+
         # raise errors based on http status codes, let the client handle those if needed
         if (
             not (200 <= response.status_code < 300)

elasticsearch/connection/http_urllib3.py

@@ -240,6 +240,10 @@ def perform_request(
                 raise ConnectionTimeout("TIMEOUT", str(e), e)
             raise ConnectionError("N/A", str(e), e)
 
+        # raise warnings if any from the 'Warnings' header.
+        warning_headers = response.headers.get_all("warning", ())
+        self._raise_warnings(warning_headers)
+
         # raise errors based on http status codes, let the client handle those if needed
         if not (200 <= response.status < 300) and response.status not in ignore:
             self.log_request_fail(

elasticsearch/exceptions.py

@@ -136,6 +136,12 @@ class AuthorizationException(TransportError):
     """ Exception representing a 403 status code. """
 
 
+class ElasticsearchDeprecationWarning(Warning):
+    """ Warning that is raised when a deprecated option
+    is flagged via the 'Warning' HTTP header.
+    """
+
+
 # more generic mappings from status_code to python exceptions
 HTTP_EXCEPTIONS = {
     400: RequestError,

elasticsearch/helpers/test.py

@@ -50,8 +50,13 @@ def setUpClass(cls):
 
     def tearDown(self):
         super(ElasticsearchTestCase, self).tearDown()
+        # Hidden indices expanded in wildcards in ES 7.7
+        expand_wildcards = ["open", "closed"]
+        if self.es_version >= (7, 7):
+            expand_wildcards.append("hidden")
+
         self.client.indices.delete(
-            index="*", ignore=404, expand_wildcards="open,closed,hidden"
+            index="*", ignore=404, expand_wildcards=expand_wildcards
         )
         self.client.indices.delete_template(name="*", ignore=404)
 

test_elasticsearch/test_connection.py

@@ -4,6 +4,7 @@
 import io
 from mock import Mock, patch
 import urllib3
+from urllib3._collections import HTTPHeaderDict
 import warnings
 from requests.auth import AuthBase
 from platform import python_version
@@ -87,14 +88,56 @@ def test_parse_cloud_id(self):
             "8af7ee35420f458e903026b4064081f2.westeurope.azure.elastic-cloud.com",
         )
 
+    def test_empty_warnings(self):
+        con = Connection()
+        with warnings.catch_warnings(record=True) as w:
+            con._raise_warnings(())
+            con._raise_warnings([])
+
+        self.assertEquals(w, [])
+
+    def test_raises_warnings(self):
+        con = Connection()
+
+        with warnings.catch_warnings(record=True) as warn:
+            con._raise_warnings(['299 Elasticsearch-7.6.1-aa751 "this is deprecated"'])
+
+        self.assertEquals([str(w.message) for w in warn], ["this is deprecated"])
+
+        with warnings.catch_warnings(record=True) as warn:
+            con._raise_warnings(
+                [
+                    '299 Elasticsearch-7.6.1-aa751 "this is also deprecated"',
+                    '299 Elasticsearch-7.6.1-aa751 "this is also deprecated"',
+                    '299 Elasticsearch-7.6.1-aa751 "guess what? deprecated"',
+                ]
+            )
+
+        self.assertEquals(
+            [str(w.message) for w in warn],
+            ["this is also deprecated", "guess what? deprecated"],
+        )
+
+    def test_raises_warnings_when_folded(self):
+        con = Connection()
+        with warnings.catch_warnings(record=True) as warn:
+            con._raise_warnings(
+                [
+                    '299 Elasticsearch-7.6.1-aa751 "warning",'
+                    '299 Elasticsearch-7.6.1-aa751 "folded"',
+                ]
+            )
+
+        self.assertEquals([str(w.message) for w in warn], ["warning", "folded"])
+
 
 class TestUrllib3Connection(TestCase):
     def _get_mock_connection(self, connection_params={}, response_body=b"{}"):
         con = Urllib3HttpConnection(**connection_params)
 
         def _dummy_urlopen(*args, **kwargs):
             dummy_response = Mock()
-            dummy_response.headers = {}
+            dummy_response.headers = HTTPHeaderDict({})
             dummy_response.status = 200
             dummy_response.data = response_body
             _dummy_urlopen.call_args = (args, kwargs)

test_elasticsearch/test_server/test_common.py

@@ -8,8 +8,9 @@
 from os.path import exists, join, dirname, pardir
 import yaml
 from shutil import rmtree
+import warnings
 
-from elasticsearch import TransportError, RequestError
+from elasticsearch import TransportError, RequestError, ElasticsearchDeprecationWarning
 from elasticsearch.compat import string_types
 from elasticsearch.helpers.test import _get_version
 
@@ -30,6 +31,7 @@
     "headers",
     "catch_unauthorized",
     "default_shards",
+    "warnings",
 }
 
 # broken YAML tests on some releases
@@ -40,6 +42,8 @@
         # Disallowing expensive queries is 7.7+
         "TestSearch320DisallowQueries",
         "TestIndicesPutIndexTemplate10Basic",
+        "TestIndicesGetIndexTemplate10Basic",
+        "TestIndicesGetIndexTemplate20GetMissing",
     }
 }
 
@@ -142,6 +146,7 @@ def _lookup(self, path):
 
     def run_code(self, test):
         """ Execute an instruction based on it's type. """
+        print(test)
         for action in test:
             self.assertEquals(1, len(action))
             action_type, action = list(action.items())[0]
@@ -156,6 +161,7 @@ def run_do(self, action):
         api = self.client
         headers = action.pop("headers", None)
         catch = action.pop("catch", None)
+        warn = action.pop("warnings", ())
         self.assertEquals(1, len(action))
 
         method, args = list(action.items())[0]
@@ -176,17 +182,34 @@ def run_do(self, action):
         for k in args:
             args[k] = self._resolve(args[k])
 
-        try:
-            self.last_response = 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)
-                )
+        warnings.simplefilter("always", category=ElasticsearchDeprecationWarning)
+        with warnings.catch_warnings(record=True) as caught_warnings:
+            try:
+                self.last_response = 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
+        ]
+
+        # Sorting removes the issue with order raised. We only care about
+        # if all warnings are raised in the single API call.
+        if sorted(warn) != sorted(caught_warnings):
+            raise AssertionError(
+                "Expected warnings not equal to actual warnings: expected=%r actual=%r"
+                % (warn, caught_warnings)
+            )
 
     def _get_nodes(self):
         if not hasattr(self, "_node_info"):