[7.x] Add support for 'url_prefix' to AIOHttpConnection

(cherry picked from commit 64e81434feddb37a271a31c4b9bc7576e2500008)

Co-authored-by: Seth Michael Larson <[email protected]>

elasticsearch/_async/http_aiohttp.py

@@ -81,6 +81,7 @@ class AIOHttpConnection(Connection):
 
         :arg host: hostname of the node (default: localhost)
         :arg port: port to use (integer, default: 9200)
+        :arg url_prefix: optional url prefix for elasticsearch
         :arg timeout: default timeout in seconds (float, default: 10)
         :arg http_auth: optional http auth information as either ':' separated
             string or a tuple
@@ -200,7 +201,7 @@ class AIOHttpConnection(Connection):
             await self._create_aiohttp_session()
 
         orig_body = body
-        url_path = url
+        url_path = self.url_prefix + url
         if params:
             query_string = urlencode(params)
         else:
@@ -219,7 +220,7 @@ class AIOHttpConnection(Connection):
             scheme=self.scheme,
             host=self.hostname,
             port=self.port,
-            path=url,
+            path=url_path,
             query_string=query_string,
             encoded=True,
         )

test_elasticsearch/test_async/test_connection.py

@@ -23,11 +23,11 @@ from mock import patch
 import warnings
 from platform import python_version
 import aiohttp
+from multidict import CIMultiDict
 import pytest
 
 from elasticsearch import AIOHttpConnection
 from elasticsearch import __versionstr__
-from ..test_cases import TestCase, SkipTest
 
 pytestmark = pytest.mark.asyncio
 
@@ -37,7 +37,7 @@ def gzip_decompress(data):
     return buf.read()
 
 
-class TestAIOHttpConnection(TestCase):
+class TestAIOHttpConnection:
     async def _get_mock_connection(self, connection_params={}, response_body=b"{}"):
         con = AIOHttpConnection(**connection_params)
         await con._create_aiohttp_session()
@@ -54,7 +54,7 @@ class TestAIOHttpConnection(TestCase):
                     return response_body.decode("utf-8", "surrogatepass")
 
             dummy_response = DummyResponse()
-            dummy_response.headers = {}
+            dummy_response.headers = CIMultiDict()
             dummy_response.status = 200
             _dummy_request.call_args = (args, kwargs)
             return dummy_response
@@ -69,45 +69,42 @@ class TestAIOHttpConnection(TestCase):
             # if create_default_context raises an AttributeError Exception
             # it means SSLContext is not available for that version of python
             # and we should skip this test.
-            raise SkipTest(
+            pytest.skip(
-                "Test test_ssl_context is skipped cause SSLContext is not available for this version of ptyhon"
+                "Test test_ssl_context is skipped cause SSLContext is not available for this version of Python"
             )
 
         con = AIOHttpConnection(use_ssl=True, ssl_context=context)
         await con._create_aiohttp_session()
-        self.assertTrue(con.use_ssl)
+        assert con.use_ssl
-        self.assertEqual(con.session.connector._ssl, context)
+        assert con.session.connector._ssl == context
 
     def test_opaque_id(self):
         con = AIOHttpConnection(opaque_id="app-1")
-        self.assertEqual(con.headers["x-opaque-id"], "app-1")
+        assert con.headers["x-opaque-id"] == "app-1"
 
     def test_http_cloud_id(self):
         con = AIOHttpConnection(
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng=="
         )
-        self.assertTrue(con.use_ssl)
+        assert con.use_ssl
-        self.assertEqual(
+        assert (
-            con.host, "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
+            con.host
+            == "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
         )
-        self.assertEqual(con.port, None)
+        assert con.port is None
-        self.assertEqual(
+        assert con.hostname == "4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
-            con.hostname, "4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
+        assert con.http_compress
-        )
-        self.assertTrue(con.http_compress)
 
         con = AIOHttpConnection(
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==",
             port=9243,
         )
-        self.assertEqual(
+        assert (
-            con.host,
+            con.host
-            "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io:9243",
+            == "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io:9243"
-        )
-        self.assertEqual(con.port, 9243)
-        self.assertEqual(
-            con.hostname, "4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
         )
+        assert con.port == 9243
+        assert con.hostname == "4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
 
     def test_api_key_auth(self):
         # test with tuple
@@ -115,11 +112,10 @@ class TestAIOHttpConnection(TestCase):
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==",
             api_key=("elastic", "changeme1"),
         )
-        self.assertEqual(
+        assert con.headers["authorization"] == "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTE="
-            con.headers["authorization"], "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTE="
+        assert (
-        )
+            con.host
-        self.assertEqual(
+            == "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
-            con.host, "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
         )
 
         # test with base64 encoded string
@@ -127,50 +123,49 @@ class TestAIOHttpConnection(TestCase):
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==",
             api_key="ZWxhc3RpYzpjaGFuZ2VtZTI=",
         )
-        self.assertEqual(
+        assert con.headers["authorization"] == "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTI="
-            con.headers["authorization"], "ApiKey ZWxhc3RpYzpjaGFuZ2VtZTI="
+        assert (
-        )
+            con.host
-        self.assertEqual(
+            == "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
-            con.host, "https://4fa8821e75634032bed1cf22110e2f97.us-east-1.aws.found.io"
         )
 
     async def test_no_http_compression(self):
         con = await self._get_mock_connection()
-        self.assertFalse(con.http_compress)
+        assert not con.http_compress
-        self.assertNotIn("accept-encoding", con.headers)
+        assert "accept-encoding" not in con.headers
 
         await con.perform_request("GET", "/")
 
         _, kwargs = con.session.request.call_args
 
-        self.assertFalse(kwargs["data"])
+        assert not kwargs["data"]
-        self.assertNotIn("accept-encoding", kwargs["headers"])
+        assert "accept-encoding" not in kwargs["headers"]
-        self.assertNotIn("content-encoding", kwargs["headers"])
+        assert "content-encoding" not in kwargs["headers"]
 
     async def test_http_compression(self):
         con = await self._get_mock_connection({"http_compress": True})
-        self.assertTrue(con.http_compress)
+        assert con.http_compress
-        self.assertEqual(con.headers["accept-encoding"], "gzip,deflate")
+        assert con.headers["accept-encoding"] == "gzip,deflate"
 
         # 'content-encoding' shouldn't be set at a connection level.
         # Should be applied only if the request is sent with a body.
-        self.assertNotIn("content-encoding", con.headers)
+        assert "content-encoding" not in con.headers
 
         await con.perform_request("GET", "/", body=b"{}")
 
         _, kwargs = con.session.request.call_args
 
-        self.assertEqual(gzip_decompress(kwargs["data"]), b"{}")
+        assert gzip_decompress(kwargs["data"]) == b"{}"
-        self.assertEqual(kwargs["headers"]["accept-encoding"], "gzip,deflate")
+        assert kwargs["headers"]["accept-encoding"] == "gzip,deflate"
-        self.assertEqual(kwargs["headers"]["content-encoding"], "gzip")
+        assert kwargs["headers"]["content-encoding"] == "gzip"
 
         await con.perform_request("GET", "/")
 
         _, kwargs = con.session.request.call_args
 
-        self.assertFalse(kwargs["data"])
+        assert not kwargs["data"]
-        self.assertEqual(kwargs["headers"]["accept-encoding"], "gzip,deflate")
+        assert kwargs["headers"]["accept-encoding"] == "gzip,deflate"
-        self.assertNotIn("content-encoding", kwargs["headers"])
+        assert "content-encoding" not in kwargs["headers"]
 
     def test_cloud_id_http_compress_override(self):
         # 'http_compress' will be 'True' by default for connections with
@@ -178,90 +173,90 @@ class TestAIOHttpConnection(TestCase):
         con = AIOHttpConnection(
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==",
         )
-        self.assertEqual(con.http_compress, True)
+        assert con.http_compress is True
 
         con = AIOHttpConnection(
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==",
             http_compress=False,
         )
-        self.assertEqual(con.http_compress, False)
+        assert con.http_compress is False
 
         con = AIOHttpConnection(
             cloud_id="cluster:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5NyQ0ZmE4ODIxZTc1NjM0MDMyYmVkMWNmMjIxMTBlMmY5Ng==",
             http_compress=True,
         )
-        self.assertEqual(con.http_compress, True)
+        assert con.http_compress is True
+
+    async def test_url_prefix(self):
+        con = await self._get_mock_connection(
+            connection_params={"url_prefix": "/_search/"}
+        )
+        assert con.url_prefix == "/_search"
+
+        await con.perform_request("GET", "/")
+
+        # Need to convert the yarl URL to a string to compare.
+        method, yarl_url = con.session.request.call_args[0]
+        assert method == "GET" and str(yarl_url) == "http://localhost:9200/_search/"
 
     def test_default_user_agent(self):
         con = AIOHttpConnection()
-        self.assertEqual(
+        assert con._get_default_user_agent() == "elasticsearch-py/%s (Python %s)" % (
-            con._get_default_user_agent(),
+            __versionstr__,
-            "elasticsearch-py/%s (Python %s)" % (__versionstr__, python_version()),
+            python_version(),
         )
 
     def test_timeout_set(self):
         con = AIOHttpConnection(timeout=42)
-        self.assertEqual(42, con.timeout)
+        assert 42 == con.timeout
 
     def test_keep_alive_is_on_by_default(self):
         con = AIOHttpConnection()
-        self.assertEqual(
+        assert {
-            {
             "connection": "keep-alive",
             "content-type": "application/json",
             "user-agent": con._get_default_user_agent(),
-            },
+        } == con.headers
-            con.headers,
-        )
 
     def test_http_auth(self):
         con = AIOHttpConnection(http_auth="username:secret")
-        self.assertEqual(
+        assert {
-            {
             "authorization": "Basic dXNlcm5hbWU6c2VjcmV0",
             "connection": "keep-alive",
             "content-type": "application/json",
             "user-agent": con._get_default_user_agent(),
-            },
+        } == con.headers
-            con.headers,
-        )
 
     def test_http_auth_tuple(self):
         con = AIOHttpConnection(http_auth=("username", "secret"))
-        self.assertEqual(
+        assert {
-            {
             "authorization": "Basic dXNlcm5hbWU6c2VjcmV0",
             "content-type": "application/json",
             "connection": "keep-alive",
             "user-agent": con._get_default_user_agent(),
-            },
+        } == con.headers
-            con.headers,
-        )
 
     def test_http_auth_list(self):
         con = AIOHttpConnection(http_auth=["username", "secret"])
-        self.assertEqual(
+        assert {
-            {
             "authorization": "Basic dXNlcm5hbWU6c2VjcmV0",
             "content-type": "application/json",
             "connection": "keep-alive",
             "user-agent": con._get_default_user_agent(),
-            },
+        } == con.headers
-            con.headers,
-        )
 
     def test_uses_https_if_verify_certs_is_off(self):
         with warnings.catch_warnings(record=True) as w:
             con = AIOHttpConnection(use_ssl=True, verify_certs=False)
-            self.assertEqual(1, len(w))
+            assert 1 == len(w)
-            self.assertEqual(
+            assert (
-                "Connecting to https://localhost:9200 using SSL with verify_certs=False is insecure.",
+                "Connecting to https://localhost:9200 using SSL with verify_certs=False is insecure."
-                str(w[0].message),
+                == str(w[0].message)
             )
 
-        self.assertTrue(con.use_ssl)
+        assert con.use_ssl
-        self.assertEqual(con.scheme, "https")
+        assert con.scheme == "https"
-        self.assertEqual(con.host, "https://localhost:9200")
+        assert con.host == "https://localhost:9200"
 
     async def test_nowarn_when_test_uses_https_if_verify_certs_is_off(self):
         with warnings.catch_warnings(record=True) as w:
@@ -269,19 +264,19 @@ class TestAIOHttpConnection(TestCase):
                 use_ssl=True, verify_certs=False, ssl_show_warn=False
             )
             await con._create_aiohttp_session()
-            self.assertEqual(0, len(w))
+            assert 0 == len(w)
 
-        self.assertIsInstance(con.session, aiohttp.ClientSession)
+        assert isinstance(con.session, aiohttp.ClientSession)
 
     def test_doesnt_use_https_if_not_specified(self):
         con = AIOHttpConnection()
-        self.assertFalse(con.use_ssl)
+        assert not con.use_ssl
 
     def test_no_warning_when_using_ssl_context(self):
         ctx = ssl.create_default_context()
         with warnings.catch_warnings(record=True) as w:
             AIOHttpConnection(ssl_context=ctx)
-            self.assertEqual(0, len(w), str([x.message for x in w]))
+            assert 0 == len(w), str([x.message for x in w])
 
     def test_warns_if_using_non_default_ssl_kwargs_with_ssl_context(self):
         for kwargs in (
@@ -299,10 +294,10 @@ class TestAIOHttpConnection(TestCase):
 
                 AIOHttpConnection(**kwargs)
 
-                self.assertEqual(1, len(w))
+                assert 1 == len(w)
-                self.assertEqual(
+                assert (
-                    "When using `ssl_context`, all other SSL related kwargs are ignored",
+                    "When using `ssl_context`, all other SSL related kwargs are ignored"
-                    str(w[0].message),
+                    == str(w[0].message)
                 )
 
     @patch("elasticsearch.connection.base.logger")
@@ -310,14 +305,14 @@ class TestAIOHttpConnection(TestCase):
         con = await self._get_mock_connection(connection_params={"http_compress": True})
         await con.perform_request("GET", "/", body=b'{"example": "body"}')
 
-        self.assertEqual(2, logger.debug.call_count)
+        assert 2 == logger.debug.call_count
         req, resp = logger.debug.call_args_list
 
-        self.assertEqual('> {"example": "body"}', req[0][0] % req[0][1:])
+        assert '> {"example": "body"}' == req[0][0] % req[0][1:]
-        self.assertEqual("< {}", resp[0][0] % resp[0][1:])
+        assert "< {}" == resp[0][0] % resp[0][1:]
 
     async def test_surrogatepass_into_bytes(self):
         buf = b"\xe4\xbd\xa0\xe5\xa5\xbd\xed\xa9\xaa"
         con = await self._get_mock_connection(response_body=buf)
         status, headers, data = await con.perform_request("GET", "/")
-        self.assertEqual(u"你好\uda6a", data)
+        assert u"你好\uda6a" == data

test_elasticsearch/test_async/test_server/test_helpers.py

@@ -387,11 +387,14 @@ class MockResponse:
         return self.resp
 
 
-class TestScan(object):
+@pytest.fixture(scope="function")
-    async def teardown_method(self, async_client):
+async def scan_teardown(async_client):
-        await async_client.transport.perform_request("DELETE", "/_search/scroll/_all")
+    yield
+    async_client.clear_scroll(scroll_id="_all")
 
-    async def test_order_can_be_preserved(self, async_client):
+
+class TestScan(object):
+    async def test_order_can_be_preserved(self, async_client, scan_teardown):
         bulk = []
         for x in range(100):
             bulk.append({"index": {"_index": "test_index", "_id": x}})
@@ -412,7 +415,7 @@ class TestScan(object):
         assert list(map(str, range(100))) == list(d["_id"] for d in docs)
         assert list(range(100)) == list(d["_source"]["answer"] for d in docs)
 
-    async def test_all_documents_are_read(self, async_client):
+    async def test_all_documents_are_read(self, async_client, scan_teardown):
         bulk = []
         for x in range(100):
             bulk.append({"index": {"_index": "test_index", "_id": x}})
@@ -428,7 +431,7 @@ class TestScan(object):
         assert set(map(str, range(100))) == set(d["_id"] for d in docs)
         assert set(range(100)) == set(d["_source"]["answer"] for d in docs)
 
-    async def test_scroll_error(self, async_client):
+    async def test_scroll_error(self, async_client, scan_teardown):
         bulk = []
         for x in range(4):
             bulk.append({"index": {"_index": "test_index"}})
@@ -464,7 +467,7 @@ class TestScan(object):
             assert len(data) == 3
             assert data[-1] == {"scroll_data": 42}
 
-    async def test_initial_search_error(self, async_client):
+    async def test_initial_search_error(self, async_client, scan_teardown):
         with patch.object(async_client, "clear_scroll", new_callable=AsyncMock):
             with patch.object(
                 async_client,
@@ -516,7 +519,7 @@ class TestScan(object):
                         assert data == [{"search_data": 1}]
                         assert mock_scroll.calls == []
 
-    async def test_no_scroll_id_fast_route(self, async_client):
+    async def test_no_scroll_id_fast_route(self, async_client, scan_teardown):
         with patch.object(async_client, "search", MockResponse({"no": "_scroll_id"})):
             with patch.object(async_client, "scroll") as scroll_mock:
                 with patch.object(async_client, "clear_scroll") as clear_mock:
@@ -532,7 +535,7 @@ class TestScan(object):
                     clear_mock.assert_not_called()
 
     @patch("elasticsearch._async.helpers.logger")
-    async def test_logger(self, logger_mock, async_client):
+    async def test_logger(self, logger_mock, async_client, scan_teardown):
         bulk = []
         for x in range(4):
             bulk.append({"index": {"_index": "test_index"}})
@@ -573,7 +576,7 @@ class TestScan(object):
                 5,
             )
 
-    async def test_clear_scroll(self, async_client):
+    async def test_clear_scroll(self, async_client, scan_teardown):
         bulk = []
         for x in range(4):
             bulk.append({"index": {"_index": "test_index"}})

test_elasticsearch/test_connection.py

@@ -363,7 +363,7 @@ class TestUrllib3Connection(TestCase):
 
         self.assertIsInstance(con.pool, urllib3.HTTPSConnectionPool)
 
-    def nowarn_when_test_uses_https_if_verify_certs_is_off(self):
+    def test_nowarn_when_uses_https_if_verify_certs_is_off(self):
         with warnings.catch_warnings(record=True) as w:
             con = Urllib3HttpConnection(
                 use_ssl=True, verify_certs=False, ssl_show_warn=False
@@ -591,7 +591,7 @@ class TestRequestsConnection(TestCase):
         self.assertEqual("GET", request.method)
         self.assertEqual(None, request.body)
 
-    def nowarn_when_test_uses_https_if_verify_certs_is_off(self):
+    def test_nowarn_when_uses_https_if_verify_certs_is_off(self):
         with warnings.catch_warnings(record=True) as w:
             con = self._get_mock_connection(
                 {