Add test helper to generate DynamoDB event

chalice/test.py

@@ -4,7 +4,7 @@
 import contextlib
 from types import TracebackType
 
-from typing import Optional, Type, Generator, Dict, Any, List # noqa
+from typing import Optional, Type, Generator, Dict, Any, List, Tuple # noqa
 
 from chalice import Chalice # noqa
 from chalice.config import Config
@@ -323,6 +323,29 @@ def generate_kinesis_event(self, message_bodies,
  } for body in message_bodies]
  return {'Records': records}
 
+ def generate_dynamodb_event(self, images):
+ # type: (List[Tuple[Dict, Dict]]) -> Dict[str, Any]
+ records = [{
+ "dynamodb": {
+ "ApproximateCreationDateTime": 1545084650.987,
+ "Keys": list(new_image.keys()),
+ "NewImage": new_image,
+ "OldImage": old_image,
+ "SequenceNumber": "12345",
+ "SizeBytes": 12345,
+ "StreamViewType": "NEW_AND_OLD_IMAGES",
+ },
+ "awsRegion": "us-west-2",
+ "eventID": "da037887f71a88a1f6f4cfd149709d5a",
+ "eventName": "INSERT",
+ "eventSource": "aws:dynamodb",
+ "eventSourceARN": (
+ "arn:aws:dynamodb:us-west-2:12345:table/MyTable/stream/"
+ "2015-05-11T21:21:33.291"
+ )
+ } for new_image, old_image in images]
+ return {'Records': records}
+
 
 class TestLambdaClient(BaseClient):
  def __init__(self, app, config):

docs/source/api.rst

@@ -1800,6 +1800,10 @@ Testing
 
  Generates a Kinesis event.
 
+ .. method:: generate_dynamodb_event(images)
+
+ Generates a DynamoDB Stream event.
+
 
 .. class:: HTTPResponse()
 

tests/unit/test_test.py

@@ -313,6 +313,29 @@ def foo(event):
  assert response.payload == [b'foo', b'bar', b'baz']
 
 
+def test_can_generate_dynamodb_event():
+ app = Chalice('dynamodb')
+
+ @app.on_dynamodb_record(
+ stream_arn=('arn:aws:dynamodb:us-west-2:12345:table/MyTable/stream/'
+ '2015-05-11T21:21:33.291')
+ )
+ def foo(event):
+ return [(record.new_image, record.old_image) for record in event]
+
+ old_image = {'PK': {'S': 'foo'}, 'SK': {'S': 'bar'}}
+ new_image = {'PK': {'S': 'hello'}, 'SK': {'S': 'world'}}
+
+ with Client(app) as client:
+ event = client.events.generate_dynamodb_event(
+ images=[(old_image, new_image)]
+ )
+ response = client.lambda_.invoke('foo', event)
+ assert len(response.payload) == 1
+ assert response.payload[0][0] == old_image
+ assert response.payload[0][1] == new_image
+
+
 def test_can_mix_pure_lambda_and_event_handlers():
  app = Chalice('lambda-only')