Fix string/tuple/no auth on AsyncHttpConnection class

CHANGELOG.md

@@ -22,7 +22,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Fixed
 - Fixed import cycle when importing async helpers ([#311](https://github.com/opensearch-project/opensearch-py/pull/311))
 - 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)
+- Include parsed error info in TransportError in async connections (fixes #225) ([#226](https://github.com/opensearch-project/opensearch-py/pull/226))
+- 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))

dev-requirements.txt

@@ -1,6 +1,7 @@
 requests>=2, <3
 pytest
 pytest-cov
+asynctest
 coverage
 mock
 sphinx<7.1

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,20 @@ 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_http_connection.py

@@ -0,0 +1,78 @@
+from unittest import IsolatedAsyncioTestCase
+
+import pytest
+from asynctest import MagicMock, patch, CoroutineMock
+
+from opensearchpy import AsyncHttpConnection
+from opensearchpy._async._extra_imports import aiohttp
+from opensearchpy._async.compat import get_running_loop
+
+
+class AsyncContextManagerMock(MagicMock):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        type(self).__aenter__ = CoroutineMock(
+            return_value=MagicMock(
+                text=CoroutineMock(return_value='test'),
+                status=200,
+            )
+        )
+        type(self).__aexit__ = CoroutineMock(return_value=MagicMock())
+
+
+@pytest.mark.asyncio
+class TestAsyncHttpConnection(IsolatedAsyncioTestCase):
+    def test_auth_as_tuple(self):
+        c = AsyncHttpConnection(http_auth=('username', 'password'))
+        self.assertIsInstance(c._http_auth, aiohttp.BasicAuth)
+        self.assertEqual(c._http_auth.login, 'username')
+        self.assertEqual(c._http_auth.password, 'password')
+
+    def test_auth_as_string(self):
+        c = AsyncHttpConnection(http_auth='username:password')
+        self.assertIsInstance(c._http_auth, aiohttp.BasicAuth)
+        self.assertEqual(c._http_auth.login, 'username')
+        self.assertEqual(c._http_auth.password, 'password')
+
+    def test_auth_as_callable(self):
+        def auth_fn():
+            pass
+
+        c = AsyncHttpConnection(http_auth=auth_fn)
+        self.assertTrue(callable(c._http_auth))
+
+    @patch('aiohttp.ClientSession.request', new_callable=AsyncContextManagerMock)
+    async def test_basicauth_in_request_session(self, mock_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,
+        )
+
+    @patch('aiohttp.ClientSession.request', new_callable=AsyncContextManagerMock)
+    async def test_callable_in_request_session(self, mock_request):
+        def auth_fn(*args, **kwargs):
+            return {
+                'Test': 'PASSED'
+            }
+
+        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,
+        )