service test by lambda

lambda_function.py

import json
import os
import unittest
from unittest import TestLoader
from unittest import TextTestRunner
import requests
import boto3
sns = boto3.resource("sns")
sns_topic = sns.Topic(os.environ["result_sns_topic_arn"])

def lambda_handler(event, context):
    # run test
    loader = TestLoader()
    test = loader.discover("./tests", pattern="*.py")
    runner = TextTestRunner()
    r = runner.run(test)
    
    # report
    summary = f"Finished test ran:{r.testsRun} failed:{len(r.failures)}"
    body = f"request_id:{context.aws_request_id}\nfailures = {r.failures}"
    print(f"{summary}\n{body}")
    sns_topic.publish(Subject=summary,Message=body)
    
    return "ok"

tests_acceptance.py

import unittest
import boto3
import os
import requests
import time

ses = boto3.client('ses')
WAIT_SECONDS = 2
TARGET_EMAIL = os.environ['target_email']
SENDER_EMAIL = os.environ['ok_sender_email']
NOP_SENDER_EMAIL = os.environ['nop_sender_email']
KEYWORD = os.environ['keyword']
BACKLOG_ENDPOINT = f'https://{os.environ["backlog_host"]}'
BACKLOG_STATUS_NOTSTARTED = 1
BACKLOG_STATUS_COMPLETED = 4

# Test Cases
class AcceptanceTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        """
        テストクラスsetUp
        テスト起票対象を全てcompleteさせておく(前準備)
        """
        complete_all_test_issues()
    
    def tearDown(self):
        """
        テストケースtearDown
        テスト起票対象を全てcompleteさせておく(後片付け
        """
        complete_all_test_issues()
    
    def test_target_keyword(self):
        """
        キーワードそのままsubjectで送信 -> 起票されること
        """
        # input
        subject = KEYWORD
        send_email(src=SENDER_EMAIL, subject=subject)
        
        # wait
        time.sleep(WAIT_SECONDS)
        
        # assert
        issues = get_test_issues()
        self.assertEqual(1, len(issues))
        self.assertEqual(subject, issues[0]['summary'])

    def test_target_include_keyword(self):
        """
        キーワードを含むsubjectで送信-> 起票されること
        """
        # input
        subject = f'xxx{KEYWORD}yyy'
        send_email(src=SENDER_EMAIL, subject=subject)
        
        # wait
        time.sleep(WAIT_SECONDS)
        
        # assert
        issues = get_test_issues()
        self.assertEqual(1, len(issues))
        self.assertEqual(subject, issues[0]['summary'])
    
    def test_not_target_not_include_keyword(self):
        """
        キーワードを含まないsubjectで送信 -> 起票されないこと
        """
        # input
        subject = KEYWORD
        send_email(src=SENDER_EMAIL, subject=f"xxxyyyy")
        
        # wait
        time.sleep(WAIT_SECONDS)
        
        # assert
        issues = get_test_issues()
        self.assertEqual(0, len(issues))

    def test_not_target_invalid_sender(self):
        """
        キーワードそのままsubject 対象外のアドレスから送信 -> 起票されないこと
        """
        # input
        subject = KEYWORD
        send_email(src=NOP_SENDER_EMAIL, subject=subject)
        
        # wait
        time.sleep(WAIT_SECONDS)
        
        # assert
        issues = get_test_issues()
        self.assertEqual(0, len(issues))


# Utility
def send_email(src, subject):
    """
    メール送信
    """
    ses.send_email(
        Source=src,
        Destination={"ToAddresses": [TARGET_EMAIL]},
        Message={
            'Subject': {'Data': subject,'Charset': 'UTF-8'},
            'Body': {'Text': {'Data': 'test','Charset': 'UTF-8'}}
        }
    )

def get_test_issues():
    """
    テスト起票対象の課題取得
        ・対象の課題種別
        ・未着手
    """
    res = requests.get(
        BACKLOG_ENDPOINT + "/api/v2/issues", 
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        params={
            'apiKey':os.environ['backlog_api_key'],
            'projectId[]': [int(os.environ['backlog_projectId'])],
            'issueTypeId[]': [int(os.environ['backlog_target_issueType'])],
            'statusId[]': [BACKLOG_STATUS_NOTSTARTED]
        }
    )
    return res.json()

def complete_all_test_issues():
    """
    全てのテスト起票対象課題を完了にする
    """
    issues = get_test_issues()
    for issue in issues:
        complete_issue(issue)

def complete_issue(issue):
    """
    課題を完了にする
    """
    res = requests.patch(
        BACKLOG_ENDPOINT + "/api/v2/issues/" +  issue['issueKey'],
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        params={
            'apiKey':os.environ['backlog_api_key'],
            'statusId': BACKLOG_STATUS_COMPLETED
        }
    )
    return res.json()