Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add support for tags to rds snapshots

Browse Source
This commit is contained in:
Jon Beilke 2018-09-21 12:03:13 -05:00
parent 1b8b32a663
commit 6eb490ac78
2 changed files with 127 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
moto/rds2/models.py
View File

@ -463,6 +463,20 @@ class Snapshot(BaseModel):
</DBSnapshot>""") </DBSnapshot>""")
return template.render(snapshot=self, database=self.database) return template.render(snapshot=self, database=self.database)
def get_tags(self):
return self.tags
def add_tags(self, tags):
new_keys = [tag_set['Key'] for tag_set in tags]
self.tags = [tag_set for tag_set in self.tags if tag_set[
'Key'] not in new_keys]
self.tags.extend(tags)
return self.tags
def remove_tags(self, tag_keys):
self.tags = [tag_set for tag_set in self.tags if tag_set[
'Key'] not in tag_keys]
class SecurityGroup(BaseModel): class SecurityGroup(BaseModel):
@ -1037,8 +1051,8 @@ class RDS2Backend(BaseBackend):
if resource_name in self.security_groups: if resource_name in self.security_groups:
return self.security_groups[resource_name].get_tags() return self.security_groups[resource_name].get_tags()
elif resource_type == 'snapshot': # DB Snapshot elif resource_type == 'snapshot': # DB Snapshot
# TODO: Complete call to tags on resource type DB Snapshot if resource_name in self.snapshots:
return [] return self.snapshots[resource_name].get_tags()
elif resource_type == 'subgrp': # DB subnet group elif resource_type == 'subgrp': # DB subnet group
if resource_name in self.subnet_groups: if resource_name in self.subnet_groups:
return self.subnet_groups[resource_name].get_tags() return self.subnet_groups[resource_name].get_tags()

111
tests/test_rds2/test_rds2.py
View File

@ -700,6 +700,117 @@ def test_remove_tags_db():
len(result['TagList']).should.equal(1) len(result['TagList']).should.equal(1)
@mock_rds2
def test_list_tags_snapshot():
conn = boto3.client('rds', region_name='us-west-2')
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:foo')
result['TagList'].should.equal([])
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-with-tags',
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
result = conn.list_tags_for_resource(ResourceName=snapshot['DBSnapshot']['DBSnapshotArn'])
result['TagList'].should.equal([{'Value': 'bar',
'Key': 'foo'},
{'Value': 'bar1',
'Key': 'foo1'}])
@mock_rds2
def test_add_tags_snapshot():
conn = boto3.client('rds', region_name='us-west-2')
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-without-tags',
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-without-tags')
list(result['TagList']).should.have.length_of(2)
conn.add_tags_to_resource(ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-without-tags',
Tags=[
{
'Key': 'foo',
'Value': 'fish',
},
{
'Key': 'foo2',
'Value': 'bar2',
},
])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-without-tags')
list(result['TagList']).should.have.length_of(3)
@mock_rds2
def test_remove_tags_snapshot():
conn = boto3.client('rds', region_name='us-west-2')
conn.create_db_instance(DBInstanceIdentifier='db-primary-1',
AllocatedStorage=10,
Engine='postgres',
DBName='staging-postgres',
DBInstanceClass='db.m1.small',
MasterUsername='root',
MasterUserPassword='hunter2',
Port=1234,
DBSecurityGroups=["my_sg"])
snapshot = conn.create_db_snapshot(DBInstanceIdentifier='db-primary-1',
DBSnapshotIdentifier='snapshot-with-tags',
Tags=[
{
'Key': 'foo',
'Value': 'bar',
},
{
'Key': 'foo1',
'Value': 'bar1',
},
])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-with-tags')
list(result['TagList']).should.have.length_of(2)
conn.remove_tags_from_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-with-tags', TagKeys=['foo'])
result = conn.list_tags_for_resource(
ResourceName='arn:aws:rds:us-west-2:1234567890:snapshot:snapshot-with-tags')
len(result['TagList']).should.equal(1)
@mock_rds2 @mock_rds2
def test_add_tags_option_group(): def test_add_tags_option_group():
conn = boto3.client('rds', region_name='us-west-2') conn = boto3.client('rds', region_name='us-west-2')