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

Merge pull request #1012 from eric-weaver/s3-object-tagging

Browse Source 
Implement S3 object tagging
This commit is contained in:
Jack Danger 2017-07-17 11:57:55 -07:00 committed by GitHub
commit a718ff2029
4 changed files with 175 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
moto/s3/models.py
View File

@ -11,7 +11,7 @@ import six
from bisect import insort
from moto.core import BaseBackend, BaseModel
from moto.core.utils import iso_8601_datetime_with_milliseconds, rfc_1123_datetime
from .exceptions import BucketAlreadyExists, MissingBucket, InvalidPart, EntityTooSmall
from .exceptions import BucketAlreadyExists, MissingBucket, InvalidPart, EntityTooSmall, MissingKey
from .utils import clean_key_name, _VersionedKeyStore
UPLOAD_ID_BYTES = 43
@ -43,6 +43,7 @@ class FakeKey(BaseModel):
self._etag = etag
self._version_id = version_id
self._is_versioned = is_versioned
self._tagging = FakeTagging()
@property
def version_id(self):
@ -59,6 +60,9 @@ class FakeKey(BaseModel):
self._metadata = {}
self._metadata.update(metadata)
def set_tagging(self, tagging):
self._tagging = tagging
def set_storage_class(self, storage_class):
self._storage_class = storage_class
@ -103,6 +107,10 @@ class FakeKey(BaseModel):
def metadata(self):
return self._metadata
@property
def tagging(self):
return self._tagging
@property
def response_dict(self):
res = {
@ -253,6 +261,25 @@ def get_canned_acl(acl):
return FakeAcl(grants=grants)
class FakeTagging(BaseModel):
def __init__(self, tag_set=None):
self.tag_set = tag_set or FakeTagSet()
class FakeTagSet(BaseModel):
def __init__(self, tags=None):
self.tags = tags or []
class FakeTag(BaseModel):
def __init__(self, key, value=None):
self.key = key
self.value = value
class LifecycleRule(BaseModel):
def __init__(self, id=None, prefix=None, status=None, expiration_days=None,
@ -475,6 +502,13 @@ class S3Backend(BaseBackend):
else:
return None
def set_key_tagging(self, bucket_name, key_name, tagging):
key = self.get_key(bucket_name, key_name)
if key is None:
raise MissingKey(key_name)
key.set_tagging(tagging)
return key
def initiate_multipart(self, bucket_name, key_name, metadata):
bucket = self.get_bucket(bucket_name)
new_multipart = FakeMultipart(key_name, metadata)

49
moto/s3/responses.py
View File

@ -14,7 +14,7 @@ from moto.s3bucket_path.utils import bucket_name_from_url as bucketpath_bucket_n
from .exceptions import BucketAlreadyExists, S3ClientError, MissingKey, InvalidPartOrder
from .models import s3_backend, get_canned_acl, FakeGrantee, FakeGrant, FakeAcl, FakeKey
from .models import s3_backend, get_canned_acl, FakeGrantee, FakeGrant, FakeAcl, FakeKey, FakeTagging, FakeTagSet, FakeTag
from .utils import bucket_name_from_url, metadata_from_headers
from xml.dom import minidom
@ -520,6 +520,9 @@ class ResponseObject(_TemplateEnvironmentMixin):
if 'acl' in query:
template = self.response_template(S3_OBJECT_ACL_RESPONSE)
return 200, response_headers, template.render(obj=key)
if 'tagging' in query:
template = self.response_template(S3_OBJECT_TAGGING_RESPONSE)
return 200, response_headers, template.render(obj=key)
response_headers.update(key.metadata)
response_headers.update(key.response_dict)
@ -556,6 +559,7 @@ class ResponseObject(_TemplateEnvironmentMixin):
storage_class = request.headers.get('x-amz-storage-class', 'STANDARD')
acl = self._acl_from_headers(request.headers)
tagging = self._tagging_from_headers(request.headers)
if 'acl' in query:
key = self.backend.get_key(bucket_name, key_name)
@ -563,6 +567,11 @@ class ResponseObject(_TemplateEnvironmentMixin):
key.set_acl(acl)
return 200, response_headers, ""
if 'tagging' in query:
tagging = self._tagging_from_xml(body)
self.backend.set_key_tagging(bucket_name, key_name, tagging)
return 200, response_headers, ""
if 'x-amz-copy-source' in request.headers:
# Copy key
src_key_parsed = urlparse(request.headers.get("x-amz-copy-source"))
@ -596,6 +605,7 @@ class ResponseObject(_TemplateEnvironmentMixin):
new_key.set_metadata(metadata)
new_key.set_acl(acl)
new_key.website_redirect_location = request.headers.get('x-amz-website-redirect-location')
new_key.set_tagging(tagging)
template = self.response_template(S3_OBJECT_RESPONSE)
response_headers.update(new_key.response_dict)
@ -655,6 +665,30 @@ class ResponseObject(_TemplateEnvironmentMixin):
else:
return None
def _tagging_from_headers(self, headers):
if headers.get('x-amz-tagging'):
parsed_header = parse_qs(headers['x-amz-tagging'], keep_blank_values=True)
tags = []
for tag in parsed_header.items():
tags.append(FakeTag(tag[0], tag[1][0]))
tag_set = FakeTagSet(tags)
tagging = FakeTagging(tag_set)
return tagging
else:
return FakeTagging()
def _tagging_from_xml(self, xml):
parsed_xml = xmltodict.parse(xml)
tags = []
for tag in parsed_xml['Tagging']['TagSet']['Tag']:
tags.append(FakeTag(tag['Key'], tag['Value']))
tag_set = FakeTagSet(tags)
tagging = FakeTagging(tag_set)
return tagging
def _key_response_delete(self, bucket_name, query, key_name, headers):
if query.get('uploadId'):
upload_id = query['uploadId'][0]
@ -968,6 +1002,19 @@ S3_OBJECT_ACL_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
</AccessControlList>
</AccessControlPolicy>"""
S3_OBJECT_TAGGING_RESPONSE = """\
<?xml version="1.0" encoding="UTF-8"?>
<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<TagSet>
{% for tag in obj.tagging.tag_set.tags %}
<Tag>
<Key>{{ tag.key }}</Key>
<Value>{{ tag.value }}</Value>
</Tag>
{% endfor %}
</TagSet>
</Tagging>"""
S3_OBJECT_COPY_RESPONSE = """\
<CopyObjectResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
<ETag>{{ key.etag }}</ETag>

0
tests/test_s3/__init__.py Normal file
View File

92
tests/test_s3/test_s3.py
View File

@ -1340,6 +1340,98 @@ def test_boto3_multipart_etag():
resp['ETag'].should.equal(EXPECTED_ETAG)
@mock_s3
def test_boto3_put_object_with_tagging():
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'mybucket'
key = 'key-with-tags'
s3.create_bucket(Bucket=bucket_name)
s3.put_object(
Bucket=bucket_name,
Key=key,
Body='test',
Tagging='foo=bar',
)
resp = s3.get_object_tagging(Bucket=bucket_name, Key=key)
resp['TagSet'].should.contain({'Key': 'foo', 'Value': 'bar'})
@mock_s3
def test_boto3_put_object_tagging():
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'mybucket'
key = 'key-with-tags'
s3.create_bucket(Bucket=bucket_name)
with assert_raises(ClientError) as err:
s3.put_object_tagging(
Bucket=bucket_name,
Key=key,
Tagging={'TagSet': [
{'Key': 'item1', 'Value': 'foo'},
{'Key': 'item2', 'Value': 'bar'},
]}
)
e = err.exception
e.response['Error'].should.equal({
'Code': 'NoSuchKey',
'Message': 'The specified key does not exist.',
'RequestID': '7a62c49f-347e-4fc4-9331-6e8eEXAMPLE',
})
s3.put_object(
Bucket=bucket_name,
Key=key,
Body='test'
)
resp = s3.put_object_tagging(
Bucket=bucket_name,
Key=key,
Tagging={'TagSet': [
{'Key': 'item1', 'Value': 'foo'},
{'Key': 'item2', 'Value': 'bar'},
]}
)
resp['ResponseMetadata']['HTTPStatusCode'].should.equal(200)
@mock_s3
def test_boto3_get_object_tagging():
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'mybucket'
key = 'key-with-tags'
s3.create_bucket(Bucket=bucket_name)
s3.put_object(
Bucket=bucket_name,
Key=key,
Body='test'
)
resp = s3.get_object_tagging(Bucket=bucket_name, Key=key)
resp['TagSet'].should.have.length_of(0)
resp = s3.put_object_tagging(
Bucket=bucket_name,
Key=key,
Tagging={'TagSet': [
{'Key': 'item1', 'Value': 'foo'},
{'Key': 'item2', 'Value': 'bar'},
]}
)
resp = s3.get_object_tagging(Bucket=bucket_name, Key=key)
resp['TagSet'].should.have.length_of(2)
resp['TagSet'].should.contain({'Key': 'item1', 'Value': 'foo'})
resp['TagSet'].should.contain({'Key': 'item2', 'Value': 'bar'})
@mock_s3
def test_boto3_list_object_versions():
s3 = boto3.client('s3', region_name='us-east-1')