test_abs_url.py

#!/usr/bin/env python

import boto
import boto.s3.connection
from boto.compat import six, urllib
from boto.s3.connection import SubdomainCallingFormat
from boto.connection import AWSAuthConnection

class AbsoluteCallingFormat(SubdomainCallingFormat):
    # Need this as we need to create a path, but we only pass on
    # the bucket and key, and we don't want to override make_request also
    def __init__(self,host,port,ssl):
        self.host = host
        self.port = port
        self.ssl = ssl


    def make_endpoint(self,host,port,ssl):
        if ssl:
            return "https://" + host + ":" + str(port)
        else:
            return "http://" + host + ":" + str(port)

    def build_path_base(self,bucket,key=''):
        key = boto.utils.get_utf8_value(key)
        hoststr = self.build_host(self.host,bucket)
        endpoint = self.make_endpoint(hoststr,self.port,self.ssl)
        if endpoint[-1] != '/':
            endpoint += '/'
        boto.log.debug("endpoint is %s" % endpoint)
        return endpoint + urllib.parse.quote(key)

# Walking on thin ice here, currently get_path explicitly adds a slash in front
# of all requests, which we have to avoid
def abs_get_path(self, path='/'):
    return path

AWSAuthConnection.get_path = abs_get_path


if __name__ == "__main__":

    access = '0555b35654ad1656d804'
    secret = 'h7GhxuBLTrlhVUyxSPUKUV8r/2EI4ngqJxD7iBdBYLhwluN30JaT3Q=='
    host = 'localhost'
    port = 8000
    ssl = False
    #boto.set_stream_logger('abs')

    abs_calling_format = AbsoluteCallingFormat(host,port,ssl)
    conn = boto.connect_s3(access,secret,is_secure=ssl,
                           host=host,port=port,
                           calling_format=abs_calling_format)


    testb = conn.create_bucket('testbucket')
    print "created bucket %s" % testb

    key = testb.new_key('foo.txt')
    key.set_contents_from_string('hello from boto')


    keys = testb.get_all_keys()
    print "got_keys"
    for key in keys:
        print key.name
    print "listing all the buckets"
    xs = conn.get_all_buckets()
    for x in xs:
        print "{name} : {created}".format(name=x.name, created=x.creation_date)