← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python-boto/trusty

« back to all changes in this revision

Viewing changes to tests/integration/s3/test_multipart.py

  • Committer: Package Import Robot
  • Author(s): Eric Evans
  • Date: 2013-05-10 23:38:14 UTC
  • mfrom: (1.1.10) (14.1.2 experimental)
  • Revision ID: package-import@ubuntu.com-20130510233814-701dvlop7xfh88i7
Tags: 2.9.2-1
http://bugs.debian.org/700743
New upstream release (Closes: #700743).

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/integration/s3/test_multipart.py
 
1
# -*- coding: utf-8 -*-
 
2
 
 
3
# Copyright (c) 2011 Mitch Garnaat http://garnaat.org/
 
4
# All rights reserved.
 
5
#
 
6
# Permission is hereby granted, free of charge, to any person obtaining a
 
7
# copy of this software and associated documentation files (the
 
8
# "Software"), to deal in the Software without restriction, including
 
9
# without limitation the rights to use, copy, modify, merge, publish, dis-
 
10
# tribute, sublicense, and/or sell copies of the Software, and to permit
 
11
# persons to whom the Software is furnished to do so, subject to the fol-
 
12
# lowing conditions:
 
13
#
 
14
# The above copyright notice and this permission notice shall be included
 
15
# in all copies or substantial portions of the Software.
 
16
#
 
17
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
 
18
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
 
19
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
 
20
# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 
 
21
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 
22
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 
23
# IN THE SOFTWARE.
 
24
 
 
25
"""
 
26
Some unit tests for the S3 MultiPartUpload
 
27
"""
 
28
 
 
29
# Note:
 
30
# Multipart uploads require at least one part. If you upload
 
31
# multiple parts then all parts except the last part has to be
 
32
# bigger than 5M. Hence we just use 1 part so we can keep
 
33
# things small and still test logic.
 
34
 
 
35
import unittest
 
36
import time
 
37
import StringIO
 
38
from boto.s3.connection import S3Connection
 
39
 
 
40
 
 
41
class S3MultiPartUploadTest(unittest.TestCase):
 
42
    s3 = True
 
43
 
 
44
    def setUp(self):
 
45
        self.conn = S3Connection(is_secure=False)
 
46
        self.bucket_name = 'multipart-%d' % int(time.time())
 
47
        self.bucket = self.conn.create_bucket(self.bucket_name)
 
48
 
 
49
    def tearDown(self):
 
50
        for key in self.bucket:
 
51
            key.delete()
 
52
        self.bucket.delete()
 
53
 
 
54
    def test_abort(self):
 
55
        key_name = u"テスト"
 
56
        mpu = self.bucket.initiate_multipart_upload(key_name)
 
57
        mpu.cancel_upload()
 
58
 
 
59
    def test_complete_ascii(self):
 
60
        key_name = "test"
 
61
        mpu = self.bucket.initiate_multipart_upload(key_name)
 
62
        fp = StringIO.StringIO("small file")
 
63
        mpu.upload_part_from_file(fp, part_num=1)
 
64
        fp.close()
 
65
        cmpu = mpu.complete_upload()
 
66
        self.assertEqual(cmpu.key_name, key_name)
 
67
        self.assertNotEqual(cmpu.etag, None)
 
68
 
 
69
    def test_complete_japanese(self):
 
70
        key_name = u"テスト"
 
71
        mpu = self.bucket.initiate_multipart_upload(key_name)
 
72
        fp = StringIO.StringIO("small file")
 
73
        mpu.upload_part_from_file(fp, part_num=1)
 
74
        fp.close()
 
75
        cmpu = mpu.complete_upload()
 
76
        # LOL... just found an Amazon bug when it returns the
 
77
        # key in the completemultipartupload result. AWS returns
 
78
        # ??? instead of the correctly encoded key name. We should
 
79
        # fix this to the comment line below when amazon fixes this
 
80
        # and this test starts failing due to below assertion.
 
81
        self.assertEqual(cmpu.key_name, "???")
 
82
        #self.assertEqual(cmpu.key_name, key_name)
 
83
        self.assertNotEqual(cmpu.etag, None)
 
84
 
 
85
    def test_list_japanese(self):
 
86
        key_name = u"テスト"
 
87
        mpu = self.bucket.initiate_multipart_upload(key_name)
 
88
        rs = self.bucket.list_multipart_uploads()
 
89
        # New bucket, so only one upload expected
 
90
        lmpu = iter(rs).next()
 
91
        self.assertEqual(lmpu.id, mpu.id)
 
92
        self.assertEqual(lmpu.key_name, key_name)
 
93
        # Abort using the one returned in the list
 
94
        lmpu.cancel_upload()
 
95
 
 
96
    def test_list_multipart_uploads(self):
 
97
        key_name = u"テスト"
 
98
        mpus = []
 
99
        mpus.append(self.bucket.initiate_multipart_upload(key_name))
 
100
        mpus.append(self.bucket.initiate_multipart_upload(key_name))
 
101
        rs = self.bucket.list_multipart_uploads()
 
102
        # uploads (for a key) are returned in time initiated asc order
 
103
        for lmpu in rs:
 
104
            ompu = mpus.pop(0)
 
105
            self.assertEqual(lmpu.key_name, ompu.key_name)
 
106
            self.assertEqual(lmpu.id, ompu.id)
 
107
        self.assertEqual(0, len(mpus))
 
108
 
 
109
    def test_four_part_file(self):
 
110
        key_name = "k"
 
111
        contents = "01234567890123456789"
 
112
        sfp = StringIO.StringIO(contents)
 
113
 
 
114
        # upload 20 bytes in 4 parts of 5 bytes each
 
115
        mpu = self.bucket.initiate_multipart_upload(key_name)
 
116
        mpu.upload_part_from_file(sfp, part_num=1, size=5)
 
117
        mpu.upload_part_from_file(sfp, part_num=2, size=5)
 
118
        mpu.upload_part_from_file(sfp, part_num=3, size=5)
 
119
        mpu.upload_part_from_file(sfp, part_num=4, size=5)
 
120
        sfp.close()
 
121
 
 
122
        etags = {}
 
123
        pn = 0
 
124
        for part in mpu:
 
125
            pn += 1
 
126
            self.assertEqual(5, part.size)
 
127
            etags[pn] = part.etag
 
128
        self.assertEqual(pn, 4)
 
129
        # etags for 01234
 
130
        self.assertEqual(etags[1], etags[3])
 
131
        # etags for 56789
 
132
        self.assertEqual(etags[2], etags[4])
 
133
        # etag 01234 != etag 56789
 
134
        self.assertNotEqual(etags[1], etags[2])
 
135
 
 
136
        # parts are too small to compete as each part must
 
137
        # be a min of 5MB so so we'll assume that is enough
 
138
        # testing and abort the upload.
 
139
        mpu.cancel_upload()