Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
A simple example to demonstrate how I validate source availability by dynamically calling verify functions.
""" availabilitychecker.py """
import boto3
import requests
class AvailabilityChecker:
def __init__(self):
self.ssm_client = boto3.client("ssm")
self.session = requests.Session()
def _get_param(self, key):
""" Return the SSM parameter """
return self.ssm_client.get_parameter(Name=key, WithDecryption=True)["Parameter"]["Value"]
def _get_verify_functions(self):
""" Return verify functions inside this class """
return [func for func in dir(self) if callable(getattr(self, func)) and func.startswith("verify")]
def verify_hibob_is_available(self):
""" Verify that the HiBob API is available """
api_key = self._get_param("HIBOB_API_KEY")
api_url = self._get_param("HIBOB_API_URL")
arguments = {
'method': 'get',
'url': api_url,
'headers': {
"Authorization": api_key
}
}
return self.session.request(**arguments).status_code == 200
def verify_jira_is_available(self):
""" Verify that the Jira API is available """
api_key = self._get_param("JIRA_API_KEY")
api_url = self._get_param("JIRA_URL")
jira_user = self._get_param("JIRA_USER")
arguments = {
'method': 'get',
'url': f"{api_url}/rest/api/2/project",
'auth': (jira_user, api_key)
}
return self.session.request(**arguments).status_code == 200
def verify_workable_api_is_available(self):
""" Verify that the Workable API is available """
api_key = self._get_param("WORKABLE_API_KEY")
api_url = self._get_param("WORKABLE_API_URL")
arguments = {
'method': 'get',
'url': f"{api_url}jobs",
'headers': {
"Authorization": "Bearer {}".format(api_key)
},
'params': {
"limit": 1,
"include_fields": "description"
}
}
return self.session.request(**arguments).status_code == 200
def lambda_handler(event, context):
avc = AvailabilityChecker()
for method in avc._get_verify_functions():
is_available = getattr(avc, method)()
source = ' '.join(method.split('_')[1:-2]).title()
if not is_available:
print(f"[NOK] Please check availability for `{source}`.")
else:
print(f"[OK] `{source}`")
lambda_handler({}, {})
""" testsourceavailability.py """
import boto3
import pytest
import requests
class FixtureClass:
def get_ssm_parameters(self):
""" Returns the SSM parameters """
paginator = self.ssm_client.get_paginator("get_parameters_by_path")
iterator = paginator.paginate(Path="/", WithDecryption=True)
params = {}
for page in iterator:
for param in page.get("Parameters", []):
params[param.get("Name")] = param.get("Value")
return params
@property
def session(self):
return boto3.session.Session()
@property
def ssm_client(self):
return self.session.client("ssm")
fc = FixtureClass()
@pytest.fixture(scope="session")
def ssm_parameters():
return fc.get_ssm_parameters()
class TestSourceAvailability:
""" Defines the tests to verify the source availability """
@pytest.fixture(autouse=True)
def setup_class(self, ssm_parameters):
""" Setup the test class """
self.ssm_parameters = ssm_parameters
self.session = requests.Session()
def _get_param(self, key):
return self.ssm_parameters.get(key)
def test_hibob_is_available(self):
""" Test that the HiBob API is available """
api_key = self._get_param("HIBOB_API_KEY")
api_url = self._get_param("HIBOB_API_URL")
kwargs = {
'method': 'get',
'url': api_url,
'headers': {
"Authorization": api_key
}
}
assert self.session.request(**kwargs).status_code == 200
def test_jira_is_available(self):
""" Test that the Jira API is available """
api_key = self._get_param("JIRA_API_KEY")
api_url = self._get_param("JIRA_URL")
jira_user = self._get_param("JIRA_USER")
kwargs = {
'method': 'get',
'url': f"{api_url}/rest/api/2/project",
'auth': (jira_user, api_key)
}
assert self.session.request(**kwargs).status_code == 200
def test_jira_is_unavailable_without_api_key(self):
""" Test that the Jira API is unavailable without API key """
api_key = None
api_url = self._get_param("JIRA_URL")
jira_user = self._get_param("JIRA_USER")
kwargs = {
'method': 'get',
'url': f"{api_url}/rest/api/2/project",
'auth': (jira_user, "")
}
assert self.session.request(**kwargs).status_code == 401
def test_workable_api_is_available(self):
""" Test that the Workable API is available """
api_key = self._get_param("WORKABLE_API_KEY")
api_url = self._get_param("WORKABLE_API_URL")
kwargs = {
'method': 'get',
'url': f"{api_url}jobs",
'headers': {"Authorization": "Bearer {}".format(api_key)},
'params': {"limit": 1, "include_fields": "description"}
}
assert self.session.request(**kwargs).status_code == 200
def test_workable_api_is_not_available_for_wrong_credentials(self):
""" Test that the Workable API is not available for wrong credentials """
api_key = "fake_key"
api_url = self._get_param("WORKABLE_API_URL")
kwargs = {
'method': 'get',
'url': f"{api_url}jobs",
'headers': {"Authorization": "Bearer {}".format(api_key)},
'params': {"limit": 1, "include_fields": "description"}
}
assert self.session.request(**kwargs).status_code == 401
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment