Fix string/tuple/no auth on AsyncHttpConnection class

CHANGELOG.md

@@ -25,6 +25,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Fixed userguide for async client ([#340](https://github.com/opensearch-project/opensearch-py/pull/340))
 - Include parsed error info in TransportError in async connections (fixes #225) ([#226](https://github.com/opensearch-project/opensearch-py/pull/226))
 - Enhanced existing API generator to use OpenSearch OpenAPI spec ([#412](https://github.com/opensearch-project/opensearch-py/pull/412))
+- Fix crash when attempting to authenticate with an async connection (fixes #283)) ([#424](https://github.com/opensearch-project/opensearch-py/pull/424))
 ### Security
 - Fixed CVE-2022-23491 reported in opensearch-dsl-py ([#295](https://github.com/opensearch-project/opensearch-py/pull/295))
 - Update ci workflows ([#318](https://github.com/opensearch-project/opensearch-py/pull/318))

opensearchpy/connection/http_async.py

@@ -65,9 +65,10 @@ def __init__(
 
         if http_auth is not None:
             if isinstance(http_auth, (tuple, list)):
-                http_auth = ":".join(http_auth)
+                http_auth = aiohttp.BasicAuth(login=http_auth[0], password=http_auth[1])
             elif isinstance(http_auth, string_types):
-                http_auth = tuple(http_auth.split(":", 1))
+                login, password = http_auth.split(":", 1)
+                http_auth = aiohttp.BasicAuth(login=login, password=password)
 
         # if providing an SSL context, raise error if any other SSL related flag is used
         if ssl_context and (
@@ -190,17 +191,22 @@ async def perform_request(
             body = self._gzip_compress(body)
             req_headers["content-encoding"] = "gzip"
 
-        req_headers = {
-            **req_headers,
-            **self._http_auth(method, url, query_string, body),
-        }
+        auth = (
+            self._http_auth if isinstance(self._http_auth, aiohttp.BasicAuth) else None
+        )
+        if callable(self._http_auth):
+            req_headers = {
+                **req_headers,
+                **self._http_auth(method, url, query_string, body),
+            }
 
         start = self.loop.time()
         try:
             async with self.session.request(
                 method,
                 url,
                 data=body,
+                auth=auth,
                 headers=req_headers,
                 timeout=timeout,
                 fingerprint=self.ssl_assert_fingerprint,

test_opensearchpy/test_async/test_http_connection.py

@@ -0,0 +1,124 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+#
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+
+import mock
+import pytest
+from multidict import CIMultiDict
+
+from opensearchpy._async._extra_imports import aiohttp
+from opensearchpy._async.compat import get_running_loop
+from opensearchpy.connection.http_async import AsyncHttpConnection
+
+pytestmark = pytest.mark.asyncio
+
+
+class TestAsyncHttpConnection:
+    def test_auth_as_tuple(self):
+        c = AsyncHttpConnection(http_auth=("username", "password"))
+        assert isinstance(c._http_auth, aiohttp.BasicAuth)
+        assert c._http_auth.login, "username"
+        assert c._http_auth.password, "password"
+
+    def test_auth_as_string(self):
+        c = AsyncHttpConnection(http_auth="username:password")
+        assert isinstance(c._http_auth, aiohttp.BasicAuth)
+        assert c._http_auth.login, "username"
+        assert c._http_auth.password, "password"
+
+    def test_auth_as_callable(self):
+        def auth_fn():
+            pass
+
+        c = AsyncHttpConnection(http_auth=auth_fn)
+        assert callable(c._http_auth)
+
+    @mock.patch("aiohttp.ClientSession.request", new_callable=mock.Mock)
+    async def test_basicauth_in_request_session(self, mock_request):
+        async def do_request(*args, **kwargs):
+            response_mock = mock.AsyncMock()
+            response_mock.headers = CIMultiDict()
+            response_mock.status = 200
+            return response_mock
+
+        mock_request.return_value = aiohttp.client._RequestContextManager(do_request())
+
+        c = AsyncHttpConnection(
+            http_auth=("username", "password"),
+            loop=get_running_loop(),
+        )
+        c.headers = {}
+        await c.perform_request("post", "/test")
+        mock_request.assert_called_with(
+            "post",
+            "http://localhost:9200/test",
+            data=None,
+            auth=c._http_auth,
+            headers={},
+            timeout=aiohttp.ClientTimeout(
+                total=10,
+                connect=None,
+                sock_read=None,
+                sock_connect=None,
+            ),
+            fingerprint=None,
+        )
+
+    @mock.patch("aiohttp.ClientSession.request", new_callable=mock.Mock)
+    async def test_callable_in_request_session(self, mock_request):
+        def auth_fn(*args, **kwargs):
+            return {
+                "Test": "PASSED",
+            }
+
+        async def do_request(*args, **kwargs):
+            response_mock = mock.AsyncMock()
+            response_mock.headers = CIMultiDict()
+            response_mock.status = 200
+            return response_mock
+
+        mock_request.return_value = aiohttp.client._RequestContextManager(do_request())
+
+        c = AsyncHttpConnection(http_auth=auth_fn, loop=get_running_loop())
+        c.headers = {}
+        await c.perform_request("post", "/test")
+
+        mock_request.assert_called_with(
+            "post",
+            "http://localhost:9200/test",
+            data=None,
+            auth=None,
+            headers={
+                "Test": "PASSED",
+            },
+            timeout=aiohttp.ClientTimeout(
+                total=10,
+                connect=None,
+                sock_read=None,
+                sock_connect=None,
+            ),
+            fingerprint=None,
+        )