clear rest sessions between tests;reset config data between tests;other bits and bobs
This commit is contained in:
parent
a39e0f92a0
commit
b3a768b802
@ -7,6 +7,7 @@ from requests_oauthlib import OAuth2Session
|
|||||||
|
|
||||||
def api_client(call_dict, verify_cert=False, oauth=False):
|
def api_client(call_dict, verify_cert=False, oauth=False):
|
||||||
|
|
||||||
|
method = call_dict["method"]
|
||||||
url = call_dict["url"]
|
url = call_dict["url"]
|
||||||
headers = call_dict["headers"]
|
headers = call_dict["headers"]
|
||||||
body = call_dict["body"]
|
body = call_dict["body"]
|
||||||
@ -15,8 +16,7 @@ def api_client(call_dict, verify_cert=False, oauth=False):
|
|||||||
client = OAuth2Session(token=call_dict["token"])
|
client = OAuth2Session(token=call_dict["token"])
|
||||||
else:
|
else:
|
||||||
client = requests.Session()
|
client = requests.Session()
|
||||||
|
client.cookies.clear()
|
||||||
method = call_dict["method"]
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if method == 'GET':
|
if method == 'GET':
|
||||||
|
@ -1,27 +1,36 @@
|
|||||||
import os
|
import os
|
||||||
from dotenv import load_dotenv
|
from dotenv import dotenv_values
|
||||||
from types import MappingProxyType
|
from types import MappingProxyType
|
||||||
|
|
||||||
|
|
||||||
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||||
|
|
||||||
|
# Dict to hold values from dotenv files; keys are filenames, values are dictionaries of env vars
|
||||||
|
_dotenv_values = {}
|
||||||
|
|
||||||
def get_cfg(env='qa'):
|
|
||||||
|
def get_cfg(env='qa', force_refresh=False):
|
||||||
environment_name = os.getenv('ENV_NAME', env)
|
environment_name = os.getenv('ENV_NAME', env)
|
||||||
dotenv_path = os.path.join(PROJECT_ROOT, f'.env.{environment_name}')
|
dotenv_path = os.path.join(PROJECT_ROOT, f'.env.{environment_name}')
|
||||||
if not os.path.exists(dotenv_path):
|
if not os.path.exists(dotenv_path):
|
||||||
raise FileNotFoundError(f"{dotenv_path} does not exist")
|
raise FileNotFoundError(f"{dotenv_path} does not exist")
|
||||||
load_dotenv(dotenv_path)
|
|
||||||
|
if dotenv_path not in _dotenv_values or force_refresh:
|
||||||
|
# Load the env vars from the dotenv file and store them in the _dotenv_values dict
|
||||||
|
_dotenv_values[dotenv_path] = dotenv_values(dotenv_path)
|
||||||
|
|
||||||
|
# Now, we're sure that _dotenv_values[dotenv_path] contains up-to-date env vars from dotenv_path
|
||||||
|
dotenv_vars = _dotenv_values[dotenv_path]
|
||||||
|
|
||||||
config_dict = {
|
config_dict = {
|
||||||
'client_id': os.getenv('CLIENT_ID'),
|
'client_id': dotenv_vars.get('CLIENT_ID'),
|
||||||
'client_secret': os.getenv('CLIENT_SECRET'),
|
'client_secret': dotenv_vars.get('CLIENT_SECRET'),
|
||||||
'token_url': os.getenv('TOKEN_FETCH_URL'),
|
'token_url': dotenv_vars.get('TOKEN_FETCH_URL'),
|
||||||
'login': os.getenv('LOGIN'),
|
'login': dotenv_vars.get('LOGIN'),
|
||||||
'password': os.getenv('PASSWORD'),
|
'password': dotenv_vars.get('PASSWORD'),
|
||||||
'audience': os.getenv('AUDIENCE'),
|
'audience': dotenv_vars.get('AUDIENCE'),
|
||||||
'scopes': os.getenv('SCOPES', '').split(','),
|
'scopes': dotenv_vars.get('SCOPES', '').split(','),
|
||||||
'api_url': os.getenv('API_URL')
|
'api_url': dotenv_vars.get('API_URL')
|
||||||
}
|
}
|
||||||
config = MappingProxyType(config_dict) # immutable dict
|
config = MappingProxyType(config_dict) # immutable dict
|
||||||
return config
|
return config
|
||||||
|
20
tests/conftest.py
Normal file
20
tests/conftest.py
Normal file
@ -0,0 +1,20 @@
|
|||||||
|
import pytest
|
||||||
|
import requests
|
||||||
|
from requests_oauthlib import OAuth2Session
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="function")
|
||||||
|
def reset_open_session():
|
||||||
|
open_session = requests.Session()
|
||||||
|
open_session.cookies.clear()
|
||||||
|
yield open_session
|
||||||
|
open_session.close()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="function")
|
||||||
|
def reset_oauth_session():
|
||||||
|
oauth_session = OAuth2Session()
|
||||||
|
oauth_session.cookies.clear()
|
||||||
|
yield oauth_session
|
||||||
|
oauth_session.close()
|
||||||
|
|
@ -2,14 +2,14 @@ import pytest
|
|||||||
from apiclient.config import get_cfg
|
from apiclient.config import get_cfg
|
||||||
from apiclient.oauth_helper import get_legacy_token
|
from apiclient.oauth_helper import get_legacy_token
|
||||||
|
|
||||||
ENV = 'qa' # This would be set in an actual OS env var on the execution platform
|
|
||||||
CFG = get_cfg(ENV) # needed for the token, and the full api url
|
|
||||||
|
|
||||||
|
ENV = 'qa'
|
||||||
|
CFG = get_cfg(ENV, force_refresh=True)
|
||||||
application_json = {'Content-Type': 'application/json'}
|
application_json = {'Content-Type': 'application/json'}
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def role_get_call():
|
def role_get_call(reset_oauth_session):
|
||||||
token = get_legacy_token(ENV)
|
token = get_legacy_token(ENV)
|
||||||
api_call = {
|
api_call = {
|
||||||
"token": token,
|
"token": token,
|
||||||
@ -22,7 +22,7 @@ def role_get_call():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def acms_redaction_get_call():
|
def acms_redaction_get_call(reset_oauth_session):
|
||||||
token = get_legacy_token(ENV)
|
token = get_legacy_token(ENV)
|
||||||
api_call = {
|
api_call = {
|
||||||
"token": token,
|
"token": token,
|
||||||
@ -35,7 +35,7 @@ def acms_redaction_get_call():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def client_applications_get_call():
|
def client_applications_get_call(reset_oauth_session):
|
||||||
token = get_legacy_token(ENV)
|
token = get_legacy_token(ENV)
|
||||||
api_call = {
|
api_call = {
|
||||||
"token": token,
|
"token": token,
|
||||||
@ -48,7 +48,7 @@ def client_applications_get_call():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def endpoint_get_call():
|
def endpoint_get_call(reset_oauth_session):
|
||||||
token = get_legacy_token(ENV)
|
token = get_legacy_token(ENV)
|
||||||
api_call = {
|
api_call = {
|
||||||
"token": token,
|
"token": token,
|
||||||
@ -63,7 +63,7 @@ def endpoint_get_call():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def redaction_type_get_call():
|
def redaction_type_get_call(reset_oauth_session):
|
||||||
token = get_legacy_token(ENV)
|
token = get_legacy_token(ENV)
|
||||||
api_call = {
|
api_call = {
|
||||||
"token": token,
|
"token": token,
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
import json
|
import json
|
||||||
import pytest
|
import pytest
|
||||||
from apiclient.client import api_client
|
from apiclient.client import api_client
|
||||||
from oauth_service.helpers import get_expected_response
|
from tests.oauth_service.helpers import get_expected_response
|
||||||
|
|
||||||
|
|
||||||
####
|
####
|
||||||
@ -14,7 +14,7 @@ from oauth_service.helpers import get_expected_response
|
|||||||
@pytest.mark.role
|
@pytest.mark.role
|
||||||
def test_datadelivery_role_get(request, role_get_call):
|
def test_datadelivery_role_get(request, role_get_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(role_get_call)
|
actual_response = api_client(role_get_call, oauth=True)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
||||||
|
|
||||||
|
|
||||||
@ -22,7 +22,7 @@ def test_datadelivery_role_get(request, role_get_call):
|
|||||||
@pytest.mark.redaction
|
@pytest.mark.redaction
|
||||||
def test_datadelivery_acms_redaction_get(request, acms_redaction_get_call):
|
def test_datadelivery_acms_redaction_get(request, acms_redaction_get_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(acms_redaction_get_call)
|
actual_response = api_client(acms_redaction_get_call, oauth=True)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
||||||
|
|
||||||
|
|
||||||
@ -30,7 +30,7 @@ def test_datadelivery_acms_redaction_get(request, acms_redaction_get_call):
|
|||||||
@pytest.mark.client_application
|
@pytest.mark.client_application
|
||||||
def test_datadelivery_client_applications_get(request, client_applications_get_call):
|
def test_datadelivery_client_applications_get(request, client_applications_get_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(client_applications_get_call)
|
actual_response = api_client(client_applications_get_call, oauth=True)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
||||||
|
|
||||||
|
|
||||||
@ -38,7 +38,7 @@ def test_datadelivery_client_applications_get(request, client_applications_get_c
|
|||||||
@pytest.mark.endpoint
|
@pytest.mark.endpoint
|
||||||
def test_datadelivery_endpoint_get(request, endpoint_get_call):
|
def test_datadelivery_endpoint_get(request, endpoint_get_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(endpoint_get_call)
|
actual_response = api_client(endpoint_get_call, oauth=True)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
||||||
|
|
||||||
|
|
||||||
@ -46,5 +46,5 @@ def test_datadelivery_endpoint_get(request, endpoint_get_call):
|
|||||||
@pytest.mark.redaction
|
@pytest.mark.redaction
|
||||||
def test_datadelivery_redaction_type_get(request, redaction_type_get_call):
|
def test_datadelivery_redaction_type_get(request, redaction_type_get_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(redaction_type_get_call)
|
actual_response = api_client(redaction_type_get_call, oauth=True)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
@ -1,29 +1,27 @@
|
|||||||
import pytest
|
import pytest
|
||||||
from apiclient.config import get_cfg
|
from apiclient.config import get_cfg
|
||||||
|
|
||||||
ENV = 'gw' # This would be set in an actual OS env var on the execution platform
|
|
||||||
CFG = get_cfg(ENV) # needed for the token, and the full api url
|
|
||||||
|
|
||||||
application_json = {'Content-Type': 'application/json'}
|
CFG = get_cfg('gw', force_refresh=True)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def get_creatures_call():
|
def get_creatures_call(reset_open_session):
|
||||||
api_call = {
|
api_call = {
|
||||||
"method": "GET",
|
"method": "GET",
|
||||||
"url": CFG["api_url"] + '/rules/creature',
|
"url": CFG["api_url"] + '/rules/creature',
|
||||||
"headers": application_json,
|
"headers": {},
|
||||||
"body": {}
|
"body": {}
|
||||||
}
|
}
|
||||||
return api_call
|
return api_call
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def get_creature_call():
|
def get_creature_call(reset_open_session):
|
||||||
api_call = {
|
api_call = {
|
||||||
"method": "GET",
|
"method": "GET",
|
||||||
"url": CFG["api_url"] + '/rules/creature',
|
"url": CFG["api_url"] + '/rules/creature',
|
||||||
"headers": application_json,
|
"headers": {},
|
||||||
"body": {"creature": "badder"}
|
"body": {"creature": "badder"}
|
||||||
}
|
}
|
||||||
return api_call
|
return api_call
|
||||||
|
@ -2,18 +2,18 @@ import json
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from apiclient.client import api_client
|
from apiclient.client import api_client
|
||||||
from open_service.helpers import get_expected_response
|
from tests.open_service.helpers import get_expected_response
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.rules
|
@pytest.mark.rules
|
||||||
def test_get_creatures(request, get_creatures_call):
|
def test_get_creatures(request, get_creatures_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(get_creatures_call)
|
actual_response = api_client(get_creatures_call, oauth=False)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.rules
|
@pytest.mark.rules
|
||||||
def test_get_creature(request, get_creature_call):
|
def test_get_creature(request, get_creature_call):
|
||||||
expected_response = get_expected_response(request.node.name)
|
expected_response = get_expected_response(request.node.name)
|
||||||
actual_response = api_client(get_creature_call)
|
actual_response = api_client(get_creature_call, oauth=False)
|
||||||
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
|
Loading…
Reference in New Issue
Block a user