A simple example to demonstrate how I validate source availability by dynamically calling verify functions.

availabilitychecker.py

""" 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

""" 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