From e443c3bc143e91ea0497a5fea5bd7d798f8f8825 Mon Sep 17 00:00:00 2001
From: yan <yan_sushchynski@epam.com>
Date: Wed, 3 Feb 2021 13:52:41 +0300
Subject: [PATCH] GONRG-1732: Fix end2end tests
---
.gitlab-ci.yml | 8 ++++----
src/dags/libs/create_records.py | 12 ++++++++----
src/dags/libs/refresh_token.py | 9 +++++----
tests/end-to-end-tests/mock-external-apis/app.py | 1 +
tests/end-to-end-tests/osdu_api_config.yaml | 1 +
tests/set_airflow_env.sh | 2 ++
tests/test_dags.py | 7 ++++++-
7 files changed, 27 insertions(+), 13 deletions(-)
create mode 100644 tests/end-to-end-tests/osdu_api_config.yaml
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 7d23f9b..2ad784b 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -22,7 +22,7 @@ stages:
- deploy
pylint:
- image: johnybear/osdu-airflow:python36-1
+ image: johnybear/osdu-airflow:python36-2
stage: linters
allow_failure: true
script:
@@ -32,7 +32,7 @@ pylint:
- exit ${EXIT_CODE}
isort:
- image: johnybear/osdu-airflow:python36-1
+ image: johnybear/osdu-airflow:python36-2
allow_failure: true
stage: linters
script:
@@ -42,7 +42,7 @@ isort:
test_dags:
stage: test_dags
- image: johnybear/osdu-airflow:python36-1
+ image: johnybear/osdu-airflow:python36-2
script:
- chmod +x tests/test_dags.sh
- tests/./test_dags.sh || EXIT_CODE=$?
@@ -55,7 +55,7 @@ test_dags:
unit_tests:
stage: unit_tests
- image: johnybear/osdu-airflow:python36-1
+ image: johnybear/osdu-airflow:python36-2
script:
- chmod +x tests/unit_tests.sh
- tests/./unit_tests.sh || EXIT_CODE=$?
diff --git a/src/dags/libs/create_records.py b/src/dags/libs/create_records.py
index 19995af..8e33972 100644
--- a/src/dags/libs/create_records.py
+++ b/src/dags/libs/create_records.py
@@ -17,9 +17,12 @@
import configparser
import logging
+import os
from airflow.models import Variable
-from libs.refresh_token import AirflowTokenRefresher, refresh_token
+from libs.context import Context
+from libs.refresh_token import AirflowTokenRefresher
+from osdu_api.libs.auth.authorization import authorize
from osdu_api.model.acl import Acl
from osdu_api.model.legal.legal import Legal
from osdu_api.model.legal.legal_compliance import LegalCompliance
@@ -40,7 +43,8 @@ DEFAULT_SOURCE = config.get("DEFAULTS", "authority")
DEFAULT_VERSION = config.get("DEFAULTS", "kind_version")
-@refresh_token(AirflowTokenRefresher())
+
+@authorize(AirflowTokenRefresher())
def create_update_record_request(headers, record_client, record):
"""Send request to create or update records via RecordClient.
@@ -49,7 +53,7 @@ def create_update_record_request(headers, record_client, record):
:param record: The record to create or update
:return: Storage service response
"""
- resp = record_client.create_update_records([record], headers.items())
+ resp = record_client.create_update_records([record], headers)
return resp
@@ -91,7 +95,7 @@ def create_records(**kwargs):
"AppKey": data_conf.get("app-key", "")
}
- record_client = RecordClient()
+ record_client = RecordClient(AirflowTokenRefresher(), Context("test", "test"))
record_client.data_partition_id = data_conf.get(
"partition-id", DEFAULT_SOURCE)
resp = create_update_record_request(headers, record_client, record)
diff --git a/src/dags/libs/refresh_token.py b/src/dags/libs/refresh_token.py
index 901055b..6146da7 100644
--- a/src/dags/libs/refresh_token.py
+++ b/src/dags/libs/refresh_token.py
@@ -25,6 +25,7 @@ from typing import Any, Callable, Union
import requests
from libs.exceptions import TokenRefresherNotPresentError
+from osdu_api.libs.auth.authorization import TokenRefresher as OSDUAPITokenRefresher
from providers import credentials
from providers.types import BaseCredentials
from tenacity import retry, stop_after_attempt
@@ -59,7 +60,7 @@ class TokenRefresher(ABC):
@property
@abstractmethod
def authorization_header(self) -> dict:
- """Authorization header. Must return authorization header for
+ """Authorization header. Must return authorization header for
updating headers dict.
E.g. return {"Authorization": "Bearer <access_token>"}
@@ -69,7 +70,7 @@ class TokenRefresher(ABC):
pass
-class AirflowTokenRefresher(TokenRefresher):
+class AirflowTokenRefresher(TokenRefresher, OSDUAPITokenRefresher):
"""Simple wrapper for credentials to be used in refresh_token decorator within Airflow."""
def __init__(self, creds: BaseCredentials = None):
@@ -124,7 +125,7 @@ def make_callable_request(obj: Union[object, None], request_function: Callable,
:type request_function: Callable
:param headers: The request headers
:type headers: dict
- :return: A partial callable
+ :return: A partial callable
:rtype: Callable
"""
if obj: # if wrapped function is an object's method
@@ -190,7 +191,7 @@ def send_request_with_auth_header(token_refresher: TokenRefresher, *args,
:param token_refresher: The token refresher instance
:type token_refresher: TokenRefresher
:raises e: Re-raises any requests.HTTPError
- :return: The server response
+ :return: The server response
:rtype: requests.Response
"""
obj = kwargs.pop("obj", None)
diff --git a/tests/end-to-end-tests/mock-external-apis/app.py b/tests/end-to-end-tests/mock-external-apis/app.py
index 7be8cd1..5e10f45 100644
--- a/tests/end-to-end-tests/mock-external-apis/app.py
+++ b/tests/end-to-end-tests/mock-external-apis/app.py
@@ -37,6 +37,7 @@ def index():
return 'index'
@app.route('/st', methods=['POST', 'GET', "PUT"])
+@app.route('/st/api/storage/v2/records/', methods=['POST', 'GET', "PUT"])
def storage():
logger.info(request.json)
with open("/tmp/osdu_ingest_result", "w") as f:
diff --git a/tests/end-to-end-tests/osdu_api_config.yaml b/tests/end-to-end-tests/osdu_api_config.yaml
new file mode 100644
index 0000000..11f461f
--- /dev/null
+++ b/tests/end-to-end-tests/osdu_api_config.yaml
@@ -0,0 +1 @@
+airflow_vars: 1
diff --git a/tests/set_airflow_env.sh b/tests/set_airflow_env.sh
index 3bdcc1f..13f1c05 100755
--- a/tests/set_airflow_env.sh
+++ b/tests/set_airflow_env.sh
@@ -64,7 +64,9 @@ mkdir -p /usr/local/airflow/dags/
cp -rf src/* /usr/local/airflow/
cp -r tests/end-to-end-tests/mock-external-apis /mock-server
cp -r tests/end-to-end-tests/mock-data /mock-server/mock-data
+
cp tests/end-to-end-tests/{test-osdu-ingest-r2-success.sh,test-osdu-ingest-r2-fail.sh,test-default-ingest-{success,fail}.sh} /mock-server/
+cp tests/end-to-end-tests/osdu_api_config.yaml /mock-server/
cp tests/*.py /mock-server/
chmod +x /mock-server/{test-osdu-ingest-r2-success.sh,test-osdu-ingest-r2-fail.sh,test-default-ingest-{success,fail}.sh}
diff --git a/tests/test_dags.py b/tests/test_dags.py
index 43a99ac..3fe61f3 100644
--- a/tests/test_dags.py
+++ b/tests/test_dags.py
@@ -14,6 +14,7 @@
# limitations under the License.
import enum
+import os
import subprocess
import time
@@ -23,6 +24,7 @@ class DagStatus(enum.Enum):
FAILED = enum.auto()
FINISHED = enum.auto()
+
OSDU_INGEST_SUCCESS_SH = "/mock-server/./test-osdu-ingest-r2-success.sh"
OSDU_INGEST_FAIL_SH = "/mock-server/./test-osdu-ingest-r2-fail.sh"
DEFAULT_INGEST_SUCCESS_SH = "/mock-server/./test-default-ingest-success.sh"
@@ -31,15 +33,17 @@ DEFAULT_INGEST_FAIL_SH = "/mock-server/./test-default-ingest-fail.sh"
with open("/tmp/osdu_ingest_result", "w") as f:
f.close()
+os.environ["OSDU_API_CONFIG"] = "/mock-server/osdu_api_config.yaml"
subprocess.run(f"/bin/bash -c 'airflow scheduler > /dev/null 2>&1 &'", shell=True)
+
def check_dag_status(dag_name):
time.sleep(5)
output = subprocess.getoutput(f'airflow list_dag_runs {dag_name}')
if "failed" in output:
print(dag_name)
print(output)
- return DagStatus.FAILED
+ return DagStatus.FAILED
if "running" in output:
return DagStatus.RUNNING
print(dag_name)
@@ -59,6 +63,7 @@ def test_dag_success(dag_name, script):
else:
raise Exception(f"Error {dag_name} supposed to be finished")
+
def test_dag_fail(dag_name, script):
subprocess.run(f"{script}", shell=True)
print(f"Expecting {dag_name} fail")
--
GitLab