BLOG
이 블로그 포스트에서는 AWSCodeStar프로젝트의 일부로 유닛 테스트를 수행하는 방법을 보여 줍니다. AWS CodeStar는 AWS에서 애플리케이션을 빠르게 개발, 구축 및 배포할 수 있도록 도와 줄 것 입니다. AWS CodeStar를 사용하여 연속 전송(CD)도구 체인을 설정하고 한곳에서 소프트웨어 개발을 관리할 수 있습니다.
유닛 테스트는 개별 유닛의 애플리케이션 코드를 테스트하기 때문에 문제를 신속하게 식별하고 격리하는 데 도움이 됩니다. 자동화된 CI/CD프로세스의 일환으로 잘못된 코드가 프로덕션에 배포되지 않도록 하는데 사용될 수도 있습니다.
AWS CodeStar프로젝트 템플릿은 대부분 유닛 테스트 프레임워크와 함께 사전 구성되어 제공되므로, 자신 있게 코드를 배포하시면 됩니다. 유닛 테스트는 제공된 구축 단계에서 실행되어 유닛 테스트가 통과하지 못할 경우 코드가 배포되지 않도록 구성되어 있습니다. 유닛 테스트를 포함한 AWS CodeStar프로젝트 템플릿 목록은 AWS CodeStar사용자 가이드의 AWS CodeStar프로젝트 템플릿을 참조하십시오.
시나리오
슈퍼 히어로 무비의 광팬으로서 저는 제가 만든 웹 서비스 엔드 포인트에서 사용해서 제가 가장 좋아하는 영화 목록을 만들어 제 친구들에게 그들의 영화에 투표해 달라고 부탁하기로 했죠. 예를 들어 AWSCodeCommit을 사용하는 AWS람다에서 코드 저장소로 실행되는 Python 웹 서비스를 사용합니다. CodeCommit은 Git 저장소를 호스팅하고 모든 슬롯 기반 도구와 함께 작동하는 완벽하게 관리되는 소스 제어 시스템입니다.
웹 서비스 엔드포인트를 생성하는 방법은 다음과 같습니다.
AWS CodeStar콘솔에 로그인합니다. 프로젝트 템플릿 목록으로 이동할 프로젝트 시작을 선택하십시오.
코드의 수정을 위해서 AWSCloud9를 선택하는데 이는 사용자가 코드를 쓰고, 실행하고, 디버그 하는 데 이용되는 클라우드 기반 IDE입니다.
제 시나리오에 필요한 다른 작업은 다음과 같습니다.
* 필요에 따라 표를 저장하고 검색할 수 있는 데이터베이스 테이블을 만듭니다.
* 투표를 게시하고 가져오기 위해 만든 람다 기능의 로직을 업데이트합니다.
* 유닛 테스트를 업데이트하여 논리가 예상대로 작동하는지 확인합니다.
데이터베이스 테이블 용으로는 빠르고 유연한 NoSQL 데이터베이스를 제공하는 Amazon DynamoDB를 선택했습니다.
AWSCloud9에 설정
AWS CodeStar콘솔에서 AWSCloud9 콘솔로 이동하여 프로젝트 코드로 이동합니다. 환경과 필요한 라이브러리를 설정할 최상위 폴더에 터미널을 엽니다.
다음 명령을 사용해 유닛에 PATHONPATH 환경 변수를 설정합니다.
export PYTHONPATH=/home/ec2–user/environment/vote–your–movie
프로젝트에서 유닛 테스트를 실행하기 위하여 다음 명령을 사용하면 됩니다.
python –m unittest discover vote–your–movie/tests
코딩 시작하기
로컬 환경을 설정하고 코드 복사본을 가지고 있으므로 템플릿 파일을 통해 프로젝트를 정의하여 프로젝트에 DynamoDB테이블을 추가합니다. template.yml이라는 서버리스 응용 모델(SAM)템플릿 파일을 여세요. 이 템플릿은 AWS CloudFormation을 확장하여 서버리스 애플리케이션에 필요한 AmazonAPI Gateway API, AWS람다 기능 및 Amazon DynamoDB테이블을 정의하는 간단한 방법을 제공합니다.
AWSTemplateFormatVersion: 2010-09-09
Transform:
– AWS::Serverless-2016-10-31
– AWS::CodeStar
Parameters:
ProjectId:
Type: String
Description: CodeStar projectId used to associate new resources to team members
Resources:
# The DB table to store the votes.
MovieVoteTable:
Type: AWS::Serverless::SimpleTable
Properties:
PrimaryKey:
# Name of the “Candidate” is the partition key of the table.
Name: Candidate
Type: String
# Creating a new lambda function for retrieving and storing votes.
MovieVoteLambda:
Type: AWS::Serverless::Function
Properties:
Handler: index.handler
Runtime: python3.6
Environment:
# Setting environment variables for your lambda function.
Variables:
TABLE_NAME: !Ref “MovieVoteTable”
TABLE_REGION: !Ref “AWS::Region”
Role:
Fn::ImportValue:
!Join [‘-‘, [!Ref ‘ProjectId’, !Ref ‘AWS::Region’, ‘LambdaTrustRole’]]
Events:
GetEvent:
Type: Api
Properties:
Path: /
Method: get
PostEvent:
Type: Api
Properties:
Path: /
Method: post
이제 우리는 Python의 boto3 라이브러리를 사용하여 AWS 서비스에 연결할 것이고 Pythons의 모크 라이브러리를 사용해서 유닛테스트의 AWS 서비스 콜을 진행합니다. 다음 명령을 사용해주세요.
pip install —upgrade boto3 mock –t .
이러한 라이브러리를 CodeBuild를 실행하는 데 필요한 YAML파일인 buildspec.yml에 추가해야 합니다.
version: 0.2
phases:
install:
commands:
# Upgrade AWS CLI to the latest version
– pip install –upgrade awscli boto3 mock
pre_build:
commands:
# Discover and run unit tests in the ‘tests’ directory. For more information, see <https://docs.python.org/3/library/unittest.html#test-discovery>
– python -m unittest discover tests
build:
commands:
# Use AWS SAM to package the application by using AWS CloudFormation
– aws cloudformation package –template template.yml –s3-bucket $S3_BUCKET –output-template template-export.yml
artifacts:
type: zip
files:
– template-export.yml
람다 기능에 간단한 투표 로직을 쓸 수 있는 index.py 를 오픈합니다.
import json
import datetime
import boto3
import os
table_name = os.environ[‘TABLE_NAME’]
table_region = os.environ[‘TABLE_REGION’]
VOTES_TABLE = boto3.resource(‘dynamodb’, region_name=table_region).Table(table_name)
CANDIDATES = {“A”: “Black Panther”, “B”: “Captain America: Civil War”, “C”: “Guardians of the Galaxy”, “D”: “Thor: Ragnarok”}
def handler(event, context):
if event[‘httpMethod’] == ‘GET’:
resp = VOTES_TABLE.scan()
return {‘statusCode’: 200,
‘body’: json.dumps({item[‘Candidate’]: int(item[‘Votes’]) for item in resp[‘Items’]}),
‘headers’: {‘Content-Type’: ‘application/json’}}
elif event[‘httpMethod’] == ‘POST’:
try:
body = json.loads(event[‘body’])
except:
return {‘statusCode’: 400,
‘body’: ‘Invalid input! Expecting a JSON.’,
‘headers’: {‘Content-Type’: ‘application/json’}}
if ‘candidate’ not in body:
return {‘statusCode’: 400,
‘body’: ‘Missing “candidate” in request.’,
‘headers’: {‘Content-Type’: ‘application/json’}}
if body[‘candidate’] not in CANDIDATES.keys():
return {‘statusCode’: 400,
‘body’: ‘You must vote for one of the following candidates – {}.’.format(get_allowed_candidates()),
‘headers’: {‘Content-Type’: ‘application/json’}}
resp = VOTES_TABLE.update_item(
Key={‘Candidate’: CANDIDATES.get(body[‘candidate’])},
UpdateExpression=’ADD Votes :incr’,
ExpressionAttributeValues={‘:incr’: 1},
ReturnValues=’ALL_NEW’
)
return {‘statusCode’: 200,
‘body’: “{} now has {} votes”.format(CANDIDATES.get(body[‘candidate’]), resp[‘Attributes’][‘Votes’]),
‘headers’: {‘Content-Type’: ‘application/json’}}
def get_allowed_candidates():
l = []
for key in CANDIDATES:
l.append(“‘{}’ for ‘{}'”.format(key, CANDIDATES.get(key)))
return “, “.join(l)
기본적으로 저희 코드는 HTTPS 요청을 하나의 이벤트로 받아들이는 것입니다. HTTP GET 요청인 경우 테이블에서 결과를 가져옵니다. HTTP POST 요청인 경우 선택한 후보자에게 투표하도록 설정합니다. POST요청의 입력 내용도 확인하여 악성으로 보이는 요청을 필터링 합니다. 그러면 유효한 전화만 테이블에 저장됩니다.
제공된 예제 코드에서는 후보를 저장하기 위해 CANDIDATES 변수를 사용하지만, 후보를 JSON파일에 저장한 다음 대신 Python’s json 라이브러리를 사용할 수 있습니다.
이제 테스트를 업데이트하겠습니다. 테스트 폴더에서 test_handler.py을 열고 로직을 확인하도록 수정합니다.
import os
# Some mock environment variables that would be used by the mock for DynamoDB
os.environ[‘TABLE_NAME’] = “MockHelloWorldTable”
os.environ[‘TABLE_REGION’] = “us-east-1”
# The library containing our logic.
import index
# Boto3’s core library
import botocore
# For handling JSON.
import json
# Unit test library
import unittest
## Getting StringIO based on your setup.
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
## Python mock library
from mock import patch, call
from decimal import Decimal
@patch(‘botocore.client.BaseClient._make_api_call’)
class TestCandidateVotes(unittest.TestCase):
## Test the HTTP GET request flow.
## We expect to get back a successful response with results of votes from the table (mocked).
def test_get_votes(self, boto_mock):
# Input event to our method to test.
expected_event = {‘httpMethod’: ‘GET’}
# The mocked values in our DynamoDB table.
items_in_db = [{‘Candidate’: ‘Black Panther’, ‘Votes’: Decimal(‘3’)},
{‘Candidate’: ‘Captain America: Civil War’, ‘Votes’: Decimal(‘8’)},
{‘Candidate’: ‘Guardians of the Galaxy’, ‘Votes’: Decimal(‘8’)},
{‘Candidate’: “Thor: Ragnarok”, ‘Votes’: Decimal(‘1’)}
]
# The mocked DynamoDB response.
expected_ddb_response = {‘Items’: items_in_db}
# The mocked response we expect back by calling DynamoDB through boto.
response_body = botocore.response.StreamingBody(StringIO(str(expected_ddb_response)),
len(str(expected_ddb_response)))
# Setting the expected value in the mock.
boto_mock.side_effect = [expected_ddb_response]
# Expecting that there would be a call to DynamoDB Scan function during execution with these parameters.
expected_calls = [call(‘Scan’, {‘TableName’: os.environ[‘TABLE_NAME’]})]
# Call the function to test.
result = index.handler(expected_event, {})
# Run unit test assertions to verify the expected calls to mock have occurred and verify the response.
assert result.get(‘headers’).get(‘Content-Type’) == ‘application/json’
assert result.get(‘statusCode’) == 200
result_body = json.loads(result.get(‘body’))
# Verifying that the results match to that from the table.
assert len(result_body) == len(items_in_db)
for i in range(len(result_body)):
assert result_body.get(items_in_db[i].get(“Candidate”)) == int(items_in_db[i].get(“Votes”))
assert boto_mock.call_count == 1
boto_mock.assert_has_calls(expected_calls)
## Test the HTTP POST request flow that places a vote for a selected candidate.
## We expect to get back a successful response with a confirmation message.
def test_place_valid_candidate_vote(self, boto_mock):
# Input event to our method to test.
expected_event = {‘httpMethod’: ‘POST’, ‘body’: “{\”candidate\”: \”D\”}”}
# The mocked response in our DynamoDB table.
expected_ddb_response = {‘Attributes’: {‘Candidate’: “Thor: Ragnarok”, ‘Votes’: Decimal(‘2’)}}
# The mocked response we expect back by calling DynamoDB through boto.
response_body = botocore.response.StreamingBody(StringIO(str(expected_ddb_response)),
len(str(expected_ddb_response)))
# Setting the expected value in the mock.
boto_mock.side_effect = [expected_ddb_response]
# Expecting that there would be a call to DynamoDB UpdateItem function during execution with these parameters.
expected_calls = [call(‘UpdateItem’, {
‘TableName’: os.environ[‘TABLE_NAME’],
‘Key’: {‘Candidate’: ‘Thor: Ragnarok’},
‘UpdateExpression’: ‘ADD Votes :incr’,
‘ExpressionAttributeValues’: {‘:incr’: 1},
‘ReturnValues’: ‘ALL_NEW’
})]
# Call the function to test.
result = index.handler(expected_event, {})
# Run unit test assertions to verify the expected calls to mock have occurred and verify the response.
assert result.get(‘headers’).get(‘Content-Type’) == ‘application/json’
assert result.get(‘statusCode’) == 200
assert result.get(‘body’) == “{} now has {} votes”.format(
expected_ddb_response[‘Attributes’][‘Candidate’],
expected_ddb_response[‘Attributes’][‘Votes’])
assert boto_mock.call_count == 1
boto_mock.assert_has_calls(expected_calls)
## Test the HTTP POST request flow that places a vote for an non-existant candidate.
## We expect to get back a successful response with a confirmation message.
def test_place_invalid_candidate_vote(self, boto_mock):
# Input event to our method to test.
# The valid IDs for the candidates are A, B, C, and D
expected_event = {‘httpMethod’: ‘POST’, ‘body’: “{\”candidate\”: \”E\”}”}
# Call the function to test.
result = index.handler(expected_event, {})
# Run unit test assertions to verify the expected calls to mock have occurred and verify the response.
assert result.get(‘headers’).get(‘Content-Type’) == ‘application/json’
assert result.get(‘statusCode’) == 400
assert result.get(‘body’) == ‘You must vote for one of the following candidates – {}.’.format(index.get_allowed_candidates())
## Test the HTTP POST request flow that places a vote for a selected candidate but associated with an invalid key in the POST body.
## We expect to get back a failed (400) response with an appropriate error message.
def test_place_invalid_data_vote(self, boto_mock):
# Input event to our method to test.
# “name” is not the expected input key.
expected_event = {‘httpMethod’: ‘POST’, ‘body’: “{\”name\”: \”D\”}”}
# Call the function to test.
result = index.handler(expected_event, {})
# Run unit test assertions to verify the expected calls to mock have occurred and verify the response.
assert result.get(‘headers’).get(‘Content-Type’) == ‘application/json’
assert result.get(‘statusCode’) == 400
assert result.get(‘body’) == ‘Missing “candidate” in request.’
## Test the HTTP POST request flow that places a vote for a selected candidate but not as a JSON string which the body of the request expects.
## We expect to get back a failed (400) response with an appropriate error message.
def test_place_malformed_json_vote(self, boto_mock):
# Input event to our method to test.
# “body” receives a string rather than a JSON string.
expected_event = {‘httpMethod’: ‘POST’, ‘body’: “Thor: Ragnarok”}
# Call the function to test.
result = index.handler(expected_event, {})
# Run unit test assertions to verify the expected calls to mock have occurred and verify the response.
assert result.get(‘headers’).get(‘Content-Type’) == ‘application/json’
assert result.get(‘statusCode’) == 400
assert result.get(‘body’) == ‘Invalid input! Expecting a JSON.’
if __name__ == ‘__main__’:
unittest.main()
각 단위 시험이 무엇을 달성하는지 명확하게 하기 위해 코드 샘플을 잘 유지하고 있습니다. 이는 논리적으로 처리되는 성공 조건과 실패 경로를 테스트합니다.
유닛 테스트에서는 모크 라이브러리에 있는 패치 스타일리스트(@patch)를 사용합니다.@patch는 호출하려는(이 경우 botocore Library의_BaseClient class 에서의 make_api_call 기능)기능을 모의 실험하는 데 도움이 됩니다.
변화를 주기 전에, 그 시험을 국지적으로 실시하고 터미널에서 테스트를 다시 실행합니다. 모든 장치 테스트에 통과하면 다음과 같은 결과를 기대할 수 있습니다.
You:~/environment $ python -m unittest discover vote-your-movie/tests
…..
———————————————————————-
Ran 5 tests in 0.003s
OK
You:~/environment $
AWS 에 업로드
테스트에 성공했으니 이제 소스 저장소에 코드를 커밋하고 푸시 할 시간입니다.
변경 사항 추가
터미널에서 프로젝트 폴더로 이동하여 다음 명령을 사용하여 푸시 할 변경 내용을 확인하십시오.
git status
수정된 파일만 추가하려면 다음 명령을 사용합니다.
git add –u
Commit your changes
변경사항에 커밋하기 위해 다음의 명령을 사용하세요.
git commit –m “Logic and tests for the voting webservice.”
AWS CodeCommit 에 변경사항을 푸쉬하기
CodeCommit 에 변경사항을 푸쉬하게 위하여 다음을 사용하세요.
git push
AWS CodeStar콘솔에서는 변경 사항이 파이프라인을 통과하여 배포되는 것을 확인할 수 있습니다. AWS CodeBuild에서 실행되는 테스트를 확인할 수 있도록 이 프로젝트의 빌드 실행으로 이동하는 AWS CodeStar콘솔에도 링크가 있습니다. 빌드 실행 테이블 아래의 최신 링크를 사용하여 로그로 이동할 수 있습니다.
배포가 완료되면 AWS CodeStar에 AWS람다 기능과 이 프로젝트와 동기화된 DynamoDB표가 표시됩니다. AWS CodeStar프로젝트의 탐색 모음에 있는 프로젝트 링크는 이 프로젝트에 연결된 AWS 리소스를 표시합니다.
이것은 새로운 데이터베이스 테이블이기 때문에 데이터가 들어 있지 않을 것입니다. 그렇다면, 투표를 해볼 수 있습니다. Postman을 다운로드하여 POST및 GET 호출을 위한 응용 프로그램 엔드 포인트를 테스트할 수 있습니다. 테스트할 엔드 포인트는 AWSCodeStar콘솔의 애플리케이션 엔드포인트 아래에 표시되는 URL입니다.
이제 Postman를 열고 결과를 살펴봅시다. POST요청을 통해 투표를 만들어 봅시다. 이 예에 기초하여 유효한 투표는 A, B, C또는 D의 값을 갖습니다.
POST 요청의 좋은 예는 다음과 같습니다.
A,B,C, 혹은 D 이외의 값을 사용하면 다음과 같이 보입니다.
이제 GET요청을 사용하여 데이터베이스에서 투표 결과를 가져오겠습니다.
AWS람다, Amazon APIGateway, DynamoDB를 사용하여 간단한 투표용 웹 서비스를 만들어 사용하여 논리를 확인함으로써 좋은 코드를 전송할 수 있습니다.
원문 URL: https://aws.amazon.com/ko/blogs/devops/performing-unit-testing-in-an-aws-codestar-project/
** 메가존 TechBlog는 AWS BLOG 영문 게재글중에서 한국 사용자들에게 유용한 정보 및 콘텐츠를 우선적으로 번역하여 내부 엔지니어 검수를 받아서, 정기적으로 게재하고 있습니다. 추가로 번역및 게재를 희망하는 글에 대해서 관리자에게 메일 또는 SNS페이지에 댓글을 남겨주시면, 우선적으로 번역해서 전달해드리도록 하겠습니다.