Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

adding more tests for copy

  • Loading branch information...
commit 7f2d90fa8526542d4c7a245e52918229923f88b7 1 parent 8f53da9
BuzzTroll authored
View
1  cumulus/cb/pycb/cbRequest.py
@@ -791,6 +791,7 @@ def end_copy(self):
try:
self.user.put_object(self.dst_file, self.dstBucketName, self.dstObjectName)
+ grant_public_permissions(self.dstBucketName, self.dstObjectName)
doc = Document()
cor = doc.createElement("CopyObjectResult")
View
112 cumulus/tests/client/copy_object_tests.py
@@ -0,0 +1,112 @@
+import string
+import random
+import os
+import sys
+import nose.tools
+import boto
+from boto.s3.connection import OrdinaryCallingFormat
+from boto.s3.connection import VHostCallingFormat
+from boto.s3.connection import SubdomainCallingFormat
+import sys
+from ConfigParser import SafeConfigParser
+from pycb.cumulus import *
+import time
+import pycb.test_common
+import unittest
+import tempfile
+import filecmp
+#
+class TestCopyObjectAuthWithBoto(unittest.TestCase):
+
+ def setUp(self):
+ (self.host, self.port) = pycb.test_common.get_contact()
+ self.user_id_list = []
+
+ def tearDown(self):
+ for (id, pw) in self.user_id_list:
+ self.clean_all(id, pw)
+ pycb.test_common.clean_user(id)
+ pass
+
+ def make_user(self):
+ # note this only works if access to the same FS
+ (id, pw) = pycb.test_common.make_user()
+ self.user_id_list.append((id, pw))
+ return (id, pw)
+
+ def clean_all(self, id, pw):
+ conn = pycb.test_common.cb_get_conn(self.host, self.port, id, pw)
+ nbs = conn.get_all_buckets()
+ for b in nbs:
+ rs = b.list()
+ for key in rs:
+ try:
+ key.delete()
+ except:
+ pass
+ try:
+ b.delete()
+ except:
+ pass
+
+ def cb_random_bucketname(self, len):
+ chars = string.letters + string.digits
+ newpasswd = ""
+ for i in range(len):
+ newpasswd = newpasswd + random.choice(chars)
+ return newpasswd
+
+ def test_get_and_compare_copy(self, type):
+ (id, pw) = self.make_user()
+ conn = pycb.test_common.cb_get_conn(self.host, self.port, id, pw)
+ bucketname = self.cb_random_bucketname(20)
+ bucket = conn.create_bucket(bucketname, policy='public-read')
+ key = self.cb_random_bucketname(20)
+ k = boto.s3.key.Key(bucket)
+ k.key = key
+ k.set_contents_from_filename("/etc/group")
+ bucket.set_acl(type, key)
+
+ (osf, filename) = tempfile.mkstemp()
+ os.close(osf)
+ k.get_contents_to_filename(filename)
+
+ rc = filecmp.cmp("/etc/group", filename)
+ self.assertTrue(rc)
+
+ def cp_object_perm_read(self, type):
+ (id, pw) = self.make_user()
+ conn = pycb.test_common.cb_get_conn(self.host, self.port, id, pw)
+ bucketname = self.cb_random_bucketname(20)
+ bucket = conn.create_bucket(bucketname, policy='public-read')
+ key = self.cb_random_bucketname(20)
+ k = boto.s3.key.Key(bucket)
+ k.key = key
+ k.set_contents_from_filename("/etc/group")
+
+ new_key = self.cb_random_bucketname(20)
+ new_k = boto.s3.key.Key(bucket)
+ new_k.key = new_key
+
+ k.copy(bucket, new_key)
+
+ for t in type:
+ new_k.add_user_grant(t, id)
+
+ (id, pw) = self.make_user()
+ conn = pycb.test_common.cb_get_conn(self.host, self.port, id, pw)
+ bucket2 = conn.get_bucket(bucketname)
+ k = bucket2.get_key(new_key)
+ k.get_contents_to_filename("/dev/null")
+
+
+ def test_cp_object_perm_authed(self):
+ self.cp_object_perm_read(["READ",])
+
+ def test_cp_object_perm_authed(self):
+ self.cp_object_perm_read(["WRITE",])
+
+ def test_cp_object_perm_authed(self):
+ self.cp_object_perm_read(["FULL_CONTROL",])
+
+
View
48 cumulus/tests/s3cmd/test_basic_move.sh
@@ -0,0 +1,48 @@
+#!/bin/bash
+
+bucket_name=CumulusTest$RANDOM
+fname=GRP$RANDOM
+s3cmd mb s3://$bucket_name
+s3cmd put /etc/group s3://$bucket_name/$fname
+if [ "X$?" != "X0" ]; then
+ echo "ERROR"
+ exit 1
+fi
+
+s3cmd mv s3://$bucket_name/$fname s3://$bucket_name/$fname.copy
+if [ "X$?" != "X0" ]; then
+ echo "ERROR"
+ exit 1
+fi
+x=`s3cmd ls s3://$bucket_name/ | grep $fname.copy`
+if [ "X$x" == "X" ]; then
+ echo "error listing"
+ exit 1
+fi
+
+f=`mktemp -t tmp.XXXXXXXXXX`
+s3cmd --force get s3://$bucket_name/$fname.copy $f
+diff -q /etc/group $f
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: diff failed $f"
+ exit 1
+fi
+
+s3cmd del s3://$bucket_name/$fname
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: delete failed"
+ exit 1
+fi
+s3cmd del s3://$bucket_name/$fname.copy
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: delete failed"
+ exit 1
+fi
+
+s3cmd rb s3://$bucket_name
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: delete bucket failed"
+ exit 1
+fi
+
+exit 0
View
30 cumulus/tests/s3cmd/test_copy_error.sh
@@ -0,0 +1,30 @@
+#!/bin/bash
+
+bucket_name=CumulusTest$RANDOM
+fname=GRP$RANDOM
+s3cmd mb s3://$bucket_name
+s3cmd put /etc/group s3://$bucket_name/$fname
+if [ "X$?" != "X0" ]; then
+ echo "ERROR"
+ exit 1
+fi
+
+s3cmd cp s3://$bucket_name/$fname s3://$bucket_name/$fname
+if [ $? -eq 0 ]; then
+ echo "the copy should have failed"
+ exit 1
+fi
+
+s3cmd del s3://$bucket_name/$fname
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: delete failed"
+ exit 1
+fi
+
+s3cmd rb s3://$bucket_name
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: delete bucket failed"
+ exit 1
+fi
+
+exit 0
View
24 cumulus/tests/s3cmd/test_cp_auth.sh
@@ -0,0 +1,24 @@
+#!/bin/bash
+
+bucket_name=CumulusTest$RANDOM
+fname=GRP$RANDOM
+
+s3cmd mb s3://$bucket_name
+# just run it a few times for races
+s3cmd -r put /etc/group s3://$bucket_name/$fname
+if [ "X$?" != "X0" ]; then
+ echo "recursive put failed"
+ exit 1
+fi
+s3cmd --acl-public cp s3://$bucket_name/$fname s3://$bucket_name/$fname.copy
+if [ "X$?" != "X0" ]; then
+ echo "setacl failed failed"
+ exit 1
+fi
+
+s3cmd -r --force del s3://$bucket_name/
+if [ "X$?" != "X0" ]; then
+ echo "ERROR: delete bucket failed"
+ exit 1
+fi
+exit 0
Please sign in to comment.
Something went wrong with that request. Please try again.