1022
1022
amz_headers={"acl": "public"})
1023
1023
return deferred.addCallback(check_result)
1025
def test_upload_part(self):
1027
class StubQuery(client.Query):
1029
def __init__(query, action, creds, endpoint, bucket=None,
1030
object_name=None, data="", body_producer=None,
1031
content_type=None, receiver_factory=None, metadata={}):
1032
super(StubQuery, query).__init__(action=action, creds=creds,
1034
object_name=object_name,
1036
self.assertEquals(action, "PUT")
1037
self.assertEqual(creds.access_key, "foo")
1038
self.assertEqual(creds.secret_key, "bar")
1039
self.assertEqual(query.bucket, "example-bucket")
1040
self.assertEqual(query.object_name,
1041
"example-object?partNumber=3&uploadId=testid")
1042
self.assertEqual(query.data, "some data")
1043
self.assertEqual(query.metadata, {})
1045
def submit(query, url_context=None):
1046
return succeed(None)
1048
creds = AWSCredentials("foo", "bar")
1049
s3 = client.S3Client(creds, query_factory=StubQuery)
1050
return s3.upload_part("example-bucket", "example-object", "testid", 3,
1025
1053
S3ClientTestCase.skip = s3clientSkip