example_self_tested_lambda/src/lambda_verification.py

52 lines
1.8 KiB
Python

""" Generic lambda function tester that uses environmental variables """
# pylint: disable=broad-except
import os
import boto3
AWSLAMBDA = boto3.client("lambda")
CODEDEPLOY = boto3.client("codedeploy")
# pylint: disable=unused-argument
def handler(event, context):
""" Entry point for test """
try:
function = os.environ["CurrentVersion"]
expected = os.environ["TestExpected"]
payload = os.environ["TestData"]
codedeploy_params = {
"deploymentId": event["DeploymentId"],
"lifecycleEventHookExecutionId": event["LifecycleEventHookExecutionId"],
"status": "Failed",
}
except Exception as error:
print(f"Exception, failed to unpack event, {function}, {error}, event, {event}")
raise error
try:
print(f"Info, testing, {function}, {payload}")
response = AWSLAMBDA.invoke(FunctionName=function, Payload=payload)
response_value = response["Payload"].read()
except Exception as error:
print(f"Failed, invocation failed, {function}, {error}")
CODEDEPLOY.put_lifecycle_event_hook_execution_status(**codedeploy_params)
return
try:
assert response_value == expected
except Exception as error:
print(
f"Failed, incorrect response, {function}, {error}, expected, {expected}, actual, {response}"
)
CODEDEPLOY.put_lifecycle_event_hook_execution_status(**codedeploy_params)
return
try:
print(f"Success, notifying codedeploy, {function}, {event['DeploymentId']}")
codedeploy_params["status"] = "Succeeded"
CODEDEPLOY.put_lifecycle_event_hook_execution_status(**codedeploy_params)
return
except Exception as error:
print(f"Exception, failed to notify codedeploy, {function}, {error}")
raise error