From 0b57ffe26a894ecc7d39a9c1155dadedc0c5f7ba Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Wed, 7 Nov 2018 15:03:25 -0500 Subject: [PATCH 1/7] Add StreamSpecification to dynamodb2 package --- moto/dynamodb2/models.py | 22 +++++++++- moto/dynamodb2/responses.py | 9 +++- tests/test_dynamodb2/test_dynamodb.py | 59 +++++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 2 deletions(-) diff --git a/moto/dynamodb2/models.py b/moto/dynamodb2/models.py index a54c4f7d0..16e97ea2f 100644 --- a/moto/dynamodb2/models.py +++ b/moto/dynamodb2/models.py @@ -294,7 +294,7 @@ class Item(BaseModel): class Table(BaseModel): - def __init__(self, table_name, schema=None, attr=None, throughput=None, indexes=None, global_indexes=None): + def __init__(self, table_name, schema=None, attr=None, throughput=None, indexes=None, global_indexes=None, streams=None): self.name = table_name self.attr = attr self.schema = schema @@ -325,10 +325,18 @@ class Table(BaseModel): 'TimeToLiveStatus': 'DISABLED' # One of 'ENABLING'|'DISABLING'|'ENABLED'|'DISABLED', # 'AttributeName': 'string' # Can contain this } + self.set_stream_specification(streams) def _generate_arn(self, name): return 'arn:aws:dynamodb:us-east-1:123456789011:table/' + name + def set_stream_specification(self, streams): + self.stream_specification = streams + if streams and streams.get('StreamEnabled'): + self.latest_stream_label = datetime.datetime.utcnow().isoformat() + else: + self.latest_stream_label = None + def describe(self, base_key='TableDescription'): results = { base_key: { @@ -345,6 +353,11 @@ class Table(BaseModel): 'LocalSecondaryIndexes': [index for index in self.indexes], } } + if self.stream_specification: + results[base_key]['StreamSpecification'] = self.stream_specification + if self.latest_stream_label: + results[base_key]['LatestStreamLabel'] = self.latest_stream_label + results[base_key]['LatestStreamArn'] = self.table_arn + '/stream/' + self.latest_stream_label return results def __len__(self): @@ -680,6 +693,13 @@ class DynamoDBBackend(BaseBackend): table.throughput = throughput return table + def update_table_streams(self, name, stream_specification): + table = self.tables[name] + if stream_specification['StreamEnabled'] and table.latest_stream_label: + raise ValueError('Table already has stream enabled') + table.set_stream_specification(stream_specification) + return table + def update_table_global_indexes(self, name, global_index_updates): table = self.tables[name] gsis_by_name = dict((i['IndexName'], i) for i in table.global_indexes) diff --git a/moto/dynamodb2/responses.py b/moto/dynamodb2/responses.py index e2f1ef1cc..73bd3ae38 100644 --- a/moto/dynamodb2/responses.py +++ b/moto/dynamodb2/responses.py @@ -104,13 +104,16 @@ class DynamoHandler(BaseResponse): # getting the indexes global_indexes = body.get("GlobalSecondaryIndexes", []) local_secondary_indexes = body.get("LocalSecondaryIndexes", []) + # get the stream specification + streams = body.get("StreamSpecification") table = self.dynamodb_backend.create_table(table_name, schema=key_schema, throughput=throughput, attr=attr, global_indexes=global_indexes, - indexes=local_secondary_indexes) + indexes=local_secondary_indexes, + streams=streams) if table is not None: return dynamo_json_dump(table.describe()) else: @@ -163,12 +166,16 @@ class DynamoHandler(BaseResponse): def update_table(self): name = self.body['TableName'] + table = self.dynamodb_backend.get_table(name) if 'GlobalSecondaryIndexUpdates' in self.body: table = self.dynamodb_backend.update_table_global_indexes( name, self.body['GlobalSecondaryIndexUpdates']) if 'ProvisionedThroughput' in self.body: throughput = self.body["ProvisionedThroughput"] table = self.dynamodb_backend.update_table_throughput(name, throughput) + if 'StreamSpecification' in self.body: + table = self.dynamodb_backend.update_table_streams(name, self.body['StreamSpecification']) + return dynamo_json_dump(table.describe()) def describe_table(self): diff --git a/tests/test_dynamodb2/test_dynamodb.py b/tests/test_dynamodb2/test_dynamodb.py index afc919dd7..7f30bbccf 100644 --- a/tests/test_dynamodb2/test_dynamodb.py +++ b/tests/test_dynamodb2/test_dynamodb.py @@ -1336,3 +1336,62 @@ def test_query_global_secondary_index_when_created_via_update_table_resource(): assert len(forum_and_subject_items) == 1 assert forum_and_subject_items[0] == {'user_id': Decimal('1'), 'forum_name': 'cats', 'subject': 'my pet is the cutest'} + + +@mock_dynamodb2 +def test_dynamodb_streams_1(): + conn = boto3.client('dynamodb', region_name='us-east-1') + + resp = conn.create_table( + TableName='test-streams', + KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}], + AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}], + ProvisionedThroughput={'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1}, + StreamSpecification={ + 'StreamEnabled': True, + 'StreamViewType': 'NEW_AND_OLD_IMAGES' + } + ) + + assert 'StreamSpecification' in resp['TableDescription'] + assert resp['TableDescription']['StreamSpecification'] == { + 'StreamEnabled': True, + 'StreamViewType': 'NEW_AND_OLD_IMAGES' + } + assert 'LatestStreamLabel' in resp['TableDescription'] + assert 'LatestStreamArn' in resp['TableDescription'] + + resp = conn.delete_table(TableName='test-streams') + + assert 'StreamSpecification' in resp['TableDescription'] + + +@mock_dynamodb2 +def test_dynamodb_streams_2(): + conn = boto3.client('dynamodb', region_name='us-east-1') + + resp = conn.create_table( + TableName='test-stream-update', + KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}], + AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}], + ProvisionedThroughput={'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1}, + ) + + assert 'StreamSpecification' not in resp['TableDescription'] + + resp = conn.update_table( + TableName='test-stream-update', + StreamSpecification={ + 'StreamEnabled': True, + 'StreamViewType': 'NEW_IMAGE' + } + ) + + assert 'StreamSpecification' in resp['TableDescription'] + assert resp['TableDescription']['StreamSpecification'] == { + 'StreamEnabled': True, + 'StreamViewType': 'NEW_IMAGE' + } + assert 'LatestStreamLabel' in resp['TableDescription'] + assert 'LatestStreamArn' in resp['TableDescription'] + From 519899f74f6d50c33ffe82d8d6fee2c86afd3a8e Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Wed, 7 Nov 2018 17:10:00 -0500 Subject: [PATCH 2/7] Much of the way towards complete DynamoDB Streams implementation --- moto/__init__.py | 1 + moto/backends.py | 1 + moto/dynamodb2/models.py | 37 ++++++- moto/dynamodbstreams/__init__.py | 6 ++ moto/dynamodbstreams/models.py | 99 +++++++++++++++++++ moto/dynamodbstreams/responses.py | 29 ++++++ moto/dynamodbstreams/urls.py | 10 ++ setup.cfg | 2 +- .../test_dynamodbstreams.py | 96 ++++++++++++++++++ 9 files changed, 278 insertions(+), 3 deletions(-) create mode 100644 moto/dynamodbstreams/__init__.py create mode 100644 moto/dynamodbstreams/models.py create mode 100644 moto/dynamodbstreams/responses.py create mode 100644 moto/dynamodbstreams/urls.py create mode 100644 tests/test_dynamodbstreams/test_dynamodbstreams.py diff --git a/moto/__init__.py b/moto/__init__.py index dd3593d5d..e86c499a7 100644 --- a/moto/__init__.py +++ b/moto/__init__.py @@ -16,6 +16,7 @@ from .cognitoidp import mock_cognitoidp, mock_cognitoidp_deprecated # flake8: n from .datapipeline import mock_datapipeline, mock_datapipeline_deprecated # flake8: noqa from .dynamodb import mock_dynamodb, mock_dynamodb_deprecated # flake8: noqa from .dynamodb2 import mock_dynamodb2, mock_dynamodb2_deprecated # flake8: noqa +from .dynamodbstreams import mock_dynamodbstreams # flake8: noqa from .ec2 import mock_ec2, mock_ec2_deprecated # flake8: noqa from .ecr import mock_ecr, mock_ecr_deprecated # flake8: noqa from .ecs import mock_ecs, mock_ecs_deprecated # flake8: noqa diff --git a/moto/backends.py b/moto/backends.py index d95424385..7df167a06 100644 --- a/moto/backends.py +++ b/moto/backends.py @@ -12,6 +12,7 @@ from moto.core import moto_api_backends from moto.datapipeline import datapipeline_backends from moto.dynamodb import dynamodb_backends from moto.dynamodb2 import dynamodb_backends2 +from moto.dynamodbstreams import dynamodbstreams_backends from moto.ec2 import ec2_backends from moto.ecr import ecr_backends from moto.ecs import ecs_backends diff --git a/moto/dynamodb2/models.py b/moto/dynamodb2/models.py index 16e97ea2f..d58241cf6 100644 --- a/moto/dynamodb2/models.py +++ b/moto/dynamodb2/models.py @@ -292,6 +292,34 @@ class Item(BaseModel): 'ADD not supported for %s' % ', '.join(update_action['Value'].keys())) +class StreamShard(BaseModel): + def __init__(self, table): + self.table = table + self.id = 'shardId-00000001541626099285-f35f62ef' + self.starting_sequence_number = 1100000000017454423009 + self.items = [] + self.created_on = datetime.datetime.utcnow() + + def to_json(self): + return { + 'ShardId': self.id, + 'SequenceNumberRange': { + 'StartingSequenceNumber': str(self.starting_sequence_number) + } + } + + def add(self, old, new): + t = self.table.stream_specification['StreamViewType'] + if t == 'KEYS_ONLY': + self.items.append(new.key) + elif t == 'NEW_IMAGE': + self.items.append(new) + elif t == 'OLD_IMAGE': + self.items.append(old) + elif t == 'NEW_AND_OLD_IMAGES': + self.items.append((old, new)) + + class Table(BaseModel): def __init__(self, table_name, schema=None, attr=None, throughput=None, indexes=None, global_indexes=None, streams=None): @@ -334,8 +362,10 @@ class Table(BaseModel): self.stream_specification = streams if streams and streams.get('StreamEnabled'): self.latest_stream_label = datetime.datetime.utcnow().isoformat() + self.stream_shard = StreamShard(self) else: self.latest_stream_label = None + self.stream_shard = None def describe(self, base_key='TableDescription'): results = { @@ -398,6 +428,7 @@ class Table(BaseModel): else: range_value = None + current = self.get_item(hash_value, lookup_range_value) item = Item(hash_value, self.hash_key_type, range_value, self.range_key_type, item_attrs) @@ -413,8 +444,6 @@ class Table(BaseModel): else: lookup_range_value = DynamoType(expected_range_value) - current = self.get_item(hash_value, lookup_range_value) - if current is None: current_attr = {} elif hasattr(current, 'attrs'): @@ -445,6 +474,10 @@ class Table(BaseModel): self.items[hash_value][range_value] = item else: self.items[hash_value] = item + + if self.stream_shard is not None: + self.stream_shard.add(current, item) + return item def __nonzero__(self): diff --git a/moto/dynamodbstreams/__init__.py b/moto/dynamodbstreams/__init__.py new file mode 100644 index 000000000..b35879eba --- /dev/null +++ b/moto/dynamodbstreams/__init__.py @@ -0,0 +1,6 @@ +from __future__ import unicode_literals +from .models import dynamodbstreams_backends +from ..core.models import base_decorator + +dynamodbstreams_backend = dynamodbstreams_backends['us-east-1'] +mock_dynamodbstreams = base_decorator(dynamodbstreams_backends) diff --git a/moto/dynamodbstreams/models.py b/moto/dynamodbstreams/models.py new file mode 100644 index 000000000..35704de63 --- /dev/null +++ b/moto/dynamodbstreams/models.py @@ -0,0 +1,99 @@ +from __future__ import unicode_literals + +import os +import json +import boto3 +import base64 +import datetime + +from moto.core import BaseBackend, BaseModel +from moto.dynamodb2.models import dynamodb_backends + + +class ShardIterator(BaseModel): + def __init__(self, stream_shard, shard_iterator_type, sequence_number=None): + self.id = base64.b64encode(os.urandom(472)).decode('utf-8') + self.stream_shard = stream_shard + self.shard_iterator_type = shard_iterator_type + if shard_iterator_type == 'TRIM_HORIZON': + self.sequence_number = stream_shard.starting_sequence_number + elif shard_iterator_type == 'LATEST': + self.sequence_number = stream_shard.starting_sequence_number + len(stream_shard.items) + elif shard_iterator_type == 'AT_SEQUENCE_NUMBER': + self.sequence_number = sequence_number + elif shard_iterator_type == 'AFTER_SEQUENCE_NUMBER': + self.sequence_number = sequence_number + 1 + + def to_json(self): + return { + 'ShardIterator': '{}/stream/{}|1|{}'.format( + self.stream_shard.table.table_arn, + self.stream_shard.table.latest_stream_label, + self.id) + } + + +class DynamoDBStreamsBackend(BaseBackend): + def __init__(self, region): + self.region = region + + def reset(self): + region = self.region + self.__dict__ = {} + self.__init__(region) + + @property + def dynamodb(self): + return dynamodb_backends[self.region] + + def _get_table_from_arn(self, arn): + table_name = arn.split(':', 6)[5].split('/')[1] + return self.dynamodb.get_table(table_name) + + def describe_stream(self, arn): + table = self._get_table_from_arn(arn) + resp = {'StreamDescription': { + 'StreamArn': arn, + 'StreamLabel': table.latest_stream_label, + 'StreamStatus': ('ENABLED' if table.latest_stream_label + else 'DISABLED'), + 'StreamViewType': table.stream_specification['StreamViewType'], + 'CreationRequestDateTime': table.stream_shard.created_on.isoformat(), + 'TableName': table.name, + 'KeySchema': table.schema, + 'Shards': ([table.stream_shard.to_json()] if table.stream_shard + else []) + }} + + return json.dumps(resp) + + def list_streams(self, table_name=None): + streams = [] + for table in self.dynamodb.tables.values(): + if table_name is not None and table.name != table_name: + continue + if table.latest_stream_label: + d = table.describe(base_key='Table') + streams.append({ + 'StreamArn': d['Table']['LatestStreamArn'], + 'TableName': d['Table']['TableName'], + 'StreamLabel': d['Table']['LatestStreamLabel'] + }) + + return json.dumps({'Streams': streams}) + + def get_shard_iterator(self, arn, shard_id, shard_iterator_type, sequence_number=None): + table = self._get_table_from_arn(arn) + assert table.stream_shard.id == shard_id + + shard_iterator = ShardIterator(table.stream_shard, shard_iterator_type, + sequence_number) + + return json.dumps(shard_iterator.to_json()) + + + +available_regions = boto3.session.Session().get_available_regions( + 'dynamodbstreams') +dynamodbstreams_backends = {region: DynamoDBStreamsBackend(region=region) + for region in available_regions} diff --git a/moto/dynamodbstreams/responses.py b/moto/dynamodbstreams/responses.py new file mode 100644 index 000000000..ef05c1718 --- /dev/null +++ b/moto/dynamodbstreams/responses.py @@ -0,0 +1,29 @@ +from __future__ import unicode_literals + +import json + +from moto.core.responses import BaseResponse + +from .models import dynamodbstreams_backends + + +class DynamoDBStreamsHandler(BaseResponse): + + @property + def backend(self): + return dynamodbstreams_backends[self.region] + + def describe_stream(self): + arn = self._get_param('StreamArn') + return self.backend.describe_stream(arn) + + def list_streams(self): + table_name = self._get_param('TableName') + return self.backend.list_streams(table_name) + + def get_shard_iterator(self): + arn = self._get_param('StreamArn') + shard_id = self._get_param('ShardId') + shard_iterator_type = self._get_param('ShardIteratorType') + return self.backend.get_shard_iterator(arn, shard_id, + shard_iterator_type) diff --git a/moto/dynamodbstreams/urls.py b/moto/dynamodbstreams/urls.py new file mode 100644 index 000000000..1d0f94c35 --- /dev/null +++ b/moto/dynamodbstreams/urls.py @@ -0,0 +1,10 @@ +from __future__ import unicode_literals +from .responses import DynamoDBStreamsHandler + +url_bases = [ + "https?://streams.dynamodb.(.+).amazonaws.com" +] + +url_paths = { + "{0}/$": DynamoDBStreamsHandler.dispatch, +} diff --git a/setup.cfg b/setup.cfg index fb04c16a8..9dbd988db 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,7 +1,7 @@ [nosetests] verbosity=1 detailed-errors=1 -with-coverage=1 +#with-coverage=1 cover-package=moto [bdist_wheel] diff --git a/tests/test_dynamodbstreams/test_dynamodbstreams.py b/tests/test_dynamodbstreams/test_dynamodbstreams.py new file mode 100644 index 000000000..a10445aac --- /dev/null +++ b/tests/test_dynamodbstreams/test_dynamodbstreams.py @@ -0,0 +1,96 @@ +from __future__ import unicode_literals, print_function + +import boto3 +from moto import mock_dynamodb2, mock_dynamodbstreams + + +class TestClass(): + stream_arn = None + mocks = [] + + def setup(self): + self.mocks = [mock_dynamodb2(), mock_dynamodbstreams()] + for m in self.mocks: + m.start() + + # create a table with a stream + conn = boto3.client('dynamodb', region_name='us-east-1') + + resp = conn.create_table( + TableName='test-streams', + KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}], + AttributeDefinitions=[{'AttributeName': 'id', + 'AttributeType': 'S'}], + ProvisionedThroughput={'ReadCapacityUnits': 1, + 'WriteCapacityUnits': 1}, + StreamSpecification={ + 'StreamEnabled': True, + 'StreamViewType': 'NEW_AND_OLD_IMAGES' + } + ) + self.stream_arn = resp['TableDescription']['LatestStreamArn'] + + def teardown(self): + conn = boto3.client('dynamodb', region_name='us-east-1') + conn.delete_table(TableName='test-streams') + self.stream_arn = None + + for m in self.mocks: + m.stop() + + + def test_verify_stream(self): + conn = boto3.client('dynamodb', region_name='us-east-1') + resp = conn.describe_table(TableName='test-streams') + assert 'LatestStreamArn' in resp['Table'] + + def test_describe_stream(self): + conn = boto3.client('dynamodbstreams', region_name='us-east-1') + + resp = conn.describe_stream(StreamArn=self.stream_arn) + assert 'StreamDescription' in resp + desc = resp['StreamDescription'] + assert desc['StreamArn'] == self.stream_arn + assert desc['TableName'] == 'test-streams' + + def test_list_streams(self): + conn = boto3.client('dynamodbstreams', region_name='us-east-1') + + resp = conn.list_streams() + assert resp['Streams'][0]['StreamArn'] == self.stream_arn + + resp = conn.list_streams(TableName='no-stream') + assert not resp['Streams'] + + def test_get_shard_iterator(self): + conn = boto3.client('dynamodbstreams', region_name='us-east-1') + + resp = conn.describe_stream(StreamArn=self.stream_arn) + shard_id = resp['StreamDescription']['Shards'][0]['ShardId'] + + resp = conn.get_shard_iterator( + StreamArn=self.stream_arn, + ShardId=shard_id, + ShardIteratorType='TRIM_HORIZON' + ) + assert 'ShardIterator' in resp + + def test_get_records(self): + conn = boto3.client('dynamodbstreams', region_name='us-east-1') + + resp = conn.describe_stream(StreamArn=self.stream_arn) + shard_id = resp['StreamDescription']['Shards'][0]['ShardId'] + + resp = conn.get_shard_iterator( + StreamArn=self.stream_arn, + ShardId=shard_id, + ShardIteratorType='TRIM_HORIZON' + ) + iterator_id = resp['ShardIterator'] + + resp = conn.get_records(ShardIterator=iterator_id) + assert 'Records' in resp + + # TODO: Add tests for inserting records into the stream, and + # the various stream types + From 0f6086f708112c024eb7f3079339b1698a507012 Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Thu, 8 Nov 2018 10:54:54 -0500 Subject: [PATCH 3/7] Finalize implementation of DynamoDB Streams --- moto/dynamodb2/models.py | 97 ++++++++++++++----- moto/dynamodbstreams/models.py | 44 +++++++-- moto/dynamodbstreams/responses.py | 7 ++ setup.cfg | 2 +- .../test_dynamodbstreams.py | 49 +++++++++- 5 files changed, 166 insertions(+), 33 deletions(-) diff --git a/moto/dynamodb2/models.py b/moto/dynamodb2/models.py index d58241cf6..4283c038b 100644 --- a/moto/dynamodb2/models.py +++ b/moto/dynamodb2/models.py @@ -1,10 +1,11 @@ from __future__ import unicode_literals -from collections import defaultdict +from collections import defaultdict, namedtuple import copy import datetime import decimal import json import re +import uuid import boto3 from moto.compat import OrderedDict @@ -292,6 +293,44 @@ class Item(BaseModel): 'ADD not supported for %s' % ', '.join(update_action['Value'].keys())) +class StreamRecord(BaseModel): + def __init__(self, table, stream_type, event_name, old, new, seq): + old_a = old.to_json()['Attributes'] if old is not None else {} + new_a = new.to_json()['Attributes'] if new is not None else {} + + rec = old if old is not None else new + keys = {table.hash_key_attr: rec.hash_key.to_json()} + if table.range_key_attr is not None: + keys[table.range_key_attr] = rec.range_key.to_json() + + self.record = { + 'eventID': uuid.uuid4().hex, + 'eventName': event_name, + 'eventSource': 'aws:dynamodb', + 'eventVersion': '1.0', + 'awsRegion': 'us-east-1', + 'dynamodb': { + 'StreamViewType': stream_type, + 'ApproximateCreationDateTime': datetime.datetime.utcnow().isoformat(), + 'SequenceNumber': seq, + 'SizeBytes': 1, + 'Keys': keys + } + } + + if stream_type in ('NEW_IMAGE', 'NEW_AND_OLD_IMAGES'): + self.record['dynamodb']['NewImage'] = new_a + if stream_type in ('OLD_IMAGE', 'NEW_AND_OLD_IMAGES'): + self.record['dynamodb']['OldImage'] = old_a + + # This is a substantial overestimate but it's the easiest to do now + self.record['dynamodb']['SizeBytes'] = len( + json.dumps(self.record['dynamodb'])) + + def to_json(self): + return self.record + + class StreamShard(BaseModel): def __init__(self, table): self.table = table @@ -310,15 +349,22 @@ class StreamShard(BaseModel): def add(self, old, new): t = self.table.stream_specification['StreamViewType'] - if t == 'KEYS_ONLY': - self.items.append(new.key) - elif t == 'NEW_IMAGE': - self.items.append(new) - elif t == 'OLD_IMAGE': - self.items.append(old) - elif t == 'NEW_AND_OLD_IMAGES': - self.items.append((old, new)) - + if old is None: + event_name = 'INSERT' + elif new is None: + event_name = 'DELETE' + else: + event_name = 'MODIFY' + seq = len(self.items) + self.starting_sequence_number + self.items.append( + StreamRecord(self.table, t, event_name, old, new, seq)) + + def get(self, start, quantity): + start -= self.starting_sequence_number + assert start >= 0 + end = start + quantity + return [i.to_json() for i in self.items[start:end]] + class Table(BaseModel): @@ -428,22 +474,22 @@ class Table(BaseModel): else: range_value = None + if expected is None: + expected = {} + lookup_range_value = range_value + else: + expected_range_value = expected.get( + self.range_key_attr, {}).get("Value") + if(expected_range_value is None): + lookup_range_value = range_value + else: + lookup_range_value = DynamoType(expected_range_value) current = self.get_item(hash_value, lookup_range_value) + item = Item(hash_value, self.hash_key_type, range_value, self.range_key_type, item_attrs) if not overwrite: - if expected is None: - expected = {} - lookup_range_value = range_value - else: - expected_range_value = expected.get( - self.range_key_attr, {}).get("Value") - if(expected_range_value is None): - lookup_range_value = range_value - else: - lookup_range_value = DynamoType(expected_range_value) - if current is None: current_attr = {} elif hasattr(current, 'attrs'): @@ -508,9 +554,14 @@ class Table(BaseModel): def delete_item(self, hash_key, range_key): try: if range_key: - return self.items[hash_key].pop(range_key) + item = self.items[hash_key].pop(range_key) else: - return self.items.pop(hash_key) + item = self.items.pop(hash_key) + + if self.stream_shard is not None: + self.stream_shard.add(item, None) + + return item except KeyError: return None diff --git a/moto/dynamodbstreams/models.py b/moto/dynamodbstreams/models.py index 35704de63..7b43eb744 100644 --- a/moto/dynamodbstreams/models.py +++ b/moto/dynamodbstreams/models.py @@ -11,8 +11,9 @@ from moto.dynamodb2.models import dynamodb_backends class ShardIterator(BaseModel): - def __init__(self, stream_shard, shard_iterator_type, sequence_number=None): + def __init__(self, streams_backend, stream_shard, shard_iterator_type, sequence_number=None): self.id = base64.b64encode(os.urandom(472)).decode('utf-8') + self.streams_backend = streams_backend self.stream_shard = stream_shard self.shard_iterator_type = shard_iterator_type if shard_iterator_type == 'TRIM_HORIZON': @@ -24,18 +25,43 @@ class ShardIterator(BaseModel): elif shard_iterator_type == 'AFTER_SEQUENCE_NUMBER': self.sequence_number = sequence_number + 1 + @property + def arn(self): + return '{}/stream/{}|1|{}'.format( + self.stream_shard.table.table_arn, + self.stream_shard.table.latest_stream_label, + self.id) + def to_json(self): return { - 'ShardIterator': '{}/stream/{}|1|{}'.format( - self.stream_shard.table.table_arn, - self.stream_shard.table.latest_stream_label, - self.id) + 'ShardIterator': self.arn + } + + def get(self, limit=1000): + items = self.stream_shard.get(self.sequence_number, limit) + try: + last_sequence_number = max(i['dynamodb']['SequenceNumber'] for i in items) + new_shard_iterator = ShardIterator(self.streams_backend, + self.stream_shard, + 'AFTER_SEQUENCE_NUMBER', + last_sequence_number) + except ValueError: + new_shard_iterator = ShardIterator(self.streams_backend, + self.stream_shard, + 'AT_SEQUENCE_NUMBER', + self.sequence_number) + + self.streams_backend.shard_iterators[new_shard_iterator.arn] = new_shard_iterator + return { + 'NextShardIterator': new_shard_iterator.arn, + 'Records': items } class DynamoDBStreamsBackend(BaseBackend): def __init__(self, region): self.region = region + self.shard_iterators = {} def reset(self): region = self.region @@ -86,11 +112,17 @@ class DynamoDBStreamsBackend(BaseBackend): table = self._get_table_from_arn(arn) assert table.stream_shard.id == shard_id - shard_iterator = ShardIterator(table.stream_shard, shard_iterator_type, + shard_iterator = ShardIterator(self, table.stream_shard, + shard_iterator_type, sequence_number) + self.shard_iterators[shard_iterator.arn] = shard_iterator return json.dumps(shard_iterator.to_json()) + def get_records(self, iterator_arn, limit): + shard_iterator = self.shard_iterators[iterator_arn] + return json.dumps(shard_iterator.get(limit)) + available_regions = boto3.session.Session().get_available_regions( diff --git a/moto/dynamodbstreams/responses.py b/moto/dynamodbstreams/responses.py index ef05c1718..c07377d38 100644 --- a/moto/dynamodbstreams/responses.py +++ b/moto/dynamodbstreams/responses.py @@ -27,3 +27,10 @@ class DynamoDBStreamsHandler(BaseResponse): shard_iterator_type = self._get_param('ShardIteratorType') return self.backend.get_shard_iterator(arn, shard_id, shard_iterator_type) + + def get_records(self): + arn = self._get_param('ShardIterator') + limit = self._get_param('Limit') + if limit is None: + limit = 1000 + return self.backend.get_records(arn, limit) diff --git a/setup.cfg b/setup.cfg index 9dbd988db..fb04c16a8 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,7 +1,7 @@ [nosetests] verbosity=1 detailed-errors=1 -#with-coverage=1 +with-coverage=1 cover-package=moto [bdist_wheel] diff --git a/tests/test_dynamodbstreams/test_dynamodbstreams.py b/tests/test_dynamodbstreams/test_dynamodbstreams.py index a10445aac..94c0c51b2 100644 --- a/tests/test_dynamodbstreams/test_dynamodbstreams.py +++ b/tests/test_dynamodbstreams/test_dynamodbstreams.py @@ -75,7 +75,7 @@ class TestClass(): ) assert 'ShardIterator' in resp - def test_get_records(self): + def test_get_records_empty(self): conn = boto3.client('dynamodbstreams', region_name='us-east-1') resp = conn.describe_stream(StreamArn=self.stream_arn) @@ -90,7 +90,50 @@ class TestClass(): resp = conn.get_records(ShardIterator=iterator_id) assert 'Records' in resp + assert len(resp['Records']) == 0 - # TODO: Add tests for inserting records into the stream, and - # the various stream types + def test_get_records_seq(self): + conn = boto3.client('dynamodb', region_name='us-east-1') + + conn.put_item( + TableName='test-streams', + Item={ + 'id': {'S': 'entry1'}, + 'first_col': {'S': 'foo'} + } + ) + conn.put_item( + TableName='test-streams', + Item={ + 'id': {'S': 'entry1'}, + 'first_col': {'S': 'bar'}, + 'second_col': {'S': 'baz'} + } + ) + conn.delete_item( + TableName='test-streams', + Key={'id': {'S': 'entry1'}} + ) + conn = boto3.client('dynamodbstreams', region_name='us-east-1') + + resp = conn.describe_stream(StreamArn=self.stream_arn) + shard_id = resp['StreamDescription']['Shards'][0]['ShardId'] + + resp = conn.get_shard_iterator( + StreamArn=self.stream_arn, + ShardId=shard_id, + ShardIteratorType='TRIM_HORIZON' + ) + iterator_id = resp['ShardIterator'] + + resp = conn.get_records(ShardIterator=iterator_id) + assert len(resp['Records']) == 3 + assert resp['Records'][0]['eventName'] == 'INSERT' + assert resp['Records'][1]['eventName'] == 'MODIFY' + assert resp['Records'][2]['eventName'] == 'DELETE' + + # now try fetching from the next shard iterator, it should be + # empty + resp = conn.get_records(ShardIterator=resp['NextShardIterator']) + assert len(resp['Records']) == 0 From ff6a57f44391e2e8523c1ca49022ec39ed0facd6 Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Thu, 8 Nov 2018 11:08:24 -0500 Subject: [PATCH 4/7] Fix flake8 failures --- moto/backends.py | 1 + moto/dynamodb2/models.py | 16 ++++++++-------- moto/dynamodb2/responses.py | 2 +- moto/dynamodbstreams/models.py | 14 ++++++-------- moto/dynamodbstreams/responses.py | 4 +--- 5 files changed, 17 insertions(+), 20 deletions(-) diff --git a/moto/backends.py b/moto/backends.py index 7df167a06..1a333415e 100644 --- a/moto/backends.py +++ b/moto/backends.py @@ -60,6 +60,7 @@ BACKENDS = { 'datapipeline': datapipeline_backends, 'dynamodb': dynamodb_backends, 'dynamodb2': dynamodb_backends2, + 'dynamodbstreams': dynamodbstreams_backends, 'ec2': ec2_backends, 'ecr': ecr_backends, 'ecs': ecs_backends, diff --git a/moto/dynamodb2/models.py b/moto/dynamodb2/models.py index 4283c038b..e2882f1e4 100644 --- a/moto/dynamodb2/models.py +++ b/moto/dynamodb2/models.py @@ -1,5 +1,5 @@ from __future__ import unicode_literals -from collections import defaultdict, namedtuple +from collections import defaultdict import copy import datetime import decimal @@ -302,7 +302,7 @@ class StreamRecord(BaseModel): keys = {table.hash_key_attr: rec.hash_key.to_json()} if table.range_key_attr is not None: keys[table.range_key_attr] = rec.range_key.to_json() - + self.record = { 'eventID': uuid.uuid4().hex, 'eventName': event_name, @@ -317,7 +317,7 @@ class StreamRecord(BaseModel): 'Keys': keys } } - + if stream_type in ('NEW_IMAGE', 'NEW_AND_OLD_IMAGES'): self.record['dynamodb']['NewImage'] = new_a if stream_type in ('OLD_IMAGE', 'NEW_AND_OLD_IMAGES'): @@ -364,7 +364,7 @@ class StreamShard(BaseModel): assert start >= 0 end = start + quantity return [i.to_json() for i in self.items[start:end]] - + class Table(BaseModel): @@ -485,7 +485,7 @@ class Table(BaseModel): else: lookup_range_value = DynamoType(expected_range_value) current = self.get_item(hash_value, lookup_range_value) - + item = Item(hash_value, self.hash_key_type, range_value, self.range_key_type, item_attrs) @@ -520,10 +520,10 @@ class Table(BaseModel): self.items[hash_value][range_value] = item else: self.items[hash_value] = item - + if self.stream_shard is not None: self.stream_shard.add(current, item) - + return item def __nonzero__(self): @@ -560,7 +560,7 @@ class Table(BaseModel): if self.stream_shard is not None: self.stream_shard.add(item, None) - + return item except KeyError: return None diff --git a/moto/dynamodb2/responses.py b/moto/dynamodb2/responses.py index 73bd3ae38..d2bf99bfe 100644 --- a/moto/dynamodb2/responses.py +++ b/moto/dynamodb2/responses.py @@ -175,7 +175,7 @@ class DynamoHandler(BaseResponse): table = self.dynamodb_backend.update_table_throughput(name, throughput) if 'StreamSpecification' in self.body: table = self.dynamodb_backend.update_table_streams(name, self.body['StreamSpecification']) - + return dynamo_json_dump(table.describe()) def describe_table(self): diff --git a/moto/dynamodbstreams/models.py b/moto/dynamodbstreams/models.py index 7b43eb744..41cc6e280 100644 --- a/moto/dynamodbstreams/models.py +++ b/moto/dynamodbstreams/models.py @@ -4,7 +4,6 @@ import os import json import boto3 import base64 -import datetime from moto.core import BaseBackend, BaseModel from moto.dynamodb2.models import dynamodb_backends @@ -31,7 +30,7 @@ class ShardIterator(BaseModel): self.stream_shard.table.table_arn, self.stream_shard.table.latest_stream_label, self.id) - + def to_json(self): return { 'ShardIterator': self.arn @@ -67,7 +66,7 @@ class DynamoDBStreamsBackend(BaseBackend): region = self.region self.__dict__ = {} self.__init__(region) - + @property def dynamodb(self): return dynamodb_backends[self.region] @@ -75,7 +74,7 @@ class DynamoDBStreamsBackend(BaseBackend): def _get_table_from_arn(self, arn): table_name = arn.split(':', 6)[5].split('/')[1] return self.dynamodb.get_table(table_name) - + def describe_stream(self, arn): table = self._get_table_from_arn(arn) resp = {'StreamDescription': { @@ -90,7 +89,7 @@ class DynamoDBStreamsBackend(BaseBackend): 'Shards': ([table.stream_shard.to_json()] if table.stream_shard else []) }} - + return json.dumps(resp) def list_streams(self, table_name=None): @@ -105,7 +104,7 @@ class DynamoDBStreamsBackend(BaseBackend): 'TableName': d['Table']['TableName'], 'StreamLabel': d['Table']['LatestStreamLabel'] }) - + return json.dumps({'Streams': streams}) def get_shard_iterator(self, arn, shard_id, shard_iterator_type, sequence_number=None): @@ -116,14 +115,13 @@ class DynamoDBStreamsBackend(BaseBackend): shard_iterator_type, sequence_number) self.shard_iterators[shard_iterator.arn] = shard_iterator - + return json.dumps(shard_iterator.to_json()) def get_records(self, iterator_arn, limit): shard_iterator = self.shard_iterators[iterator_arn] return json.dumps(shard_iterator.get(limit)) - available_regions = boto3.session.Session().get_available_regions( 'dynamodbstreams') diff --git a/moto/dynamodbstreams/responses.py b/moto/dynamodbstreams/responses.py index c07377d38..c9c113615 100644 --- a/moto/dynamodbstreams/responses.py +++ b/moto/dynamodbstreams/responses.py @@ -1,7 +1,5 @@ from __future__ import unicode_literals -import json - from moto.core.responses import BaseResponse from .models import dynamodbstreams_backends @@ -12,7 +10,7 @@ class DynamoDBStreamsHandler(BaseResponse): @property def backend(self): return dynamodbstreams_backends[self.region] - + def describe_stream(self): arn = self._get_param('StreamArn') return self.backend.describe_stream(arn) From 9d190aa04e4c66831449a1a5cd3673c35e995bf9 Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Thu, 8 Nov 2018 13:22:24 -0500 Subject: [PATCH 5/7] Tweak functionality and add tests --- moto/dynamodb2/models.py | 17 ++-- .../test_dynamodbstreams.py | 99 ++++++++++++++++++- 2 files changed, 108 insertions(+), 8 deletions(-) diff --git a/moto/dynamodb2/models.py b/moto/dynamodb2/models.py index e2882f1e4..d15b9fce5 100644 --- a/moto/dynamodb2/models.py +++ b/moto/dynamodb2/models.py @@ -406,10 +406,13 @@ class Table(BaseModel): def set_stream_specification(self, streams): self.stream_specification = streams - if streams and streams.get('StreamEnabled'): + if streams and (streams.get('StreamEnabled') + or streams.get('StreamViewType')): + self.stream_specification['StreamEnabled'] = True self.latest_stream_label = datetime.datetime.utcnow().isoformat() self.stream_shard = StreamShard(self) else: + self.stream_specification = {'StreamEnabled': False} self.latest_stream_label = None self.stream_shard = None @@ -429,11 +432,11 @@ class Table(BaseModel): 'LocalSecondaryIndexes': [index for index in self.indexes], } } - if self.stream_specification: + if self.stream_specification and self.stream_specification['StreamEnabled']: results[base_key]['StreamSpecification'] = self.stream_specification - if self.latest_stream_label: - results[base_key]['LatestStreamLabel'] = self.latest_stream_label - results[base_key]['LatestStreamArn'] = self.table_arn + '/stream/' + self.latest_stream_label + if self.latest_stream_label: + results[base_key]['LatestStreamLabel'] = self.latest_stream_label + results[base_key]['LatestStreamArn'] = self.table_arn + '/stream/' + self.latest_stream_label return results def __len__(self): @@ -779,7 +782,9 @@ class DynamoDBBackend(BaseBackend): def update_table_streams(self, name, stream_specification): table = self.tables[name] - if stream_specification['StreamEnabled'] and table.latest_stream_label: + if ((stream_specification.get('StreamEnabled') + or stream_specification.get('StreamViewType')) + and table.latest_stream_label): raise ValueError('Table already has stream enabled') table.set_stream_specification(stream_specification) return table diff --git a/tests/test_dynamodbstreams/test_dynamodbstreams.py b/tests/test_dynamodbstreams/test_dynamodbstreams.py index 94c0c51b2..7e4025626 100644 --- a/tests/test_dynamodbstreams/test_dynamodbstreams.py +++ b/tests/test_dynamodbstreams/test_dynamodbstreams.py @@ -1,10 +1,12 @@ from __future__ import unicode_literals, print_function +from nose.tools import assert_raises + import boto3 from moto import mock_dynamodb2, mock_dynamodbstreams -class TestClass(): +class TestCore(): stream_arn = None mocks = [] @@ -84,7 +86,7 @@ class TestClass(): resp = conn.get_shard_iterator( StreamArn=self.stream_arn, ShardId=shard_id, - ShardIteratorType='TRIM_HORIZON' + ShardIteratorType='LATEST' ) iterator_id = resp['ShardIterator'] @@ -137,3 +139,96 @@ class TestClass(): # empty resp = conn.get_records(ShardIterator=resp['NextShardIterator']) assert len(resp['Records']) == 0 + + +class TestEdges(): + mocks = [] + + def setup(self): + self.mocks = [mock_dynamodb2(), mock_dynamodbstreams()] + for m in self.mocks: + m.start() + + def teardown(self): + for m in self.mocks: + m.stop() + + + def test_enable_stream_on_table(self): + conn = boto3.client('dynamodb', region_name='us-east-1') + resp = conn.create_table( + TableName='test-streams', + KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}], + AttributeDefinitions=[{'AttributeName': 'id', + 'AttributeType': 'S'}], + ProvisionedThroughput={'ReadCapacityUnits': 1, + 'WriteCapacityUnits': 1} + ) + assert 'StreamSpecification' not in resp['TableDescription'] + + resp = conn.update_table( + TableName='test-streams', + StreamSpecification={ + 'StreamViewType': 'KEYS_ONLY' + } + ) + assert 'StreamSpecification' in resp['TableDescription'] + assert resp['TableDescription']['StreamSpecification'] == { + 'StreamEnabled': True, + 'StreamViewType': 'KEYS_ONLY' + } + assert 'LatestStreamLabel' in resp['TableDescription'] + + # now try to enable it again + with assert_raises(ValueError): + resp = conn.update_table( + TableName='test-streams', + StreamSpecification={ + 'StreamViewType': 'OLD_IMAGES' + } + ) + + def test_stream_with_range_key(self): + dyn = boto3.client('dynamodb', region_name='us-east-1') + + resp = dyn.create_table( + TableName='test-streams', + KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}, + {'AttributeName': 'color', 'KeyType': 'RANGE'}], + AttributeDefinitions=[{'AttributeName': 'id', + 'AttributeType': 'S'}, + {'AttributeName': 'color', + 'AttributeType': 'S'}], + ProvisionedThroughput={'ReadCapacityUnits': 1, + 'WriteCapacityUnits': 1}, + StreamSpecification={ + 'StreamViewType': 'NEW_IMAGES' + } + ) + stream_arn = resp['TableDescription']['LatestStreamArn'] + + streams = boto3.client('dynamodbstreams', region_name='us-east-1') + resp = streams.describe_stream(StreamArn=stream_arn) + shard_id = resp['StreamDescription']['Shards'][0]['ShardId'] + + resp = streams.get_shard_iterator( + StreamArn=stream_arn, + ShardId=shard_id, + ShardIteratorType='LATEST' + ) + iterator_id = resp['ShardIterator'] + + dyn.put_item( + TableName='test-streams', + Item={'id': {'S': 'row1'}, 'color': {'S': 'blue'}} + ) + dyn.put_item( + TableName='test-streams', + Item={'id': {'S': 'row2'}, 'color': {'S': 'green'}} + ) + + resp = streams.get_records(ShardIterator=iterator_id) + assert len(resp['Records']) == 2 + assert resp['Records'][0]['eventName'] == 'INSERT' + assert resp['Records'][1]['eventName'] == 'INSERT' + From e7b01292e96172a1d716fc80c7df628e00c2b83e Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Thu, 8 Nov 2018 13:57:44 -0500 Subject: [PATCH 6/7] Fix test failures in server mode --- moto/dynamodb2/responses.py | 6 +++++- moto/server.py | 11 +++++++---- tests/test_dynamodbstreams/test_dynamodbstreams.py | 2 +- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/moto/dynamodb2/responses.py b/moto/dynamodb2/responses.py index d2bf99bfe..79dd749e2 100644 --- a/moto/dynamodb2/responses.py +++ b/moto/dynamodb2/responses.py @@ -174,7 +174,11 @@ class DynamoHandler(BaseResponse): throughput = self.body["ProvisionedThroughput"] table = self.dynamodb_backend.update_table_throughput(name, throughput) if 'StreamSpecification' in self.body: - table = self.dynamodb_backend.update_table_streams(name, self.body['StreamSpecification']) + try: + table = self.dynamodb_backend.update_table_streams(name, self.body['StreamSpecification']) + except ValueError: + er = 'com.amazonaws.dynamodb.v20111205#ResourceInUseException' + return self.error(er, 'Cannot enable stream') return dynamo_json_dump(table.describe()) diff --git a/moto/server.py b/moto/server.py index ba2470478..5ad02d383 100644 --- a/moto/server.py +++ b/moto/server.py @@ -80,10 +80,13 @@ class DomainDispatcherApplication(object): region = 'us-east-1' if service == 'dynamodb': - dynamo_api_version = environ['HTTP_X_AMZ_TARGET'].split("_")[1].split(".")[0] - # If Newer API version, use dynamodb2 - if dynamo_api_version > "20111205": - host = "dynamodb2" + if environ['HTTP_X_AMZ_TARGET'].startswith('DynamoDBStreams'): + host = 'dynamodbstreams' + else: + dynamo_api_version = environ['HTTP_X_AMZ_TARGET'].split("_")[1].split(".")[0] + # If Newer API version, use dynamodb2 + if dynamo_api_version > "20111205": + host = "dynamodb2" else: host = "{service}.{region}.amazonaws.com".format( service=service, region=region) diff --git a/tests/test_dynamodbstreams/test_dynamodbstreams.py b/tests/test_dynamodbstreams/test_dynamodbstreams.py index 7e4025626..b60c21053 100644 --- a/tests/test_dynamodbstreams/test_dynamodbstreams.py +++ b/tests/test_dynamodbstreams/test_dynamodbstreams.py @@ -180,7 +180,7 @@ class TestEdges(): assert 'LatestStreamLabel' in resp['TableDescription'] # now try to enable it again - with assert_raises(ValueError): + with assert_raises(conn.exceptions.ResourceInUseException): resp = conn.update_table( TableName='test-streams', StreamSpecification={ From 46c0f8915f913b92190986ba74ea2fb266dd28c9 Mon Sep 17 00:00:00 2001 From: Karl Gutwin Date: Fri, 9 Nov 2018 09:32:19 -0500 Subject: [PATCH 7/7] Fix flake8 failures (boo) --- moto/dynamodb2/models.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/moto/dynamodb2/models.py b/moto/dynamodb2/models.py index d15b9fce5..8187ceaf9 100644 --- a/moto/dynamodb2/models.py +++ b/moto/dynamodb2/models.py @@ -406,8 +406,7 @@ class Table(BaseModel): def set_stream_specification(self, streams): self.stream_specification = streams - if streams and (streams.get('StreamEnabled') - or streams.get('StreamViewType')): + if streams and (streams.get('StreamEnabled') or streams.get('StreamViewType')): self.stream_specification['StreamEnabled'] = True self.latest_stream_label = datetime.datetime.utcnow().isoformat() self.stream_shard = StreamShard(self) @@ -782,9 +781,7 @@ class DynamoDBBackend(BaseBackend): def update_table_streams(self, name, stream_specification): table = self.tables[name] - if ((stream_specification.get('StreamEnabled') - or stream_specification.get('StreamViewType')) - and table.latest_stream_label): + if (stream_specification.get('StreamEnabled') or stream_specification.get('StreamViewType')) and table.latest_stream_label: raise ValueError('Table already has stream enabled') table.set_stream_specification(stream_specification) return table