add support for using elastic cloud api key for auth

the introduces the option to use api keys

Ref: https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html

README.md

@@ -60,6 +60,9 @@ For more info on this Elasticsearch feature check their [index documention](http
 pytest --es-address 127.0.0.1:9200
 # or if you need user/password to authenticate
 pytest --es-address my-elk-server.io:9200 --es-username fruch --es-password 'passwordsarenicetohave'
+
+# or with api key (see https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html)
+pytest --es-address my-elk-server.io:9200 --es-api-key 'VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw=='
 ```
 
 ### Configure from code (ideally in conftest.py)
@@ -71,10 +74,12 @@ def pytest_plugin_registered(plugin, manager):
     if isinstance(plugin, ElkReporter):
       # TODO: get credentials in more secure fashion programmatically, maybe AWS secrets or the likes
       # or put them in plain-text in the code... what can ever go wrong...
+      plugin.es_index_name = 'test_data'
       plugin.es_address = "my-elk-server.io:9200"
       plugin.es_user = 'fruch'
       plugin.es_password = 'passwordsarenicetohave'
-      plugin.es_index_name = 'test_data'
+      # or use api key (see https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html
+      plugin.es_api_key = 'VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw=='
 
 ```
 
@@ -87,6 +92,9 @@ es_address = my-elk-server.io:9200
 es_user = fruch
 es_password = passwordsarenicetohave
 es_index_name = test_data
+
+# or with api key (see https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html)
+es_api_key = VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw==
 ```
 
 see [pytest docs](https://docs.pytest.org/en/latest/customize.html)

pytest_elk_reporter.py

@@ -11,6 +11,7 @@
 import pprint
 import fnmatch
 import concurrent.futures
+from typing import Any
 
 import six
 import pytest
@@ -70,6 +71,14 @@ def pytest_addoption(parser):
         help="Elasticsearch password",
     )
 
+    group.addoption(
+        "--es-api-key",
+        action="store",
+        dest="es_api_key",
+        default=None,
+        help="Elasticsearch api key in base64 format",
+    )
+
     group.addoption(
         "--es-timeout",
         action="store",
@@ -106,6 +115,9 @@ def pytest_addoption(parser):
     parser.addini("es_address", help="Elasticsearch address", default=None)
     parser.addini("es_username", help="Elasticsearch username", default=None)
     parser.addini("es_password", help="Elasticsearch password", default=None)
+    parser.addini(
+        "es_api_key", help="Elasticsearch api key in base64 format", default=None
+    )
     parser.addini(
         "es_index_name",
         help="name of the elasticsearch index to save results to",
@@ -154,6 +166,7 @@ def __init__(self, config):
         else:  # default to True
             self.es_post_reports = True
         self.es_address = config.getoption("es_address") or config.getini("es_address")
+        self.es_api_key = config.getoption("es_api_key") or config.getini("es_api_key")
         self.es_username = config.getoption("es_username") or config.getini(
             "es_username"
         )
@@ -192,8 +205,13 @@ def __init__(self, config):
         self.is_slave = False
 
     @property
-    def es_auth(self):
-        return self.es_username, self.es_password
+    def es_auth_args(self) -> dict[str, Any]:
+        output = {}
+        if self.es_api_key:
+            output.update(dict(headers={"Authorization": f"ApiKey {self.es_api_key}"}))
+        if self.es_username and self.es_password:
+            output.update(dict(auth=(self.es_username, self.es_password)))
+        return output
 
     @property
     def es_url(self):
@@ -283,7 +301,7 @@ def report_test(self, item_report, outcome, old_report=None):
             outcome=outcome,
             duration=item_report.duration,
             markers=item_report.keywords,
-            **self.session_data
+            **self.session_data,
         )
         test_data.update(self.test_data[item_report.nodeid])
         del self.test_data[item_report.nodeid]
@@ -318,7 +336,7 @@ def pytest_internalerror(self, excrepr):
             timestamp=datetime.datetime.utcnow().isoformat(),
             outcome="internal-error",
             faiure_message=str(excrepr),
-            **self.session_data
+            **self.session_data,
         )
         self.post_to_elasticsearch(test_data)
 
@@ -327,7 +345,7 @@ def post_to_elasticsearch(self, test_data):
             try:
                 url = "{0.es_url}/{0.es_index_name}/_doc".format(self)
                 res = requests.post(
-                    url, json=test_data, auth=self.es_auth, timeout=self.es_timeout
+                    url, json=test_data, timeout=self.es_timeout, **self.es_auth_args
                 )
                 res.raise_for_status()
             except Exception as ex:  # pylint: disable=broad-except
@@ -363,7 +381,7 @@ def get_test_stats(test_id):
             }
             try:
                 res = session.post(
-                    url, json=body, auth=self.es_auth, timeout=self.es_timeout
+                    url, json=body, timeout=self.es_timeout, **self.es_auth_args
                 )
                 res.raise_for_status()
                 return dict(

tests/test_elk_reporter.py

@@ -87,7 +87,7 @@ def test_skip_in_teardown(skip_in_teardown):
     result.stdout.fnmatch_lines(["*::test_failure_in_fin_2 FAILED*"])
     result.stdout.fnmatch_lines(["*::test_failure_in_fin_3 ERROR*"])
 
-    # make sure that that we get a '1' exit code for the testsuite
+    # make sure that we get a '1' exit code for the testsuite
     assert result.ret == 1
 
     last_report = json.loads(requests_mock.request_history[-1].text)
@@ -130,7 +130,7 @@ def test_sth(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_sth PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
@@ -152,7 +152,7 @@ def test_sth(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_sth PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
@@ -177,7 +177,7 @@ def test_collection_elk(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_collection_elk PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
@@ -197,7 +197,7 @@ def test_collection_elk(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_collection_elk PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
@@ -217,7 +217,7 @@ def test_collection_elk(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_collection_elk PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
@@ -237,11 +237,11 @@ def test_collection_elk(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_collection_elk PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
-def test_setup_es_from_code(testdir):
+def test_setup_es_from_code(testdir, requests_mock):
     # create a temporary pytest test module
     testdir.makepyfile(
         """
@@ -250,7 +250,7 @@ def test_setup_es_from_code(testdir):
         @pytest.fixture(scope='session', autouse=True)
         def configure_es(elk_reporter):
             elk_reporter.es_address = "127.0.0.1:9200"
-            elk_reporter.es_user = 'test'
+            elk_reporter.es_username = 'test'
             elk_reporter.es_password = 'mypassword'
             elk_reporter.es_index_name = 'test_data'
 
@@ -261,11 +261,41 @@ def test_should_pass():
 
     result = testdir.runpytest("-v", "-s")
 
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines(["*::test_should_pass PASSED*"])
+    # make sure that that we get a '0' exit code for the testsuite
+    assert result.ret == 0
+    auth_header = requests_mock.request_history[-1].headers.get("Authorization")
+    assert "Basic" in auth_header
+
+
+def test_setup_es_api_key_from_code(testdir, requests_mock):
+    # create a temporary pytest test module
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.fixture(scope='session', autouse=True)
+        def configure_es(elk_reporter):
+            elk_reporter.es_address = "127.0.0.1:9200"
+            elk_reporter.es_api_key = 'VnVhQ2ZHY0JDZGJrUW0tZTVhT3g6dWkybHAyYXhUTm1zeWFrdzl0dk5udw=='
+            elk_reporter.es_index_name = 'test_data'
+
+        def test_should_pass():
+            pass
+        """
+    )
+
+    result = testdir.runpytest("-v", "-s")
+
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_should_pass PASSED*"])
     # make sure that that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
+    auth_header = requests_mock.request_history[-1].headers.get("Authorization")
+    assert "ApiKey" in auth_header
+
 
 def test_git_info(testdir, requests_mock):  # pylint: disable=redefined-outer-name
 
@@ -293,7 +323,7 @@ def test_should_pass():
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_should_pass PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     last_report = json.loads(requests_mock.request_history[-1].text)
@@ -324,7 +354,7 @@ def test_2():
     result.stdout.fnmatch_lines(["*::test_1 PASSED*"])
     result.stdout.fnmatch_lines(["*::test_2 PASSED*"])
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     first_report = json.loads(requests_mock.request_history[0].text)
@@ -355,7 +385,7 @@ def test_should_pass(elk_reporter):
 
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_should_pass PASSED*"])
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
 
@@ -383,7 +413,7 @@ def test_2():
     result.stdout.fnmatch_lines(["*::test_1 PASSED*"])
     result.stdout.fnmatch_lines(["*::test_2 PASSED*"])
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     first_report = json.loads(requests_mock.request_history[0].text)
@@ -412,7 +442,7 @@ def test_1(record_property):
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_1 PASSED*"])
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     report = json.loads(requests_mock.request_history[0].text)
@@ -442,7 +472,7 @@ def test_1(record_property):
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_1 PASSED*"])
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     report = json.loads(requests_mock.request_history[0].text)
@@ -467,7 +497,7 @@ def test_1(elk_reporter):
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_1 PASSED*"])
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     assert (
@@ -495,7 +525,7 @@ def test_1(elk_reporter):
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines(["*::test_1 PASSED*"])
 
-    # make sure that that we get a '0' exit code for the testsuite
+    # make sure that we get a '0' exit code for the testsuite
     assert result.ret == 0
 
     assert (