This commit is contained in:
Daniel Wallace 2020-04-21 21:43:32 -05:00
parent 4b0ba73204
commit 80b27a6b93
No known key found for this signature in database
GPG Key ID: 9FD9048A8BD617F2
2 changed files with 24 additions and 17 deletions

View File

@ -797,12 +797,12 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
if hasattr(request, "form"): if hasattr(request, "form"):
# Not HTTPretty # Not HTTPretty
form = request.form form = request.form
elif request.headers.get('Content-Type').startswith('multipart/form-data'): elif request.headers.get("Content-Type").startswith("multipart/form-data"):
request = Request.from_values( request = Request.from_values(
input_stream=six.BytesIO(request.body), input_stream=six.BytesIO(request.body),
content_length=request.headers['Content-Length'], content_length=request.headers["Content-Length"],
content_type=request.headers['Content-Type'], content_type=request.headers["Content-Type"],
method='POST', method="POST",
) )
form = request.form form = request.form
else: else:

View File

@ -4433,30 +4433,37 @@ def test_s3_config_dict():
@mock_s3 @mock_s3
def test_creating_presigned_post(): def test_creating_presigned_post():
bucket = 'presigned-test' bucket = "presigned-test"
s3 = boto3.client('s3', region_name='us-east-1') s3 = boto3.client("s3", region_name="us-east-1")
s3.create_bucket(Bucket=bucket) s3.create_bucket(Bucket=bucket)
success_url = 'http://localhost/completed' success_url = "http://localhost/completed"
fdata = b'test data\n' fdata = b"test data\n"
file_uid = uuid.uuid4() file_uid = uuid.uuid4()
conditions = [ conditions = [
{"Content-Type": 'text/plain'}, {"Content-Type": "text/plain"},
{"x-amz-server-side-encryption": "AES256"}, {"x-amz-server-side-encryption": "AES256"},
{'success_action_redirect': success_url}, {"success_action_redirect": success_url},
] ]
conditions.append(["content-length-range", 1, 30]) conditions.append(["content-length-range", 1, 30])
data = s3.generate_presigned_post( data = s3.generate_presigned_post(
Bucket=bucket, Bucket=bucket,
Key='{file_uid}.txt'.format(file_uid=file_uid), Key="{file_uid}.txt".format(file_uid=file_uid),
Fields={ Fields={
'content-type': 'text/plain', "content-type": "text/plain",
'success_action_redirect': success_url, "success_action_redirect": success_url,
'x-amz-server-side-encryption': 'AES256' "x-amz-server-side-encryption": "AES256",
}, },
Conditions=conditions, Conditions=conditions,
ExpiresIn=1000, ExpiresIn=1000,
) )
resp = requests.post(data['url'], data=data['fields'], files={'file': fdata}, allow_redirects=False) resp = requests.post(
assert resp.headers['Location'] == success_url data["url"], data=data["fields"], files={"file": fdata}, allow_redirects=False
)
assert resp.headers["Location"] == success_url
assert resp.status_code == 303 assert resp.status_code == 303
assert s3.get_object(Bucket=bucket, Key='{file_uid}.txt'.format(file_uid=file_uid))['Body'].read() == fdata assert (
s3.get_object(Bucket=bucket, Key="{file_uid}.txt".format(file_uid=file_uid))[
"Body"
].read()
== fdata
)