""" # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: MIT-0 # Start of unit test code: tests/unit/src/test_sample_lambda.py """ import sys import os import json from unittest import TestCase from unittest.mock import MagicMock, patch from boto3 import resource, client import moto from aws_lambda_powertools.utilities.validation import validate # [0] Import the Globals, Classes, Schemas, and Functions from the Lambda Handler sys.path.append('./src/sample_lambda') from src.sample_lambda.app import LambdaDynamoDBClass, LambdaS3Class # pylint: disable=wrong-import-position from src.sample_lambda.app import lambda_handler, create_letter_in_s3 # pylint: disable=wrong-import-position from src.sample_lambda.schemas import INPUT_SCHEMA # pylint: disable=wrong-import-position # [1] Mock all AWS Services in use @moto.mock_dynamodb @moto.mock_s3 class TestSampleLambda(TestCase): """ Test class for the application sample AWS Lambda Function """ # Test Setup def setUp(self) -> None: """ Create mocked resources for use during tests """ # [2] Mock environment & override resources self.test_ddb_table_name = "unit_test_ddb" self.test_s3_bucket_name = "unit_test_s3_bucket" os.environ["DYNAMODB_TABLE_NAME"] = self.test_ddb_table_name os.environ["S3_BUCKET_NAME"] = self.test_s3_bucket_name # [3a] Set up the services: construct a (mocked!) DynamoDB table dynamodb = resource("dynamodb", region_name="us-east-1") dynamodb.create_table( TableName = self.test_ddb_table_name, KeySchema=[{"AttributeName": "PK", "KeyType": "HASH"}], AttributeDefinitions=[{"AttributeName": "PK", "AttributeType": "S"}], BillingMode='PAY_PER_REQUEST' ) # [3b] Set up the services: construct a (mocked!) S3 Bucket table s3_client = client('s3', region_name="us-east-1") s3_client.create_bucket(Bucket = self.test_s3_bucket_name ) # [4] Establish the "GLOBAL" environment for use in tests. mocked_dynamodb_resource = resource("dynamodb") mocked_s3_resource = resource("s3") mocked_dynamodb_resource = { "resource" : resource('dynamodb'), "table_name" : self.test_ddb_table_name } mocked_s3_resource = { "resource" : resource('s3'), "bucket_name" : self.test_s3_bucket_name } self.mocked_dynamodb_class = LambdaDynamoDBClass(mocked_dynamodb_resource) self.mocked_s3_class = LambdaS3Class(mocked_s3_resource) def test_create_letter_in_s3(self) -> None: """ Verify given correct parameters, the document will be written to S3 with proper contents. """ # [5] Post test items to a mocked database self.mocked_dynamodb_class.table.put_item(Item={"PK":"D#UnitTestDoc", "data":"Unit Test Doc Corpi"}) self.mocked_dynamodb_class.table.put_item(Item={"PK":"C#UnitTestCust", "data":"Unit Test Customer"}) # [6] Run DynamoDB to S3 file function test_return_value = create_letter_in_s3( dynamo_db = self.mocked_dynamodb_class, s3=self.mocked_s3_class, doc_type = "UnitTestDoc", cust_id = "UnitTestCust" ) # [7] Ensure the data was written to S3 correctly, with correct contents bucket_key = "UnitTestCust/UnitTestDoc.txt" body = self.mocked_s3_class.bucket.Object(bucket_key).get()['Body'].read() # Test self.assertEqual(test_return_value["statusCode"], 200) self.assertIn("UnitTestCust/UnitTestDoc.txt", test_return_value["body"]) self.assertEqual(body.decode('ascii'),"Dear Unit Test Customer;\nUnit Test Doc Corpi") def test_create_letter_in_s3_doc_type_notfound_404(self) -> None: """ Verify given a document type not present in the data table, a 404 error is returned. """ # [8] Post test items to a mocked database self.mocked_dynamodb_class.table.put_item(Item={"PK":"D#UnitTestDoc", "data":"Unit Test Doc Corpi"}) self.mocked_dynamodb_class.table.put_item(Item={"PK":"C#UnitTestCust", "data":"Unit Test Customer"}) # [9] Run DynamoDB to S3 file function test_return_value = create_letter_in_s3( dynamo_db = self.mocked_dynamodb_class, s3=self.mocked_s3_class, doc_type = "NOTVALID", cust_id = "UnitTestCust" ) # Test self.assertEqual(test_return_value["statusCode"], 404) self.assertIn("Not Found", test_return_value["body"]) def test_create_letter_in_s3_customer_notfound_404(self) -> None: """ Verify given a user id not present in the data table, a 404 error is returned. """ # [10] Post test items to a mocked database self.mocked_dynamodb_class.table.put_item(Item={"PK":"D#UnitTestDoc", "data":"Unit Test Doc Corpi"}) self.mocked_dynamodb_class.table.put_item(Item={"PK":"C#UnitTestCust", "data":"Unit Test Customer"}) # [11] Run DynamoDB to S3 file function test_return_value = create_letter_in_s3( dynamo_db = self.mocked_dynamodb_class, s3=self.mocked_s3_class, doc_type = "UnitTestDoc", cust_id = "NOTVALID" ) # Test self.assertEqual(test_return_value["statusCode"], 404) self.assertIn("Not Found", test_return_value["body"]) # [12] Load and validate test events from the file system def load_sample_event_from_file(self, test_event_file_name: str) -> dict: """ Loads and validate test events from the file system """ event_file_name = f"tests/events/{test_event_file_name}.json" with open(event_file_name, "r", encoding='UTF-8') as file_handle: event = json.load(file_handle) validate(event=event, schema=INPUT_SCHEMA) return event # [13] Patch the Global Class and any function calls @patch("src.sample_lambda.app.LambdaDynamoDBClass") @patch("src.sample_lambda.app.LambdaS3Class") @patch("src.sample_lambda.app.create_letter_in_s3") def test_lambda_handler_valid_event_returns_200(self, patch_create_letter_in_s3 : MagicMock, patch_lambda_s3_class : MagicMock, patch_lambda_dynamodb_class : MagicMock ): """ Verify the event is parsed, AWS resources are passed, the create_letter_in_s3 function is called, and a 200 is returned. """ # [14] Test setup - Return a mock for the global variables and resources patch_lambda_dynamodb_class.return_value = self.mocked_dynamodb_class patch_lambda_s3_class.return_value = self.mocked_s3_class return_value_200 = {"statusCode" : 200, "body":"OK"} patch_create_letter_in_s3.return_value = return_value_200 # [15] Run Test using a test event from /tests/events/*.json test_event = self.load_sample_event_from_file("sampleEvent1") test_return_value = lambda_handler(event=test_event, context=None) # [16] Validate the function was called with the mocked globals # and event values patch_create_letter_in_s3.assert_called_once_with( dynamo_db=self.mocked_dynamodb_class, s3=self.mocked_s3_class, doc_type=test_event["pathParameters"]["docType"], cust_id=test_event["pathParameters"]["customerId"]) self.assertEqual(test_return_value, return_value_200) def tearDown(self) -> None: # [13] Remove (mocked!) S3 Objects and Bucket s3_resource = resource("s3",region_name="us-east-1") s3_bucket = s3_resource.Bucket( self.test_s3_bucket_name ) for key in s3_bucket.objects.all(): key.delete() s3_bucket.delete() # [14] Remove (mocked!) DynamoDB Table dynamodb_resource = client("dynamodb", region_name="us-east-1") dynamodb_resource.delete_table(TableName = self.test_ddb_table_name ) # End of unit test code