Check signed url uploads

test.py

"""This scripts tests that signed url uploads are supported by an S3 server.

First install dependencies with:

pip install requests boto3

Then run the script with the following env variables set:

HOST=http://s3.hostname.net AWS_SECRET_ACCESS_KEY=XXX AWS_ACCESS_KEY_ID=XXX BUCKET=bucket REGION=region python test.py

"""
import os
import requests
import boto3
from botocore.client import Config
from botocore.exceptions import ClientError

TEST_MD5="T0rMXYxx9fvwTazgC1NgyA=="
bucket = os.environ.get("BUCKET")
host = os.environ.get("HOST")
region = os.environ.get("REGION", "wandb-local")
assert os.environ.get("AWS_SECRET_ACCESS_KEY"), "Missing s3 credential key, set AWS_SECRET_ACCESS_KEY"
assert os.environ.get("AWS_ACCESS_KEY_ID"), "Missing s3 access key, set AWS_ACCESS_KEY_ID"
assert bucket, "Missing s3 bucket, set BUCKET"
assert host, "Missing s3 host, set HOST"

print("Connecting to: {}".format(host))
s3_client = boto3.client("s3", region, endpoint_url=host,
                         config=Config(s3={'addressing_style': 'path'}, signature_version='s3v4'))

url = s3_client.generate_presigned_url('put_object', 
                                       Params={'Bucket': bucket, 'Key': "test.txt", "ContentType": "text/plain",
                                               "Metadata": {"md5": TEST_MD5},
                                               "ContentMD5": TEST_MD5}, 
                                       HttpMethod='PUT', ExpiresIn=3600)
print("Attempting HEAD to verify 404")
failed = False
try:
  res2 = s3_client.head_object(Bucket=bucket, Key="not-existing-key.txt")
except ClientError as e:
  if e.response["Error"]["Code"] == "404":
    failed = True
  else:
    print(e.response)
assert failed, "Head object didn't return a 404 error"

print("Attempting PUT: {}".format(url))
res = requests.put(url, data=b"test test", headers={"x-amz-meta-md5": TEST_MD5, "Content-Type": "text/plain", "Content-MD5": TEST_MD5})
if res.status_code == 200:
  print("Success!  Signed urls work with this server")
  exit(0)
else:
  print("!!! Request failed:")
  print("!!!   code: {}".format(res.status_code))
  print("!!!   body: {}".format(res.content.decode("utf8")))
  print("!!!   headers: {}".format(res.headers))
  exit(1)