Merge pull request #16 from bblommers/bugfix/13

Fixes #13

.travis.yml

@@ -4,8 +4,9 @@ sudo: true
 python:
 - 3.6
 - 3.7
+- 3.8
 env:
-- CONNECT_TO_AWS=NO
+- CONNECT_TO_AWS=FALSE
 before_install:
 - sudo apt-get install -y ruby-coveralls
 install:

Makefile

@@ -18,7 +18,7 @@ venv_lint: venv/bin/activate
 
 venv_test: venv/bin/activate
 	. venv/bin/activate ; \
-	venv/bin/pytest -s
+	venv/bin/pytest -sv
 
 lint:
 	flake8

requirements-to-freeze.txt

@@ -10,9 +10,8 @@ pytest-pythonpath==0.7.3
 
 # MotoServer requirements
 flask
-netifaces
 aiohttp
 
 # Functional requirements
-moto==1.3.15.dev142
+moto==1.3.15.dev479
 tenacity==5.1.4

setup.cfg

@@ -12,7 +12,8 @@ addopts =
     --cov-branch
 env =
     # OPTIONS: TRUE | FALSE
-    # YES: Runs each test in-memory first, and then runs it again against AWS
+    # FALSE: Runs each test in-memory only
+    # TRUE: Runs each test against AWS
     D:CONNECT_TO_AWS=FALSE
 
 [pycodestyle]

src/migrator/steps/AddIndexStep.py

@@ -27,6 +27,10 @@ def execute(self):
             base_name = self._get_base_name(previous_table_name)
             new_table_name = f"{base_name}_V{self._version}"
             previous_table = self.aws_utils.describe_table(previous_table_name)['Table']
+            if 'StreamSpecification' not in previous_table:
+                # Alter table to add stream
+                self.aws_utils.update_table(DynamoDButilities.get_stream_props(previous_table_name))
+                previous_table = self.aws_utils.describe_table(previous_table_name)['Table']
             new_table = DynamoDButilities.get_table_creation_details(previous_table, new_table_name,
                                                                      local_indexes=self._properties['LocalSecondaryIndexes'],
                                                                      attr_definitions=self._properties['AttributeDefinitions'])
@@ -36,10 +40,14 @@ def execute(self):
             # Create Role
             created_policy, created_role = self.aws_utils.create_iam_items(created_table, previous_table)
             # Create Lambda
-            func = self.aws_utils.create_aws_lambda(created_role, created_table['TableName'])
+            func = self.aws_utils.create_aws_lambda(created_role,
+                                                    old_table=previous_table_name,
+                                                    new_table=created_table['TableName'])
             # Create stream
             self.aws_utils.create_event_source_mapping(stream_arn=previous_table['LatestStreamArn'],
                                                        function_arn=func['FunctionArn'])
+            # Update existing data
+            self.aws_utils.update_data(previous_table_name, key_schema=previous_table['KeySchema'])
         return created_table
 
     def _get_base_name(self, name):

src/migrator/steps/CreateTableStep.py

@@ -10,5 +10,4 @@ def __init__(self, identifier, version, properties):
         self.aws_utils = AwsUtilities(self._identifier, version=self._version)
 
     def execute(self):
-        self._properties['StreamSpecification'] = {'StreamEnabled': True, 'StreamViewType': 'NEW_AND_OLD_IMAGES'}
         return self.aws_utils.create_table_if_not_exists(self._properties)

src/migrator/utilities/DynamoDButilities.py

@@ -1,3 +1,4 @@
+import boto3
 from migrator.utilities.Utilities import metadata_table_name
 
 _accepted_table_properties = ['AttributeDefinitions',
@@ -14,11 +15,16 @@
 class DynamoDButilities:
 
     def __init__(self, identifier, version):
-        from migrator.utilities.AwsUtilities import _dynamodb
-        self._dynamodb = _dynamodb
+        self._dynamodb = boto3.client('dynamodb')
         self._identifier = identifier
         self._version = str(version)
 
+    @staticmethod
+    def get_stream_props(table_name):
+        return {'TableName': table_name,
+                'StreamSpecification': {'StreamEnabled': True,
+                                        'StreamViewType': 'NEW_AND_OLD_IMAGES'}}
+
     @staticmethod
     def get_table_creation_details(existing_table: dict, new_table_name: str,
                                    local_indexes: [dict], attr_definitions: [dict]):
@@ -66,6 +72,9 @@ def add_mapping(self, uuid):
     def add_function(self, arn):
         self._set_operation("ADD", "functions", arn)
 
+    def add_attr_name(self, attr_name):
+        self._set_operation("ADD", "attribute_name", attr_name)
+
     def remove_table(self, name):
         self._set_operation("DELETE", "tables", name)
 
@@ -84,6 +93,9 @@ def remove_function(self, arn):
     def get_created_tables(self, version=None):
         return self.get_attr("tables", version=version)
 
+    def get_created_attr(self):
+        return self.get_attr("attribute_name")
+
     def get_created_policies(self):
         return self.get_attr("policies")
 

src/migrator/utilities/IAMutilities.py

@@ -11,6 +11,12 @@
             "dynamodb:UpdateItem"
         ],
         "Resource": "$newtable"
+    }, {
+        "Effect": "Allow",
+        "Action": [
+            "dynamodb:UpdateItem"
+        ],
+        "Resource": "$oldtable"
     }, {
         "Effect": "Allow",
         "Action": [
@@ -19,7 +25,7 @@
             "dynamodb:ListStreams",
             "dynamodb:GetRecords"
         ],
-        "Resource": "$oldtable"
+        "Resource": "$oldtablestream"
     }, {
         "Effect": "Allow",
         "Action": "logs:*",

src/migrator/utilities/LambdaUtilities.py

@@ -6,27 +6,39 @@
 
 lambda_code = Template("""import boto3
 import json
+from datetime import datetime
 
 dynamodb = boto3.client('dynamodb')
+old_table_name = "$oldtable"
 table_name = "$newtable"
+unique_attr = "$uniqueattr"
 
 
 def copy(event, context):
     for record in event['Records']:
+        key = record['dynamodb']['Keys']
         if record['eventName'] == 'REMOVE':
             response = dynamodb.delete_item(TableName=table_name,
-                                 Key=record['dynamodb']['Keys'])
+                                 Key=key)
         if record['eventName'] == 'INSERT' or record['eventName'] == 'MODIFY':
-            response = dynamodb.put_item(TableName=table_name,
-                              Item=record['dynamodb']['NewImage'])
+            item = record['dynamodb']['NewImage']
+            if unique_attr in item:
+                del item[unique_attr]
+                dynamodb.put_item(TableName=table_name, Item=item)
+            else:
+                dynamodb.update_item(TableName=old_table_name,
+                                     Key=key,
+                                     UpdateExpression="set #attr = :val",
+                                     ExpressionAttributeNames={'#attr': unique_attr},
+                                     ExpressionAttributeValues={':val': {'S': str(datetime.today())}})
     return {
         'statusCode': 200
     }
 """)
 
 
-def get_zipfile(table_name):
-    lambda_content = lambda_code.substitute(newtable=table_name)
+def get_zipfile(old_table, new_table, unique_attr):
+    lambda_content = lambda_code.substitute(oldtable=old_table, newtable=new_table, uniqueattr=unique_attr)
     return zip(lambda_content)
 
 

tests/conftest.py

@@ -0,0 +1,21 @@
+import pytest
+from settings import CONNECT_TO_AWS
+from mock_wrapper import log_target, dynamodb, patch_boto, get_moto_services, verify_everything_is_deleted
+
+
+@pytest.fixture()
+def dynamodb_server_mode():
+    # Same behaviour as the #mock_server_mode decorator
+    # Used as a fixture, to ensure it plays nice with other fixtures (such as parametrize)
+    if CONNECT_TO_AWS:
+        log_target("AWS")
+        yield dynamodb
+    else:
+        log_target("MOCK SERVER MODE")
+        patch_boto()
+        moto_services = get_moto_services(['dynamodb', 'lambda', 'iam'])
+
+        yield moto_services['dynamodb']
+        verify_everything_is_deleted(dynamodb=moto_services['dynamodb'],
+                                     lmbda=moto_services['lambda'],
+                                     iam=moto_services['iam'])

tests/migration_scripts/add_index/table_copy_items_v1.py

@@ -1,9 +1,10 @@
 #!/usr/bin/python
 
 from migrator.dynamodb_migrator import Migrator
+from uuid import uuid4
 
 
-table_name = 'customers'
+table_name = str(uuid4())
 migrator = Migrator(identifier="copy_items")
 
 

tests/migration_scripts/add_index/table_copy_items_v2.py

@@ -1,9 +1,9 @@
 #!/usr/bin/python
 
 from migrator.dynamodb_migrator import Migrator
+from migration_scripts.add_index.table_copy_items_v1 import table_name
 
 
-table_name = 'customers'
 migrator = Migrator(identifier="copy_items")
 
 
@@ -15,8 +15,7 @@
                             {'AttributeName': 'last_name', 'KeyType': 'RANGE'}],
                  BillingMode='PAY_PER_REQUEST')
 def v1(created_table):
-    assert created_table['TableName'] == table_name
-    assert created_table['TableStatus'] == 'ACTIVE'
+    pass
 
 
 @migrator.version(2)
@@ -26,5 +25,4 @@ def v1(created_table):
                                                             {'AttributeName': 'postcode', 'KeyType': 'RANGE'}],
                                               'Projection': {'ProjectionType': 'ALL'}}])
 def v2(created_table):
-    assert created_table['TableName'] == f"{table_name}_V2"
-    assert created_table['TableStatus'] == 'ACTIVE'
+    pass

tests/migration_scripts/add_index/table_stream_items_v1.py

@@ -1,9 +1,10 @@
 #!/usr/bin/python
 
 from migrator.dynamodb_migrator import Migrator
+from uuid import uuid4
 
 
-table_name = 'customers'
+table_name = str(uuid4())
 migrator = Migrator(identifier="table_stream_test")
 
 

tests/migration_scripts/add_index/table_stream_items_v2.py

@@ -1,9 +1,9 @@
 #!/usr/bin/python
 
 from migrator.dynamodb_migrator import Migrator
+from migration_scripts.add_index.table_stream_items_v1 import table_name
 
 
-table_name = 'customers'
 migrator = Migrator(identifier="table_stream_test")
 
 

tests/migration_scripts/add_index/table_stream_items_v3.py

@@ -1,9 +1,9 @@
 #!/usr/bin/python
 
 from migrator.dynamodb_migrator import Migrator
+from migration_scripts.add_index.table_stream_items_v1 import table_name
 
 
-table_name = 'customers'
 migrator = Migrator(identifier="table_stream_test")