Add endpoints and authentication details (including SigV4) as parameters to E2E test script

.github/workflows/e2eTest.yml

@@ -44,4 +44,4 @@ jobs:
         run: |
           cd test
           chmod +x ./tests.py
-          pytest tests.py --proxy_endpoint="https://localhost:9200" --source_endpoint="https://localhost:19200" --target_endpoint="https://localhost:29200" --source_auth_type="basic" --source_username="admin" --source_password="admin" --target_auth_type="basic" --target_username="admin" --target_password="admin" --source_verify_ssl=False --target_verify_ssl=False
+          pytest tests.py

test/conftest.py

@@ -1,6 +1,18 @@
 # conftest.py
 import pytest
 
+import logging
+
+
+def pytest_configure(config):
+    # Configure logging
+    logging.basicConfig(level=logging.DEBUG,
+                        format='%(asctime)s - %(levelname)s - %(message)s',
+                        datefmt='%Y-%m-%d %H:%M:%S')
+
+    # This line ensures that log messages are displayed on the console during test runs
+    logging.getLogger().setLevel(logging.DEBUG)
+
 
 def pytest_addoption(parser):
     parser.addoption("--proxy_endpoint", action="store", default="https://localhost:9200")

test/operations.py

@@ -1,52 +1,49 @@
 import requests
 import json
-from typing import Optional, Tuple
 
 
-def create_index(endpoint: str, index_name: str, auth: Optional[Tuple[str, str]] = None, verify_ssl: bool = False):
+def create_index(endpoint: str, index_name: str, auth, verify_ssl: bool = False, protocol: str = 'https'):
     response = requests.put(f'{endpoint}/{index_name}', auth=auth, verify=verify_ssl)
 
     return response
 
 
-def check_index(endpoint: str, index_name: str, auth: Optional[Tuple[str, str]] = None, verify_ssl: bool = False):
+def check_index(endpoint: str, index_name: str, auth, verify_ssl: bool = False, protocol: str = 'https'):
     response = requests.get(f'{endpoint}/{index_name}', auth=auth, verify=verify_ssl)
 
     return response
 
 
-def delete_index(endpoint: str, index_name: str, auth: Optional[Tuple[str, str]] = None, verify_ssl: bool = False):
+def delete_index(endpoint: str, index_name: str, auth, verify_ssl: bool = False):
     response = requests.delete(f'{endpoint}/{index_name}', auth=auth, verify=verify_ssl)
 
     return response
 
 
-def delete_document(endpoint: str, index_name: str, doc_id: str, auth: Optional[Tuple[str, str]] = None,
+def delete_document(endpoint: str, index_name: str, doc_id: str, auth,
                     verify_ssl: bool = False):
     response = requests.delete(f'{endpoint}/{index_name}/_doc/{doc_id}', auth=auth, verify=verify_ssl)
 
     return response
 
 
-def create_document(endpoint: str, index_name: str, doc_id: str, auth: Optional[Tuple[str, str]] = None,
+def create_document(endpoint: str, index_name: str, doc_id: str, auth,
                     verify_ssl: bool = False):
     document = {
         'title': 'Test Document',
         'content': 'This is a sample document for testing OpenSearch.'
     }
     url = f'{endpoint}/{index_name}/_doc/{doc_id}'
     headers = {'Content-Type': 'application/json'}
-
     response = requests.put(url, headers=headers, data=json.dumps(document), auth=auth, verify=verify_ssl)
 
     return response
 
 
-def get_document(endpoint: str, index_name: str, doc_id: str, auth: Optional[Tuple[str, str]] = None,
+def get_document(endpoint: str, index_name: str, doc_id: str, auth,
                  verify_ssl: bool = False):
     url = f'{endpoint}/{index_name}/_doc/{doc_id}'
     headers = {'Content-Type': 'application/json'}
-
     response = requests.get(url, headers=headers, auth=auth, verify=verify_ssl)
 
     return response

test/requirements.txt

@@ -8,3 +8,5 @@ pytest==7.3.1
 pytest-xdist==3.3.1
 requests==2.31.0
 urllib3==2.0.7
+requests_aws4auth
+boto3

test/tests.py

@@ -13,6 +13,8 @@
 import string
 import secrets
 import pytest
+import boto3
+from requests_aws4auth import AWS4Auth
 
 from requests.exceptions import ConnectionError, SSLError
 
@@ -92,6 +94,12 @@ def init_fixtures(self, proxy_endpoint, source_endpoint, target_endpoint, source
     def setup_authentication(self, auth_type, username, password):
         if auth_type == "basic":
             return (username, password)
+        elif auth_type == "sigv4":
+            session = boto3.Session()
+            credentials = session.get_credentials()
+            aws_auth = AWS4Auth(credentials.access_key, credentials.secret_key, session.region_name, 'es',
+                                session_token=credentials.token)
+            return aws_auth
         return None
 
     def set_common_values(self):