added new merge_taglists() to moto.core.utils for merging lists of tags with precedence (ie. during rds2.create_snapshot)

moto/core/utils.py

@@ -286,3 +286,12 @@ def amzn_request_id(f):
         return status, headers, body
 
     return _wrapper
+
+
+def merge_taglists(taglist_a, taglist_b):
+    ''' Merges two tag lists into a single tag list with Keys in the second list taking precedence'''
+    tags_a = {t['Key']:t for t in taglist_a}
+    tags_b = {t['Key']:t for t in taglist_b}
+    merged_tags = tags_a.copy()
+    merged_tags.update(tags_b)
+    return merged_tags.values()

moto/rds2/models.py

@@ -13,6 +13,7 @@ from moto.compat import OrderedDict
 from moto.core import BaseBackend, BaseModel
 from moto.core.utils import get_random_hex
 from moto.core.utils import iso_8601_datetime_with_milliseconds
+from moto.core.utils import merge_taglists
 from moto.ec2.models import ec2_backends
 from .exceptions import (RDSClientError,
                          DBInstanceNotFoundError,
@@ -697,8 +698,8 @@ class RDS2Backend(BaseBackend):
             raise DBSnapshotAlreadyExistsError(db_snapshot_identifier)
         if len(self.snapshots) >= int(os.environ.get('MOTO_RDS_SNAPSHOT_LIMIT', '100')):
             raise SnapshotQuotaExceededError()
-        if not database.copy_tags_to_snapshot:
-            tags = None
+        if database.copy_tags_to_snapshot:
+            tags = merge_taglists(database.tags, tags)
         snapshot = Snapshot(database, db_snapshot_identifier, tags)
         self.snapshots[db_snapshot_identifier] = snapshot
         return snapshot

tests/test_core/test_utils.py

@@ -3,7 +3,7 @@ from __future__ import unicode_literals
 import sure  # noqa
 from freezegun import freeze_time
 
-from moto.core.utils import camelcase_to_underscores, underscores_to_camelcase, unix_time
+from moto.core.utils import camelcase_to_underscores, underscores_to_camelcase, unix_time, merge_taglists
 
 
 def test_camelcase_to_underscores():
@@ -28,3 +28,28 @@ def test_underscores_to_camelcase():
 @freeze_time("2015-01-01 12:00:00")
 def test_unix_time():
     unix_time().should.equal(1420113600.0)
+
+
+def test_merge_taglists():
+    taglist_a = [
+        {
+            'Key': 'foo',
+            'Value': 'bar',
+        },
+        {
+            'Key': 'foo1',
+            'Value': 'bar1',
+        },
+    ]
+    taglist_b = [
+        {
+            'Key': 'foo1',
+            'Value': 'bar1b',
+        },
+    ]
+    taglist_merged = merge_taglists(taglist_a, taglist_b)
+    len(taglist_merged).should.equal(2)
+    tag_foo = [t for t in taglist_merged if t['Key']=='foo']
+    tag_foo1 = [t for t in taglist_merged if t['Key']=='foo1']
+    tag_foo[0].should.equal({'Key': 'foo','Value': 'bar',})
+    tag_foo1[0].should.equal({'Key': 'foo1','Value': 'bar1b',})