pnh-ttf 1.0.2

Python package to test cloud infrastructure

Table of Contents

• aws
• aws.glue
  • GlueManager
    • __init__
    • check_crawler_status
• aws.lambda_functions
  • LambdaManager
    • __init__
    • invoke_lambda
• aws.s3
  • S3Manager
    • __init__
    • upload_file
    • check_file_exists
• aws.cloudwatch
  • CWManager
    • __init__
    • check_cloudwatch_alarm_state
• aws.stepfunctions
  • StepfunctionsManager
    • __init__
    • start_execution
    • describe_execution

aws

aws.glue

GlueManager Objects

class GlueManager()

__init__

def __init__(crawler_name: str,
             max_attempts: int = 10,
             delay: int = 30,
             region: str = 'eu-west-1',
             profile: str = 'default')

Initialize a GlueManager instance to manage AWS Glue crawlers.

Arguments:

• crawler_name str - The name of the AWS Glue crawler.
• max_attempts int, optional - Maximum number of attempts to check the crawler status. Default is 10.
• delay int, optional - Delay in seconds between status checks. Default is 30 seconds.
• region str, optional - AWS region. Defaults to 'eu-west-1'.
• profile str, optional - AWS profile name. Defaults to 'default'.

check_crawler_status

def check_crawler_status() -> dict

Checks the status of an AWS Glue crawler.

Returns:

• dict - A dictionary containing the status code and status message.

aws.lambda_functions

LambdaManager Objects

class LambdaManager()

__init__

def __init__(lambda_name: str,
             payload: dict = None,
             region: str = 'eu-west-1',
             profile: str = 'default')

Initialize the LambdaManager instance.

Arguments:

• lambda_name str - The name of the AWS Lambda function.
• payload dict, optional - The payload to be sent to the Lambda function. Default is None. Example: payload = '{ "key": "value" }'
• region str, optional - AWS region. Defaults to 'eu-west-1'.
• profile str, optional - AWS profile name. Defaults to 'default'.

invoke_lambda

def invoke_lambda() -> dict

Invoke the AWS Lambda function synchronously and return the status of the invocation.

Returns:

• dict - A dictionary containing the status code and status message.
  • 'statusCode' (int): HTTP status code indicating the result of the invocation.
  • 'statusMessage' (str): Status message describing the outcome of the invocation.

aws.s3

S3Manager Objects

class S3Manager()

__init__

def __init__(bucket_name: str,
             object_name: str,
             file_name: str = None,
             if_modified_since: datetime = None,
             region: str = 'eu-west-1',
             profile: str = 'default')

Initialize an S3Manager instance.

Arguments:

• bucket_name str - The name of the bucket to upload to.
• object_name str - The name of the key to upload to.
• file_name str, optional - The path to the file to upload.
• if_modified_since datetime, optional - Datetime to compare against. Defaults to five minutes earlier than now.
• region str, optional - AWS region. Defaults to 'eu-west-1'.
• profile str, optional - AWS profile name. Defaults to 'default'.

upload_file

def upload_file() -> dict

Upload a file to an S3 bucket.

Returns:

• dict - A dictionary containing the status code and status message.

check_file_exists

def check_file_exists() -> dict

Check if the specified file exists in the S3 bucket and if it was modified since the if_modified_since datetime.

Returns:

• dict - A dictionary containing the status code and status message.

aws.cloudwatch

CWManager Objects

class CWManager()

__init__

def __init__(alarm_name: str,
             start_date: datetime,
             end_date: datetime,
             region: str = 'eu-west-1',
             profile: str = 'default')

Initializes the CWManager with the specified alarm name and date range.

Arguments:

• alarm_name str - The name of the CloudWatch alarm.
• start_date datetime - The start date for the alarm history query.
• end_date datetime - The end date for the alarm history query.
• region str, optional - AWS region. Defaults to 'eu-west-1'.
• profile str, optional - AWS profile name. Defaults to 'default'.

check_cloudwatch_alarm_state

def check_cloudwatch_alarm_state() -> dict

Checks the state history of the specified CloudWatch alarm within the given date range.

Returns:

• dict - A dictionary containing the status code and message indicating whether the alarm state changed to ALARM.

aws.stepfunctions

StepfunctionsManager Objects

class StepfunctionsManager()

__init__

def __init__(statemachine_arn: str,
             payload: str = None,
             execution_arn: str = None,
             max_attempts: int = 10,
             delay: int = 10,
             region: str = 'eu-west-1',
             profile: str = 'default')

Initialize the StepfunctionManager instance.

Arguments:

• statemachine_arn str - The Amazon Resource Name (ARN) of the state machine to execute.
• payload str, optional - The string that contains the JSON input data for the execution. Default is None. Example: "payload": "{"first_name" : "test"}".
• execution_arn str, optional - The Amazon Resource Name (ARN) of the execution to describe. Default is None.
• max_attempts int, optional - Maximum number of attempts to check the execution status. Default is 10.
• delay int, optional - Delay in seconds between status checks. Default is 10 seconds.
• region str, optional - AWS region. Defaults to 'eu-west-1'.
• profile str, optional - AWS profile name. Defaults to 'default'.

start_execution

def start_execution() -> dict

Starts the execution of the specified state machine.

Returns:

• dict - A dictionary containing the status code and status message.

describe_execution

def describe_execution() -> dict

Describes the status of the specified execution.

This method checks the execution status in a loop until it either succeeds, fails, or reaches the maximum number of attempts.

Returns:

• dict - A dictionary containing the status code and status message.