How to use send_task_failure method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

...18JIRA_SUMMARY = "Requesting admin access to production"19JIRA_ISSUE_TYPE = "Task"20# we want to explicitly tell AWS step functions about the error21# if we are calling the lambda outside of step functions, just log the err22def send_task_failure(task_token, err=None):23    if task_token:24        logging.error("Notifying task of failure.", err)25        sf_client.send_task_failure(26            taskToken=task_token, error="error", cause=err)27    else:28        logging.error("No task to notify, but failure occurred.")29        logging.error(err)30def create_jira_ticket(event, task_token, jira_client):31    reason = event.get('reason')32    if reason is None:33        send_task_failure(task_token, "reason not found in input json")34        return False35    createDict = {36        "summary": JIRA_SUMMARY,37        "issuetype": {"name": JIRA_ISSUE_TYPE},38        "project": {"key": JIRA_PROJECT_KEY},39        "assignee": {"id": JIRA_ASSIGNEE_ID},40        "description": reason41    }42    logging.info(f"Creating ticket using payload: {json.dumps(createDict)}")43    try:44        # https://developer.atlassian.com/server/jira/platform/jira-rest-api-examples/#creating-an-issue-examples45        # https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issues/46        jira_issue = jira_client.create_issue(fields=createDict)47    except JIRAError as err:48        send_task_failure(49            task_token, f"Unable to create JIRA issue {str(err.text)}")50        return False51    if not jira_issue.key:52        send_task_failure(task_token, "Couldn't create new JIRA issue")53        return False54    else:55        logging.info("Created jira issue %s", jira_issue.key)56        return_dict = {"issue_key": jira_issue.key}57        # either you are calling the lambda directly 58        # or calling from AWS step functions with task token59        if task_token is None:60            logging.info("SUCCESS")61            return return_dict62        else:63            sf_client.send_task_success(taskToken=task_token,64                                        output=json.dumps(return_dict))65            return True66def add_comments(event, task_token, jira_client):67    ticket_key = event.get('issue_key')68    comment = json.dumps(event.get('issue_comment'), indent=4)69    if not ticket_key:70        send_task_failure(71            task_token, "issue_key was not passed in input json")72        return False73    if not comment:74        send_task_failure(75            task_token, "issue_comment not passed in input json")76        return False77    try:78        jira_client.add_comment(ticket_key, comment)79    except JIRAError as err:80        send_task_failure(task_token, ticket_key +81                            str(err.text) + str(comment))82        return False83    # either you are calling the lambda directly 84    # or calling from AWS step functions with task token85    if task_token is None:86        logging.info("SUCCESS")87        return False88    else:89        sf_client.send_task_success(taskToken=task_token,90                                    output='{"Success" : "true"}')91        return True92# either we are creating a new ticket or adding comments (op_type)93def lambda_handler(event, context):94    task_token = event.get("TaskToken")95    op_type = event.get("op_type")96    if op_type is None:97        send_task_failure(task_token, "op_type is missing")98        return False99    logging.info("Obtaining jira creds from secrets manager...")100    try:101        sm_response = sm_client.get_secret_value(102            SecretId=os.environ.get('secret', 'jira-creds-admintool'))103    except ClientError as e:104        error_msg = (105            f"Could not get jira creds from secretsmanager.",106            f"{str(e.response['Error']['Code'])}"107        )108        send_task_failure(task_token, error_msg)109        return False110    try:111        user_secret = json.loads(sm_response['SecretString'])['user']112    except (KeyError, json.decoder.JSONDecodeError):113        send_task_failure(task_token, f"Error loading user SecretString")114        return False115    try:116        pass_secret = json.loads(sm_response['SecretString'])['pass']117    except (KeyError, json.decoder.JSONDecodeError):118        send_task_failure(task_token, f"Error loading pass SecretString")119        return False120    if (user_secret is None or pass_secret is None):121        send_task_failure(task_token, "SecretString is missing")122        return False123    logging.info("Done. Connecting to %s", JIRA_SERVER)124    try:125        jira_client = JIRA(server=JIRA_SERVER,126                           basic_auth=(user_secret, pass_secret))127    except JIRAError as err:128        send_task_failure(task_token, f"unable to connect to JIRA {err.text}")129        return False130    if not jira_client:131        send_task_failure(task_token, "unable to connect to JIRA ")132        return False133    else:134        logging.info("Connected to %s", JIRA_SERVER)135    if(op_type == "create"):136        return create_jira_ticket(event, task_token, jira_client)137    elif(op_type == "add_comments"):138        return add_comments(event, task_token, jira_client)139    else:140        logging.error("""op_type not 'create' or 'add_comments'.141            No clue what you are trying to do""")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.