tests(pydantic): Add missing coverage

Changes:
* Make some methods static
* Run code formatter
* Some lint based changes
* Add missing test cases

Author: michaelbrewer

aws_lambda_powertools/utilities/advanced_parser/envelopes/base.py

@@ -8,7 +8,8 @@
 
 
 class BaseEnvelope(ABC):
-    def _parse_user_dict_schema(self, user_event: Dict[str, Any], schema: BaseModel) -> Any:
+    @staticmethod
+    def _parse_user_dict_schema(user_event: Dict[str, Any], schema: BaseModel) -> Any:
         if user_event is None:
             return None
         logger.debug("parsing user dictionary schema")
@@ -18,7 +19,8 @@ def _parse_user_dict_schema(self, user_event: Dict[str, Any], schema: BaseModel)
             logger.exception("Validation exception while extracting user custom schema")
             raise
 
-    def _parse_user_json_string_schema(self, user_event: str, schema: BaseModel) -> Any:
+    @staticmethod
+    def _parse_user_json_string_schema(user_event: str, schema: BaseModel) -> Any:
         if user_event is None:
             return None
         # this is used in cases where the underlying schema is not a Dict that can be parsed as baseModel

aws_lambda_powertools/utilities/advanced_parser/envelopes/dynamodb.py

@@ -1,8 +1,8 @@
 import logging
 from typing import Any, Dict, List
-from typing_extensions import Literal
 
 from pydantic import BaseModel, ValidationError
+from typing_extensions import Literal
 
 from aws_lambda_powertools.utilities.advanced_parser.envelopes.base import BaseEnvelope
 from aws_lambda_powertools.utilities.advanced_parser.schemas import DynamoDBSchema

aws_lambda_powertools/utilities/advanced_parser/schemas/dynamodb.py

@@ -18,9 +18,9 @@ class DynamoScheme(BaseModel):
     # exist in a legal schema of NEW_AND_OLD_IMAGES type
     @root_validator
     def check_one_image_exists(cls, values):
-        newimg, oldimg = values.get("NewImage"), values.get("OldImage")
+        new_img, old_img = values.get("NewImage"), values.get("OldImage")
         stream_type = values.get("StreamViewType")
-        if stream_type == "NEW_AND_OLD_IMAGES" and not newimg and not oldimg:
+        if stream_type == "NEW_AND_OLD_IMAGES" and not new_img and not old_img:
             raise TypeError("DynamoDB streams schema failed validation, missing both new & old stream images")
         return values
 

aws_lambda_powertools/utilities/advanced_parser/schemas/sqs.py

@@ -34,16 +34,16 @@ def valid_type(cls, v):  # noqa: VNE001
             raise TypeError("data type is invalid")
         return v
 
-    # validate that dataType and value are not None and match 
+    # validate that dataType and value are not None and match
     @root_validator
     def check_str_and_binary_values(cls, values):
         binary_val, str_val = values.get("binaryValue", ""), values.get("stringValue", "")
-        dataType = values.get("dataType")
+        data_type = values.get("dataType")
         if not str_val and not binary_val:
             raise TypeError("both binaryValue and stringValue are missing")
-        if dataType.startswith("Binary") and not binary_val:
+        if data_type.startswith("Binary") and not binary_val:
             raise TypeError("binaryValue is missing")
-        if (dataType.startswith("String") or dataType.startswith("Number")) and not str_val:
+        if (data_type.startswith("String") or data_type.startswith("Number")) and not str_val:
             raise TypeError("stringValue is missing")
         return values
 

tests/functional/parser/test_dynamodb.py

@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Any, Dict, List
 
 import pytest
 from pydantic.error_wrappers import ValidationError
@@ -11,7 +11,7 @@
 
 
 @parser(schema=MyDynamoBusiness, envelope=Envelope.DYNAMODB_STREAM)
-def handle_dynamodb(event: List[Dict[str, MyDynamoBusiness]], context: LambdaContext):
+def handle_dynamodb(event: List[Dict[str, MyDynamoBusiness]], _: LambdaContext):
     assert len(event) == 2
     assert event[0]["OldImage"] is None
     assert event[0]["NewImage"].Message["S"] == "New item!"
@@ -23,7 +23,7 @@ def handle_dynamodb(event: List[Dict[str, MyDynamoBusiness]], context: LambdaCon
 
 
 @parser(schema=MyAdvancedDynamoBusiness)
-def handle_dynamodb_no_envelope(event: MyAdvancedDynamoBusiness, context: LambdaContext):
+def handle_dynamodb_no_envelope(event: MyAdvancedDynamoBusiness, _: LambdaContext):
     records = event.Records
     record = records[0]
     assert record.awsRegion == "us-west-2"
@@ -60,12 +60,40 @@ def test_dynamo_db_stream_trigger_event_no_envelope():
 
 
 def test_validate_event_does_not_conform_with_schema_no_envelope():
-    event_dict = {"hello": "s"}
+    event_dict: Any = {"hello": "s"}
     with pytest.raises(ValidationError):
         handle_dynamodb_no_envelope(event_dict, LambdaContext())
 
 
 def test_validate_event_does_not_conform_with_schema():
-    event_dict = {"hello": "s"}
+    event_dict: Any = {"hello": "s"}
     with pytest.raises(ValidationError):
         handle_dynamodb(event_dict, LambdaContext())
+
+
+def test_validate_event_neither_image_exists_with_schema():
+    event_dict: Any = {
+        "Records": [
+            {
+                "eventID": "1",
+                "eventName": "INSERT",
+                "eventVersion": "1.0",
+                "eventSourceARN": "eventsource_arn",
+                "awsRegion": "us-west-2",
+                "eventSource": "aws:dynamodb",
+                "dynamodb": {
+                    "StreamViewType": "NEW_AND_OLD_IMAGES",
+                    "SequenceNumber": "111",
+                    "SizeBytes": 26,
+                    "Keys": {"Id": {"N": "101"}},
+                },
+            }
+        ]
+    }
+    with pytest.raises(ValidationError) as exc_info:
+        handle_dynamodb(event_dict, LambdaContext())
+
+    validation_error: ValidationError = exc_info.value
+    assert len(validation_error.errors()) == 1
+    error = validation_error.errors()[0]
+    assert error["msg"] == "DynamoDB streams schema failed validation, missing both new & old stream images"

tests/functional/parser/test_eventbridge.py

@@ -1,3 +1,8 @@
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
 from aws_lambda_powertools.utilities.advanced_parser.envelopes.envelopes import Envelope
 from aws_lambda_powertools.utilities.advanced_parser.parser import parser
 from aws_lambda_powertools.utilities.typing import LambdaContext
@@ -6,13 +11,13 @@
 
 
 @parser(schema=MyEventbridgeBusiness, envelope=Envelope.EVENTBRIDGE)
-def handle_eventbridge(event: MyEventbridgeBusiness, context: LambdaContext):
+def handle_eventbridge(event: MyEventbridgeBusiness, _: LambdaContext):
     assert event.instance_id == "i-1234567890abcdef0"
     assert event.state == "terminated"
 
 
 @parser(schema=MyAdvancedEventbridgeBusiness)
-def handle_eventbridge_no_envelope(event: MyAdvancedEventbridgeBusiness, context: LambdaContext):
+def handle_eventbridge_no_envelope(event: MyAdvancedEventbridgeBusiness, _: LambdaContext):
     assert event.detail.instance_id == "i-1234567890abcdef0"
     assert event.detail.state == "terminated"
     assert event.id == "6a7e8feb-b491-4cf7-a9f1-bf3703467718"
@@ -31,6 +36,23 @@ def test_handle_eventbridge_trigger_event():
     handle_eventbridge(event_dict, LambdaContext())
 
 
+def test_validate_event_does_not_conform_with_user_dict_schema():
+    event_dict: Any = {
+        "version": "0",
+        "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
+        "detail-type": "EC2 Instance State-change Notification",
+        "source": "aws.ec2",
+        "account": "111122223333",
+        "time": "2017-12-22T18:43:48Z",
+        "region": "us-west-1",
+        "resources": ["arn:aws:ec2:us-west-1:123456789012:instance/i-1234567890abcdef0"],
+        "detail": {},
+    }
+    with pytest.raises(ValidationError) as e:
+        handle_eventbridge(event_dict, LambdaContext())
+    print(e.exconly())
+
+
 def test_handle_eventbridge_trigger_event_no_envelope():
     event_dict = load_event("eventBridgeEvent.json")
     handle_eventbridge_no_envelope(event_dict, LambdaContext())

tests/functional/parser/test_sqs.py

@@ -1,4 +1,7 @@
-from typing import List
+from typing import Any, List
+
+import pytest
+from pydantic import ValidationError
 
 from aws_lambda_powertools.utilities.advanced_parser.envelopes.envelopes import Envelope
 from aws_lambda_powertools.utilities.advanced_parser.parser import parser
@@ -9,7 +12,7 @@
 
 
 @parser(schema=str, envelope=Envelope.SQS)
-def handle_sqs_str_body(event: List[str], context: LambdaContext):
+def handle_sqs_str_body(event: List[str], _: LambdaContext):
     assert len(event) == 2
     assert event[0] == "Test message."
     assert event[1] == "Test message2."
@@ -21,18 +24,53 @@ def test_handle_sqs_trigger_event_str_body():
 
 
 @parser(schema=MySqsBusiness, envelope=Envelope.SQS)
-def handle_sqs_json_body(event: List[MySqsBusiness], context: LambdaContext):
+def handle_sqs_json_body(event: List[MySqsBusiness], _: LambdaContext):
     assert len(event) == 1
     assert event[0].message == "hello world"
     assert event[0].username == "lessa"
 
 
-def test_handle_sqs_trigger_evemt_json_body(sqs_event):  # noqa: F811
+def test_handle_sqs_trigger_event_json_body(sqs_event):  # noqa: F811
     handle_sqs_json_body(sqs_event, LambdaContext())
 
 
+def test_validate_event_does_not_conform_with_schema():
+    event: Any = {"invalid": "event"}
+
+    with pytest.raises(ValidationError):
+        handle_sqs_json_body(event, LambdaContext())
+
+
+def test_validate_event_does_not_conform_user_json_string_with_schema():
+    event: Any = {
+        "Records": [
+            {
+                "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
+                "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
+                "body": "Not valid json",
+                "attributes": {
+                    "ApproximateReceiveCount": "1",
+                    "SentTimestamp": "1545082649183",
+                    "SenderId": "AIDAIENQZJOLO23YVJ4VO",
+                    "ApproximateFirstReceiveTimestamp": "1545082649185",
+                },
+                "messageAttributes": {
+                    "testAttr": {"stringValue": "100", "binaryValue": "base64Str", "dataType": "Number"}
+                },
+                "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
+                "eventSource": "aws:sqs",
+                "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
+                "awsRegion": "us-east-2",
+            }
+        ]
+    }
+
+    with pytest.raises(ValidationError):
+        handle_sqs_json_body(event, LambdaContext())
+
+
 @parser(schema=MyAdvancedSqsBusiness)
-def handle_sqs_no_envelope(event: MyAdvancedSqsBusiness, context: LambdaContext):
+def handle_sqs_no_envelope(event: MyAdvancedSqsBusiness, _: LambdaContext):
     records = event.Records
     record = records[0]
     attributes = record.attributes

tests/functional/parser/utils.py

@@ -1,12 +1,13 @@
 import json
 import os
+from typing import Any
 
 
-def get_event_file_path(file_name: str) -> dict:
+def get_event_file_path(file_name: str) -> str:
     return os.path.dirname(os.path.realpath(__file__)) + "/../../events/" + file_name
 
 
-def load_event(file_name: str) -> dict:
-    full_file_name = os.path.dirname(os.path.realpath(__file__)) + "/../../events/" + file_name
+def load_event(file_name: str) -> Any:
+    full_file_name = get_event_file_path(file_name)
     with open(full_file_name) as fp:
         return json.load(fp)