From 1b17a10f9f2fc178ebadf9cdf0b7b1a508fdb863 Mon Sep 17 00:00:00 2001 From: Israel Fruchter Date: Wed, 24 Jul 2024 11:02:06 +0300 Subject: [PATCH] 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 | 10 +++++- pytest_elk_reporter.py | 30 ++++++++++++++---- tests/test_elk_reporter.py | 64 ++++++++++++++++++++++++++++---------- 3 files changed, 80 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 80c23e9..c0599b7 100644 --- a/README.md +++ b/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) diff --git a/pytest_elk_reporter.py b/pytest_elk_reporter.py index 4a45ae5..8b93fe2 100644 --- a/pytest_elk_reporter.py +++ b/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( diff --git a/tests/test_elk_reporter.py b/tests/test_elk_reporter.py index 6393398..46f6386 100644 --- a/tests/test_elk_reporter.py +++ b/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 (