Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: 7d36ee19fc
Fetching contributors…

Octocat-spinner-32-eaf2f5

Cannot retrieve contributors at this time

file 269 lines (195 sloc) 12.455 kb
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
#!/usr/bin/env python

# This software code is made available "AS IS" without warranties of any
# kind. You may copy, display, modify and redistribute the software
# code either by itself or as incorporated into your code; provided that
# you do not remove any proprietary notices. Your use of this software
# code is at your own risk and you waive any claim against Amazon
# Digital Services, Inc. or its affiliates with respect to your use of
# this software code. (c) 2006-2007 Amazon Digital Services, Inc. or its
# affiliates.

import unittest
import S3
import httplib
import sys

AWS_ACCESS_KEY_ID = '<INSERT YOUR AWS ACCESS KEY ID HERE>'
AWS_SECRET_ACCESS_KEY = '<INSERT YOUR AWS SECRET ACCESS KEY HERE>'
# remove these next two lines when you've updated your credentials.
print "update s3-test.py with your AWS credentials"
sys.exit();

# for subdomains (bucket.s3.amazonaws.com),
# the bucket name must be lowercase since DNS is case-insensitive
BUCKET_NAME = "%s-test-bucket" % AWS_ACCESS_KEY_ID.lower();


class TestAWSAuthConnection(unittest.TestCase):
    def setUp(self):
        self.conn = S3.AWSAuthConnection(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)

    # test all operations for both regular and vanity domains
    # regular: http://s3.amazonaws.com/bucket/key
    # subdomain: http://bucket.s3.amazonaws.com/key
    # testing pure vanity domains (http://<vanity domain>/key) is not covered here
    # but is possible with some additional setup (set the server in @conn to your vanity domain)

    def test_subdomain_default(self):
        self.run_tests(S3.CallingFormat.SUBDOMAIN, S3.Location.DEFAULT)

    def test_subdomain_eu(self):
        self.run_tests(S3.CallingFormat.SUBDOMAIN, S3.Location.EU)

    def test_path_default(self):
        self.run_tests(S3.CallingFormat.PATH, S3.Location.DEFAULT)


    def run_tests(self, calling_format, location):
        self.conn.calling_format = calling_format

        response = self.conn.create_located_bucket(BUCKET_NAME, location)
        self.assertEquals(response.http_response.status, 200, 'create bucket')

        response = self.conn.list_bucket(BUCKET_NAME)
        self.assertEquals(response.http_response.status, 200, 'list bucket')
        self.assertEquals(len(response.entries), 0, 'bucket is empty')

        text = 'this is a test'
        key = 'example.txt'

        response = self.conn.put(BUCKET_NAME, key, text)
        self.assertEquals(response.http_response.status, 200, 'put with a string argument')

        response = \
            self.conn.put(
                    BUCKET_NAME,
                    key,
                    S3.S3Object(text, {'title': 'title'}),
                    {'Content-Type': 'text/plain'})

        self.assertEquals(response.http_response.status, 200, 'put with complex argument and headers')

        response = self.conn.get(BUCKET_NAME, key)
        self.assertEquals(response.http_response.status, 200, 'get object')
        self.assertEquals(response.object.data, text, 'got right data')
        self.assertEquals(response.object.metadata, { 'title': 'title' }, 'metadata is correct')
        self.assertEquals(int(response.http_response.getheader('Content-Length')), len(text), 'got content-length header')

        title_with_spaces = " \t title with leading and trailing spaces "
        response = \
            self.conn.put(
                    BUCKET_NAME,
                    key,
                    S3.S3Object(text, {'title': title_with_spaces}),
                    {'Content-Type': 'text/plain'})

        self.assertEquals(response.http_response.status, 200, 'put with headers with spaces')

        response = self.conn.get(BUCKET_NAME, key)
        self.assertEquals(response.http_response.status, 200, 'get object')
        self.assertEquals(
                response.object.metadata,
                { 'title': title_with_spaces.strip() },
                'metadata with spaces is correct')

        # delimited list tests
        inner_key = 'test/inner.txt'
        last_key = 'z-last-key.txt'
        response = self.conn.put(BUCKET_NAME, inner_key, text)
        self.assertEquals(response.http_response.status, 200, 'put inner key')

        response = self.conn.put(BUCKET_NAME, last_key, text)
        self.assertEquals(response.http_response.status, 200, 'put last key')

        response = self.do_delimited_list(BUCKET_NAME, False, {'delimiter': '/'}, 2, 1, 'root list')

        response = self.do_delimited_list(BUCKET_NAME, True, {'max-keys': 1, 'delimiter': '/'}, 1, 0, 'root list with max keys of 1', 'example.txt')

        response = self.do_delimited_list(BUCKET_NAME, True, {'max-keys': 2, 'delimiter': '/'}, 1, 1, 'root list with max keys of 2, page 1', 'test/')

        marker = response.next_marker

        response = self.do_delimited_list(BUCKET_NAME, False, {'marker': marker, 'max-keys': 2, 'delimiter': '/'}, 1, 0, 'root list with max keys of 2, page 2')

        response = self.do_delimited_list(BUCKET_NAME, False, {'prefix': 'test/', 'delimiter': '/'}, 1, 0, 'test/ list')

        response = self.conn.delete(BUCKET_NAME, inner_key)
        self.assertEquals(response.http_response.status, 204, 'delete %s' % inner_key)

        response = self.conn.delete(BUCKET_NAME, last_key)
        self.assertEquals(response.http_response.status, 204, 'delete %s' % last_key)


        weird_key = '&=//%# ++++'

        response = self.conn.put(BUCKET_NAME, weird_key, text)
        self.assertEquals(response.http_response.status, 200, 'put weird key')

        response = self.conn.get(BUCKET_NAME, weird_key)
        self.assertEquals(response.http_response.status, 200, 'get weird key')

        response = self.conn.get_acl(BUCKET_NAME, key)
        self.assertEquals(response.http_response.status, 200, 'get acl')

        acl = response.object.data

        response = self.conn.put_acl(BUCKET_NAME, key, acl)
        self.assertEquals(response.http_response.status, 200, 'put acl')

        response = self.conn.get_bucket_acl(BUCKET_NAME)
        self.assertEquals(response.http_response.status, 200, 'get bucket acl')

        bucket_acl = response.object.data

        response = self.conn.put_bucket_acl(BUCKET_NAME, bucket_acl)
        self.assertEquals(response.http_response.status, 200, 'put bucket acl')

        response = self.conn.get_bucket_acl(BUCKET_NAME)
        self.assertEquals(response.http_response.status, 200, 'get bucket logging')

        bucket_logging = response.object.data

        response = self.conn.put_bucket_acl(BUCKET_NAME, bucket_logging)
        self.assertEquals(response.http_response.status, 200, 'put bucket logging')

        response = self.conn.list_bucket(BUCKET_NAME)
        self.assertEquals(response.http_response.status, 200, 'list bucket')
        entries = response.entries
        self.assertEquals(len(entries), 2, 'got back right number of keys')
        # depends on weird_key < key
        self.assertEquals(entries[0].key, weird_key, 'first key is right')
        self.assertEquals(entries[1].key, key, 'second key is right')

        response = self.conn.list_bucket(BUCKET_NAME, {'max-keys': 1})
        self.assertEquals(response.http_response.status, 200, 'list bucket with args')
        self.assertEquals(len(response.entries), 1, 'got back right number of keys')

        for entry in entries:
            response = self.conn.delete(BUCKET_NAME, entry.key)
            self.assertEquals(response.http_response.status, 204, 'delete %s' % entry.key)

        response = self.conn.list_all_my_buckets()
        self.assertEquals(response.http_response.status, 200, 'list all my buckets')
        buckets = response.entries

        response = self.conn.delete_bucket(BUCKET_NAME)
        self.assertEquals(response.http_response.status, 204, 'delete bucket')

        response = self.conn.list_all_my_buckets()
        self.assertEquals(response.http_response.status, 200, 'list all my buckets again')

        self.assertEquals(len(response.entries), len(buckets) - 1, 'bucket count is correct')

    def verify_list_bucket_response(self, response, bucket, is_truncated, parameters, next_marker=''):
        prefix = ''
        marker = ''

        if parameters.has_key('prefix'):
            prefix = parameters['prefix']
        if parameters.has_key('marker'):
            marker = parameters['marker']

        self.assertEquals(bucket, response.name, 'bucket name should match')
        self.assertEquals(prefix, response.prefix, 'prefix should match')
        self.assertEquals(marker, response.marker, 'marker should match')
        if parameters.has_key('max-keys'):
            self.assertEquals(parameters['max-keys'], response.max_keys, 'max-keys should match')
        self.assertEquals(parameters['delimiter'], response.delimiter, 'delimiter should match')
        self.assertEquals(is_truncated, response.is_truncated, 'is_truncated should match')
        self.assertEquals(next_marker, response.next_marker, 'next_marker should match')

    def do_delimited_list(self, bucket_name, is_truncated, parameters, regular_expected, common_expected, test_name, next_marker=''):
        response = self.conn.list_bucket(bucket_name, parameters)
        self.assertEquals(response.http_response.status, 200, test_name)
        self.assertEquals(regular_expected, len(response.entries), 'right number of regular entries')
        self.assertEquals(common_expected, len(response.common_prefixes), 'right number of common prefixes')

        self.verify_list_bucket_response(response, bucket_name, is_truncated, parameters, next_marker)

        return response

class TestQueryStringAuthGenerator(unittest.TestCase):
    def setUp(self):
        self.generator = S3.QueryStringAuthGenerator(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
        if (self.generator.is_secure == True):
            self.connection = httplib.HTTPSConnection(self.generator.server_name)
        else:
            self.connection = httplib.HTTPConnection(self.generator.server_name)

    def check_url(self, url, method, status, message, data=''):
        if (method == 'PUT'):
            headers = { 'Content-Length': len(data) }
            self.connection.request(method, url, data, headers)
        else:
            self.connection.request(method, url)

        response = self.connection.getresponse()
        self.assertEquals(response.status, status, message)

        return response.read()

    # test all operations for both regular and vanity domains
    # regular: http://s3.amazonaws.com/bucket/key
    # subdomain: http://bucket.s3.amazonaws.com/key
    # testing pure vanity domains (http://<vanity domain>/key) is not covered here
    # but is possible with some additional setup (set the server in @conn to your vanity domain)

    def test_subdomain(self):
        self.run_tests(S3.CallingFormat.SUBDOMAIN)

    def test_path(self):
        self.run_tests(S3.CallingFormat.PATH)

    def run_tests(self, calling_format):
        self.generator.calling_format = calling_format

        key = 'test'

        self.check_url(self.generator.create_bucket(BUCKET_NAME), 'PUT', 200, 'create_bucket')
        self.check_url(self.generator.put(BUCKET_NAME, key, ''), 'PUT', 200, 'put object', 'test data')
        self.check_url(self.generator.get(BUCKET_NAME, key), 'GET', 200, 'get object')
        self.check_url(self.generator.list_bucket(BUCKET_NAME), 'GET', 200, 'list bucket')
        self.check_url(self.generator.list_all_my_buckets(), 'GET', 200, 'list all my buckets')
        acl = self.check_url(self.generator.get_acl(BUCKET_NAME, key), 'GET', 200, 'get acl')
        self.check_url(self.generator.put_acl(BUCKET_NAME, key, acl), 'PUT', 200, 'put acl', acl)
        bucket_acl = self.check_url(self.generator.get_bucket_acl(BUCKET_NAME), 'GET', 200, 'get bucket acl')
        self.check_url(self.generator.put_bucket_acl(BUCKET_NAME, bucket_acl), 'PUT', 200, 'put bucket acl', bucket_acl)
        bucket_logging = self.check_url(self.generator.get_bucket_logging(BUCKET_NAME), 'GET', 200, 'get bucket logging')
        self.check_url(self.generator.put_bucket_logging(BUCKET_NAME, bucket_logging), 'PUT', 200, 'put bucket logging', bucket_logging)
        self.check_url(self.generator.delete(BUCKET_NAME, key), 'DELETE', 204, 'delete object')
        self.check_url(self.generator.delete_bucket(BUCKET_NAME), 'DELETE', 204, 'delete bucket')


if __name__ == '__main__':
    unittest.main()



Something went wrong with that request. Please try again.