Pull in upstream boto from github at bcb719937de9ac2851e632d62b777352029a6d55

tests/s3/test_gsconnection.py

 # -*- coding: utf-8 -*-
 # Copyright (c) 2006-2011 Mitch Garnaat http://garnaat.org/
 # Copyright (c) 2010, Eucalyptus Systems, Inc.
 # Copyright (c) 2011, Nexenta Systems, Inc.
 # All rights reserved.
 #
 # Permission is hereby granted, free of charge, to any person obtaining a
 # copy of this software and associated documentation files (the
 # "Software"), to deal in the Software without restriction, including
 # without limitation the rights to use, copy, modify, merge, publish, dis-
@@ skipping 13 matching lines @@
 # IN THE SOFTWARE.
 
 """
 Some unit tests for the GSConnection
 """
 
 import unittest
 import time
 import os
 from boto.gs.connection import GSConnection
+from boto import storage_uri
 
 class GSConnectionTest (unittest.TestCase):
 
     def test_1_basic(self):
+        """basic regression test for Google Cloud Storage"""
         print '--- running GSConnection tests ---'
         c = GSConnection()
         # create a new, empty bucket
         bucket_name = 'test-%d' % int(time.time())
         bucket = c.create_bucket(bucket_name)
         # now try a get_bucket call and see if it's really there
         bucket = c.get_bucket(bucket_name)
         k = bucket.new_key()
         k.name = 'foobar'
         s1 = 'This is a test of file upload and download'
@@ skipping 121 matching lines @@
         assert bucket.get_subresource('logging') == empty_logging_str
         bucket.enable_logging('log-bucket', 'example',
                              canned_acl='bucket-owner-full-control')
         assert bucket.get_subresource('logging') == logging_str;
         # now delete all keys in bucket
         for k in bucket:
             bucket.delete_key(k)
         # now delete bucket
         time.sleep(5)
         c.delete_bucket(bucket)
+
+    def test_2_copy_key(self):
+        """test copying a key from one bucket to another"""
+        c = GSConnection()
+        # create two new, empty buckets
+        bucket_name_1 = 'test1-%d' % int(time.time())
+        bucket_name_2 = 'test2-%d' % int(time.time())
+        bucket1 = c.create_bucket(bucket_name_1)
+        bucket2 = c.create_bucket(bucket_name_2)
+        # verify buckets got created 
+        bucket1 = c.get_bucket(bucket_name_1)
+        bucket2 = c.get_bucket(bucket_name_2)
+        # create a key in bucket1 and give it some content
+        k1 = bucket1.new_key()
+        assert isinstance(k1, bucket1.key_class)
+        key_name = 'foobar'
+        k1.name = key_name
+        s = 'This is a test.'
+        k1.set_contents_from_string(s)
+        # copy the new key from bucket1 to bucket2
+        k1.copy(bucket_name_2, key_name) 
+        # now copy the contents from bucket2 to a local file
+        k2 = bucket2.lookup(key_name)
+        assert isinstance(k2, bucket2.key_class)
+        fp = open('foobar', 'wb')
+        k2.get_contents_to_file(fp)
+        fp.close()
+        fp = open('foobar')
+        # check to make sure content read is identical to original
+        assert s == fp.read(), 'move test failed!'
+        fp.close()
+        # delete keys
+        bucket1.delete_key(k1)
+        bucket2.delete_key(k2)
+        # delete test buckets
+        c.delete_bucket(bucket1)
+        c.delete_bucket(bucket2)
+
+    def test_3_default_object_acls(self):
+        """test default object acls"""
+        c = GSConnection()
+        # create a new bucket
+        bucket_name = 'test-%d' % int(time.time())
+        bucket = c.create_bucket(bucket_name)
+        # now call get_bucket to see if it's really there
+        bucket = c.get_bucket(bucket_name)
+        # get default acl and make sure it's empty
+        acl = bucket.get_def_acl()
+        assert acl.to_xml() == '<AccessControlList></AccessControlList>'
+        # set default acl to a canned acl and verify it gets set
+        bucket.set_def_acl('public-read')
+        acl = bucket.get_def_acl()
+        # save public-read acl for later test
+        public_read_acl = acl
+        assert acl.to_xml() == ('<AccessControlList><Entries><Entry>'    +
+          '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' +
+          '</Entry></Entries></AccessControlList>')
+        # back to private acl
+        bucket.set_def_acl('private')
+        acl = bucket.get_def_acl()
+        assert acl.to_xml() == '<AccessControlList></AccessControlList>'
+        # set default acl to an xml acl and verify it gets set
+        bucket.set_def_acl(public_read_acl)
+        acl = bucket.get_def_acl()
+        assert acl.to_xml() == ('<AccessControlList><Entries><Entry>'    +
+          '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' +
+          '</Entry></Entries></AccessControlList>')
+        # back to private acl
+        bucket.set_def_acl('private')
+        acl = bucket.get_def_acl()
+        assert acl.to_xml() == '<AccessControlList></AccessControlList>'
+        # delete bucket
+        c.delete_bucket(bucket)
+        # repeat default acl tests using boto's storage_uri interface
+        # create a new bucket
+        bucket_name = 'test-%d' % int(time.time())
+        uri = storage_uri('gs://' + bucket_name)
+        uri.create_bucket()
+        # get default acl and make sure it's empty
+        acl = uri.get_def_acl()
+        assert acl.to_xml() == '<AccessControlList></AccessControlList>'
+        # set default acl to a canned acl and verify it gets set
+        uri.set_def_acl('public-read')
+        acl = uri.get_def_acl()
+        # save public-read acl for later test
+        public_read_acl = acl
+        assert acl.to_xml() == ('<AccessControlList><Entries><Entry>'    +
+          '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' +
+          '</Entry></Entries></AccessControlList>')
+        # back to private acl
+        uri.set_def_acl('private')
+        acl = uri.get_def_acl()
+        assert acl.to_xml() == '<AccessControlList></AccessControlList>'
+        # set default acl to an xml acl and verify it gets set
+        uri.set_def_acl(public_read_acl)
+        acl = uri.get_def_acl()
+        assert acl.to_xml() == ('<AccessControlList><Entries><Entry>'    +
+          '<Scope type="AllUsers"></Scope><Permission>READ</Permission>' +
+          '</Entry></Entries></AccessControlList>')
+        # back to private acl
+        uri.set_def_acl('private')
+        acl = uri.get_def_acl()
+        assert acl.to_xml() == '<AccessControlList></AccessControlList>'
+        # delete bucket
+        uri.delete_bucket()
+        
         print '--- tests completed ---'