[WIP] Resource tagging (#1332)
* Ran scaffold * Made a start on get_resources. * Added more EC2 to tag list * More stuff * Added more methods * Added S3 region name * Added values test * Some final touchups
This commit is contained in:
Terry Cain
GitHub
parent
474023f9a1
commit
8ee05e22af
9 changed files with 877 additions and 0 deletions
0
tests/test_resourcegroupstaggingapi/__init__.py
Normal file
0
tests/test_resourcegroupstaggingapi/__init__.py
Normal file
|
|
@ -0,0 +1,226 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
import boto3
|
||||
import sure # noqa
|
||||
from moto import mock_resourcegroupstaggingapi, mock_s3, mock_ec2
|
||||
|
||||
|
||||
@mock_s3
|
||||
@mock_resourcegroupstaggingapi
|
||||
def test_get_resources_s3():
|
||||
# Tests pagination
|
||||
s3_client = boto3.client('s3', region_name='eu-central-1')
|
||||
|
||||
# Will end up having key1,key2,key3,key4
|
||||
response_keys = set()
|
||||
|
||||
# Create 4 buckets
|
||||
for i in range(1, 5):
|
||||
i_str = str(i)
|
||||
s3_client.create_bucket(Bucket='test_bucket' + i_str)
|
||||
s3_client.put_bucket_tagging(
|
||||
Bucket='test_bucket' + i_str,
|
||||
Tagging={'TagSet': [{'Key': 'key' + i_str, 'Value': 'value' + i_str}]}
|
||||
)
|
||||
response_keys.add('key' + i_str)
|
||||
|
||||
rtapi = boto3.client('resourcegroupstaggingapi', region_name='eu-central-1')
|
||||
resp = rtapi.get_resources(ResourcesPerPage=2)
|
||||
for resource in resp['ResourceTagMappingList']:
|
||||
response_keys.remove(resource['Tags'][0]['Key'])
|
||||
|
||||
response_keys.should.have.length_of(2)
|
||||
|
||||
resp = rtapi.get_resources(
|
||||
ResourcesPerPage=2,
|
||||
PaginationToken=resp['PaginationToken']
|
||||
)
|
||||
for resource in resp['ResourceTagMappingList']:
|
||||
response_keys.remove(resource['Tags'][0]['Key'])
|
||||
|
||||
response_keys.should.have.length_of(0)
|
||||
|
||||
|
||||
@mock_ec2
|
||||
@mock_resourcegroupstaggingapi
|
||||
def test_get_resources_ec2():
|
||||
client = boto3.client('ec2', region_name='eu-central-1')
|
||||
|
||||
instances = client.run_instances(
|
||||
ImageId='ami-123',
|
||||
MinCount=1,
|
||||
MaxCount=1,
|
||||
InstanceType='t2.micro',
|
||||
TagSpecifications=[
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG1',
|
||||
'Value': 'MY_VALUE1',
|
||||
},
|
||||
{
|
||||
'Key': 'MY_TAG2',
|
||||
'Value': 'MY_VALUE2',
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG3',
|
||||
'Value': 'MY_VALUE3',
|
||||
},
|
||||
]
|
||||
},
|
||||
],
|
||||
)
|
||||
instance_id = instances['Instances'][0]['InstanceId']
|
||||
image_id = client.create_image(Name='testami', InstanceId=instance_id)['ImageId']
|
||||
|
||||
client.create_tags(
|
||||
Resources=[image_id],
|
||||
Tags=[{'Key': 'ami', 'Value': 'test'}]
|
||||
)
|
||||
|
||||
rtapi = boto3.client('resourcegroupstaggingapi', region_name='eu-central-1')
|
||||
resp = rtapi.get_resources()
|
||||
# Check we have 1 entry for Instance, 1 Entry for AMI
|
||||
resp['ResourceTagMappingList'].should.have.length_of(2)
|
||||
|
||||
# 1 Entry for AMI
|
||||
resp = rtapi.get_resources(ResourceTypeFilters=['ec2:image'])
|
||||
resp['ResourceTagMappingList'].should.have.length_of(1)
|
||||
resp['ResourceTagMappingList'][0]['ResourceARN'].should.contain('image/')
|
||||
|
||||
# As were iterating the same data, this rules out that the test above was a fluke
|
||||
resp = rtapi.get_resources(ResourceTypeFilters=['ec2:instance'])
|
||||
resp['ResourceTagMappingList'].should.have.length_of(1)
|
||||
resp['ResourceTagMappingList'][0]['ResourceARN'].should.contain('instance/')
|
||||
|
||||
# Basic test of tag filters
|
||||
resp = rtapi.get_resources(TagFilters=[{'Key': 'MY_TAG1', 'Values': ['MY_VALUE1', 'some_other_value']}])
|
||||
resp['ResourceTagMappingList'].should.have.length_of(1)
|
||||
resp['ResourceTagMappingList'][0]['ResourceARN'].should.contain('instance/')
|
||||
|
||||
|
||||
@mock_ec2
|
||||
@mock_resourcegroupstaggingapi
|
||||
def test_get_tag_keys_ec2():
|
||||
client = boto3.client('ec2', region_name='eu-central-1')
|
||||
|
||||
client.run_instances(
|
||||
ImageId='ami-123',
|
||||
MinCount=1,
|
||||
MaxCount=1,
|
||||
InstanceType='t2.micro',
|
||||
TagSpecifications=[
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG1',
|
||||
'Value': 'MY_VALUE1',
|
||||
},
|
||||
{
|
||||
'Key': 'MY_TAG2',
|
||||
'Value': 'MY_VALUE2',
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG3',
|
||||
'Value': 'MY_VALUE3',
|
||||
},
|
||||
]
|
||||
},
|
||||
],
|
||||
)
|
||||
|
||||
rtapi = boto3.client('resourcegroupstaggingapi', region_name='eu-central-1')
|
||||
resp = rtapi.get_tag_keys()
|
||||
|
||||
resp['TagKeys'].should.contain('MY_TAG1')
|
||||
resp['TagKeys'].should.contain('MY_TAG2')
|
||||
resp['TagKeys'].should.contain('MY_TAG3')
|
||||
|
||||
# TODO test pagenation
|
||||
|
||||
|
||||
@mock_ec2
|
||||
@mock_resourcegroupstaggingapi
|
||||
def test_get_tag_values_ec2():
|
||||
client = boto3.client('ec2', region_name='eu-central-1')
|
||||
|
||||
client.run_instances(
|
||||
ImageId='ami-123',
|
||||
MinCount=1,
|
||||
MaxCount=1,
|
||||
InstanceType='t2.micro',
|
||||
TagSpecifications=[
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG1',
|
||||
'Value': 'MY_VALUE1',
|
||||
},
|
||||
{
|
||||
'Key': 'MY_TAG2',
|
||||
'Value': 'MY_VALUE2',
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG3',
|
||||
'Value': 'MY_VALUE3',
|
||||
},
|
||||
]
|
||||
},
|
||||
],
|
||||
)
|
||||
client.run_instances(
|
||||
ImageId='ami-123',
|
||||
MinCount=1,
|
||||
MaxCount=1,
|
||||
InstanceType='t2.micro',
|
||||
TagSpecifications=[
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG1',
|
||||
'Value': 'MY_VALUE4',
|
||||
},
|
||||
{
|
||||
'Key': 'MY_TAG2',
|
||||
'Value': 'MY_VALUE5',
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
'ResourceType': 'instance',
|
||||
'Tags': [
|
||||
{
|
||||
'Key': 'MY_TAG3',
|
||||
'Value': 'MY_VALUE6',
|
||||
},
|
||||
]
|
||||
},
|
||||
],
|
||||
)
|
||||
|
||||
rtapi = boto3.client('resourcegroupstaggingapi', region_name='eu-central-1')
|
||||
resp = rtapi.get_tag_values(Key='MY_TAG1')
|
||||
|
||||
resp['TagValues'].should.contain('MY_VALUE1')
|
||||
resp['TagValues'].should.contain('MY_VALUE4')
|
||||
|
||||
# TODO test pagenation
|
||||
24
tests/test_resourcegroupstaggingapi/test_server.py
Normal file
24
tests/test_resourcegroupstaggingapi/test_server.py
Normal file
|
|
@ -0,0 +1,24 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
import sure # noqa
|
||||
|
||||
import moto.server as server
|
||||
|
||||
'''
|
||||
Test the different server responses
|
||||
'''
|
||||
|
||||
|
||||
def test_resourcegroupstaggingapi_list():
|
||||
backend = server.create_backend_app("resourcegroupstaggingapi")
|
||||
test_client = backend.test_client()
|
||||
# do test
|
||||
|
||||
headers = {
|
||||
'X-Amz-Target': 'ResourceGroupsTaggingAPI_20170126.GetResources',
|
||||
'X-Amz-Date': '20171114T234623Z'
|
||||
}
|
||||
resp = test_client.post('/', headers=headers, data='{}')
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert b'ResourceTagMappingList' in resp.data
|
||||
Loading…
Add table
Add a link
Reference in a new issue