initial commit

.gitignore

@@ -0,0 +1,4 @@
+.idea/
+.venv/
+**/__pycache__/
+**/.env*

apiclient/client.py

@@ -0,0 +1,38 @@
+import sys
+from requests import RequestException
+from requests_oauthlib import OAuth2Session
+
+
+def api_client(call_dict, verify_cert=False):
+
+    url = call_dict["url"]
+    headers = call_dict["headers"]
+    body = call_dict["body"]
+
+    client = OAuth2Session(token=call_dict["token"])
+    method = call_dict["method"]
+
+    try:
+        if method == 'GET':
+            response = client.get(url, headers=headers, params=body, verify=verify_cert)
+        elif method == 'POST':
+            response = client.post(url, headers=headers, json=body, verify=verify_cert)
+        elif method == 'PUT':
+            response = client.put(url, headers=headers, json=body, verify=verify_cert)
+        elif method == 'OPTIONS':
+            response = client.options(url, headers=headers, json=body, verify=verify_cert)
+        elif method == 'DELETE':
+            response = client.delete(url, verify=verify_cert)
+        else:
+            raise ValueError(f"Invalid method: {method}")
+
+    except RequestException as e:
+        print(f"Request failed. Method: {method}, URL: {url}", file=sys.stderr)
+        print(f"Error details: {str(e)}", file=sys.stderr)
+        exit(e.response.status_code)
+
+    if response.status_code == 200:
+        return response.json()
+    else:
+        return response.status_code, response.reason
+

apiclient/config.py

@@ -0,0 +1,27 @@
+import os
+from dotenv import load_dotenv
+from types import MappingProxyType
+
+
+PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def get_cfg(env='qa'):
+    environment_name = os.getenv('ENV_NAME', env)
+    dotenv_path = os.path.join(PROJECT_ROOT, f'.env.{environment_name}')
+    if not os.path.exists(dotenv_path):
+        raise FileNotFoundError(f"{dotenv_path} does not exist")
+    load_dotenv(dotenv_path)
+
+    config_dict = {
+        'client_id': os.getenv('CLIENT_ID'),
+        'client_secret': os.getenv('CLIENT_SECRET'),
+        'token_url': os.getenv('TOKEN_FETCH_URL'),
+        'login': os.getenv('LOGIN'),
+        'password': os.getenv('PASSWORD'),
+        'audience': os.getenv('AUDIENCE'),
+        'scopes': os.getenv('SCOPES', '').split(','),
+        'api_url': os.getenv('API_URL')
+    }
+    config = MappingProxyType(config_dict)  # immutable dict
+    return config

apiclient/env_template

@@ -0,0 +1,8 @@
+API_URL=https://api.dummy.com
+CLIENT_ID=client_id
+CLIENT_SECRET=client_secret
+TOKEN_URL=oauth_token_fetch_url
+USERNAME=username
+PASSWORD=password
+AUDIENCE=intended_audience
+SCOPES=scope1,scope2

apiclient/main.py

@@ -0,0 +1,22 @@
+import json
+from apiclient.config import get_cfg
+from apiclient.client import api_client
+from apiclient.oauth_helper import get_legacy_token
+
+
+ENV = 'qa'
+cfg = get_cfg(ENV)
+
+if __name__ == "__main__":
+    token = get_legacy_token(ENV)
+
+    api_call = {
+        "token": token,
+        "method": "GET",
+        "url": cfg["api_url"] + '/data-delivery/role',
+        "headers": {'Content-Type': 'application/json'},
+        "body": {"application_id": 1}
+    }
+
+    api_response = api_client(api_call)
+    print(json.dumps(api_response, indent=4))

apiclient/oauth_helper.py

@@ -0,0 +1,27 @@
+from oauthlib.oauth2 import LegacyApplicationClient, OAuth2Error
+from requests_oauthlib import OAuth2Session
+from apiclient.config import get_cfg
+import sys
+
+
+def get_legacy_token(env):
+    cfg = get_cfg(env)
+
+    client = LegacyApplicationClient(client_id=cfg['client_id'])
+    oauth = OAuth2Session(client=client)
+
+    try:
+        token = oauth.fetch_token(
+            token_url=cfg['token_url'],
+            username=cfg['login'],
+            password=cfg['password'],
+            client_id=cfg['client_id'],
+            client_secret=cfg['client_secret'],
+            audience=cfg['audience'],
+            scope=cfg['scopes']
+        )
+    except OAuth2Error as e:
+        print("OAuth2 Error: ", str(e), file=sys.stderr)
+        exit(e.status_code)
+
+    return token

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+addopts = --durations=100 -rA

requirements.txt

@@ -0,0 +1,5 @@
+oauthlib~=3.2.2
+requests~=2.32.3
+requests-oauthlib~=2.0.0
+python-dotenv~=1.0.1
+pytest~=8.2.2

tests/expected_responses/test_datadelivery_role_get.json

@@ -0,0 +1,184 @@
+[
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 151,
+        "role_name": "Administrator",
+        "role_standard": 1
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5362,
+        "role_name": "API_Non_Admin_ViewOnly_Enginetypes",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6459,
+        "role_name": "API_TEST_MLBCUJ",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6363,
+        "role_name": "API_TEST_MPOYIM",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6354,
+        "role_name": "API_TEST_STOZBG",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5844,
+        "role_name": "clone_admin_role",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6200,
+        "role_name": "create_aircraft_groups_nonadmin_api",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6223,
+        "role_name": "create_aircraft_groups_nonadmin_api01",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6043,
+        "role_name": "create_not_allowed",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6056,
+        "role_name": "create_not_allowed_1",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6059,
+        "role_name": "create_not_allowed_22",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5978,
+        "role_name": "delete_non_admin_airframer",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5372,
+        "role_name": "manage_config_role",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5537,
+        "role_name": "new_role",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5733,
+        "role_name": "new_role_sample",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5839,
+        "role_name": "Sample_data",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5548,
+        "role_name": "smample_role_009",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5546,
+        "role_name": "t_role",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 1,
+        "role_id": 152,
+        "role_name": "TDY Administrator",
+        "role_standard": 1
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5628,
+        "role_name": "Test_4",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5538,
+        "role_name": "Test_Sample_21",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 4602,
+        "role_name": "test_sample_3        ",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 4609,
+        "role_name": "test_sample_4        ",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 5736,
+        "role_name": "Test_Sample_data",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 4595,
+        "role_name": "test_sample3        ",
+        "role_standard": 0
+    },
+    {
+        "application_id": 1,
+        "role_admin": 0,
+        "role_id": 6257,
+        "role_name": "view_aircraftmodel_nonadmin1",
+        "role_standard": 0
+    }
+]

tests/test_datadelivery.py

@@ -0,0 +1,43 @@
+import os
+import json
+import pytest
+from apiclient.client import api_client
+from apiclient.config import get_cfg
+from apiclient.oauth_helper import get_legacy_token
+
+
+ENV = 'qa'
+CFG = get_cfg(ENV)
+
+CWD = os.path.dirname(os.path.realpath(__file__))
+
+
+@pytest.fixture
+def api_call():
+    token = get_legacy_token(ENV)
+    api_call = {
+        "token": token,
+        "method": "GET",
+        "url": CFG["api_url"] + '/data-delivery/role',
+        "headers": {'Content-Type': 'application/json'},
+        "body": {"application_id": 1}
+    }
+    return api_call
+
+
+def test_datadelivery_role_get(request, api_call):
+
+    expected_response = get_expected_response(request.node.name)
+    actual_response = api_client(api_call)
+
+    assert json.dumps(actual_response, indent=4) == json.dumps(expected_response, indent=4)
+
+
+####
+# HELPERS
+####
+def get_expected_response(test_name):
+    expected_response_file = test_name + ".json"
+    with open(os.path.join(CWD, 'expected_responses', expected_response_file), 'r') as file:
+        expected_response = json.load(file)
+        return expected_response