| 330c51eb |
#!/usr/bin/env python |
| f11319be |
# -*- coding=utf-8 -*- |
| 330c51eb |
## Amazon S3cmd - testsuite
## Author: Michal Ludvig <michal@logix.cz>
## http://www.logix.cz/michal
## License: GPL Version 2
import sys |
| 9856527a |
import os |
| 330c51eb |
import re
from subprocess import Popen, PIPE, STDOUT
count_pass = 0
count_fail = 0
def test(label, cmd_args = [], retcode = 0, must_find = [], must_not_find = [], must_find_re = [], must_not_find_re = []):
def failure(message = ""):
global count_fail
if message:
message = " (%s)" % message |
| f11319be |
print "\x1b[31;1mFAIL%s\x1b[0m" % (message) |
| 330c51eb |
count_fail += 1
print "----"
print " ".join([arg.find(" ")>=0 and "'%s'" % arg or arg for arg in cmd_args])
print "----"
print stdout
print "----"
return 1
def success(message = ""):
global count_pass
if message:
message = " (%s)" % message |
| f11319be |
print "\x1b[32;1mOK\x1b[0m%s" % (message) |
| 330c51eb |
count_pass += 1
return 0
def compile_list(_list, regexps = False):
if type(_list) not in [ list, tuple ]:
_list = [_list]
if regexps == False: |
| f11319be |
_list = [re.escape(item.encode("utf-8")) for item in _list] |
| 330c51eb |
return [re.compile(item) for item in _list]
print (label + " ").ljust(30, "."),
sys.stdout.flush()
p = Popen(cmd_args, stdout = PIPE, stderr = STDOUT, universal_newlines = True)
stdout, stderr = p.communicate()
if retcode != p.returncode:
return failure("retcode: %d, expected: %d" % (p.returncode, retcode))
find_list = []
find_list.extend(compile_list(must_find))
find_list.extend(compile_list(must_find_re, regexps = True)) |
| f11319be |
find_list_patterns = []
find_list_patterns.extend(must_find)
find_list_patterns.extend(must_find_re)
|
| 330c51eb |
not_find_list = []
not_find_list.extend(compile_list(must_not_find))
not_find_list.extend(compile_list(must_not_find_re, regexps = True)) |
| f11319be |
not_find_list_patterns = []
not_find_list_patterns.extend(must_not_find)
not_find_list_patterns.extend(must_not_find_re) |
| 330c51eb |
|
| f11319be |
for index in range(len(find_list)):
match = find_list[index].search(stdout) |
| 330c51eb |
if not match: |
| f11319be |
return failure("pattern not found: %s" % find_list_patterns[index])
for index in range(len(not_find_list)):
match = not_find_list[index].search(stdout) |
| 330c51eb |
if match: |
| f11319be |
return failure("pattern found: %s (match: %s)" % (not_find_list_patterns[index], match.group(0))) |
| 330c51eb |
return success()
def test_s3cmd(label, cmd_args = [], **kwargs):
if not cmd_args[0].endswith("s3cmd"): |
| 9856527a |
cmd_args.insert(0, "python")
cmd_args.insert(1, "s3cmd")
|
| 330c51eb |
return test(label, cmd_args, **kwargs)
|
| f11319be |
test_s3cmd("Remove test buckets", ['rb', '-r', 's3://s3cmd-autotest-1', 's3://s3cmd-autotest-2', 's3://s3cmd-Autotest-3'], |
| 330c51eb |
must_find = [ "Bucket 's3://s3cmd-autotest-1/' removed",
"Bucket 's3://s3cmd-autotest-2/' removed", |
| f11319be |
"Bucket 's3://s3cmd-Autotest-3/' removed" ])
test_s3cmd("Create one bucket (EU)", ['mb', '--bucket-location=EU', 's3://s3cmd-autotest-1'], |
| 330c51eb |
must_find = "Bucket 's3://s3cmd-autotest-1/' created")
|
| f11319be |
test_s3cmd("Create multiple buckets", ['mb', 's3://s3cmd-autotest-2', 's3://s3cmd-Autotest-3'],
must_find = [ "Bucket 's3://s3cmd-autotest-2/' created", "Bucket 's3://s3cmd-Autotest-3/' created" ]) |
| 330c51eb |
|
| f11319be |
test_s3cmd("Invalid bucket name", ["mb", "--bucket-location=EU", "s3://s3cmd-Autotest-EU"], |
| 330c51eb |
retcode = 1, |
| f11319be |
must_find = "ERROR: Parameter problem: Bucket name 's3cmd-Autotest-EU' contains disallowed character", |
| 330c51eb |
must_not_find_re = "Bucket.*created")
test_s3cmd("Buckets list", ["ls"], |
| f11319be |
must_find = [ "autotest-1", "autotest-2", "Autotest-3" ], must_not_find_re = "Autotest-EU")
|
| 9856527a |
if os.name != "nt":
## Full testsuite - POSIX (Unix, Linux, ...)
test_s3cmd("Sync to S3", ['sync', 'testsuite', 's3://s3cmd-autotest-1/xyz/', '--exclude', '.svn/*', '--exclude', '*.png', '--no-encrypt'])
test_s3cmd("List bucket content", ['ls', 's3://s3cmd-autotest-1/xyz/'],
must_find_re = [ u"D s3://s3cmd-autotest-1/xyz/unicode/$" ],
must_not_find = [ u"ŪņЇЌœđЗ/☺ unicode € rocks ™" ])
test_s3cmd("List bucket recursive", ['ls', '--recursive', 's3://s3cmd-autotest-1'],
must_find = [ u"s3://s3cmd-autotest-1/xyz/binary/random-crap.md5",
u"s3://s3cmd-autotest-1/xyz/unicode/ŪņЇЌœđЗ/☺ unicode € rocks ™" ],
must_not_find = [ "logo.png" ])
# test_s3cmd("Recursive put", ['put', '--recursive', 'testsuite/etc', 's3://s3cmd-autotest-1/xyz/'])
test_s3cmd("Put public, guess MIME", ['put', '--guess-mime-type', '--acl-public', 'testsuite/etc/logo.png', 's3://s3cmd-autotest-1/xyz/etc/logo.png'],
must_find = [ "stored as s3://s3cmd-autotest-1/xyz/etc/logo.png" ])
test("Removing local target", ['rm', '-rf', 'testsuite-out'])
test_s3cmd("Sync from S3", ['sync', 's3://s3cmd-autotest-1/xyz', 'testsuite-out'],
must_find = [ "stored as testsuite-out/etc/logo.png ", u"unicode/ŪņЇЌœđЗ/☺ unicode € rocks ™" ]) |
| f11319be |
|
| 9856527a |
test("Retrieve public URL", ['wget', 'http://s3cmd-autotest-1.s3.amazonaws.com/xyz/etc/logo.png'],
must_find_re = [ 'logo.png.*saved \[22059/22059\]' ]) |
| f11319be |
|
| 9856527a |
test_s3cmd("Sync more to S3", ['sync', 'testsuite', 's3://s3cmd-autotest-1/xyz/', '--exclude', '*.png', '--no-encrypt']) |
| 416741b2 |
|
| 9856527a |
else:
## Reduced testsuite - Windows NT+
test_s3cmd("Sync to S3", ['sync', 'testsuite', 's3://s3cmd-autotest-1/xyz/', '--exclude', '.svn/*', '--exclude', '*.png', '--exclude', 'unicode/*', '--no-encrypt']) |
| f11319be |
|
| 9856527a |
test_s3cmd("Check bucket content (-r)", ['ls', '--recursive', 's3://s3cmd-autotest-1'],
must_find = [ u"s3://s3cmd-autotest-1/xyz/binary/random-crap.md5" ],
must_not_find = [ "logo.png" ]) |
| f11319be |
|
| 9856527a |
test_s3cmd("Check bucket content", ['ls', 's3://s3cmd-autotest-1/xyz/'],
must_find_re = [ u"D s3://s3cmd-autotest-1/xyz/binary/$" ],
must_not_find = [ u"random-crap.md5" ]) |
| f11319be |
|
| 9856527a |
test_s3cmd("Put public, guess MIME", ['put', '--guess-mime-type', '--acl-public', 'testsuite/etc/logo.png', 's3://s3cmd-autotest-1/xyz/etc/logo.png'],
must_find = [ "stored as s3://s3cmd-autotest-1/xyz/etc/logo.png" ]) |
| f11319be |
|
| 9856527a |
if os.path.isdir("testsuite-out"):
test("Removing local target", ['rmdir', '/s/q', 'testsuite-out']) |
| f11319be |
|
| 9856527a |
test_s3cmd("Sync from S3", ['sync', 's3://s3cmd-autotest-1/xyz', 'testsuite-out'],
must_find = [ "stored as testsuite-out/etc/logo.png " ]) |
| f11319be |
|
| 9856527a |
## Common for POSIX and Win32 |
| f11319be |
test_s3cmd("Rename within S3", ['mv', 's3://s3cmd-autotest-1/xyz/etc/logo.png', 's3://s3cmd-autotest-1/xyz/etc2/Logo.PNG'],
must_find = [ 'Object s3://s3cmd-autotest-1/xyz/etc/logo.png moved to s3://s3cmd-autotest-1/xyz/etc2/Logo.PNG' ])
test_s3cmd("Rename (NoSuchKey)", ['mv', 's3://s3cmd-autotest-1/xyz/etc/logo.png', 's3://s3cmd-autotest-1/xyz/etc2/Logo.PNG'],
retcode = 1,
must_find_re = [ 'ERROR:.*NoSuchKey' ],
must_not_find = [ 'Object s3://s3cmd-autotest-1/xyz/etc/logo.png moved to s3://s3cmd-autotest-1/xyz/etc2/Logo.PNG' ])
test_s3cmd("Sync more from S3", ['sync', '--delete-removed', 's3://s3cmd-autotest-1/xyz', 'testsuite-out'],
must_find = [ "deleted 'testsuite-out/etc/logo.png'", "stored as testsuite-out/etc2/Logo.PNG (22059 bytes",
"stored as testsuite-out/.svn/format " ],
must_not_find = [ "not-deleted etc/logo.png" ])
test_s3cmd("Copy between buckets", ['cp', 's3://s3cmd-autotest-1/xyz/etc2/Logo.PNG', 's3://s3cmd-Autotest-3'],
must_find = [ "Object s3://s3cmd-autotest-1/xyz/etc2/Logo.PNG copied to s3://s3cmd-Autotest-3/xyz/etc2/Logo.PNG" ])
test_s3cmd("Simple delete", ['del', 's3://s3cmd-autotest-1/xyz/etc2/Logo.PNG'],
must_find = [ "Object s3://s3cmd-autotest-1/xyz/etc2/Logo.PNG deleted" ])
|
| 9856527a |
if os.name != "nt":
test_s3cmd("Recursive delete", ['del', '--recursive', 's3://s3cmd-autotest-1/xyz/unicode'],
must_find_re = [ "Object.*unicode/ŪņЇЌœđЗ/.*deleted" ]) |
| f11319be |
test_s3cmd("Recursive delete all", ['del', '--recursive', '--force', 's3://s3cmd-autotest-1'],
must_find_re = [ "Object.*binary/random-crap deleted" ])
test_s3cmd("Remove empty bucket", ['rb', 's3://s3cmd-autotest-1'],
must_find = [ "Bucket 's3://s3cmd-autotest-1/' removed" ]) |
| 330c51eb |
|
| f11319be |
test_s3cmd("Remove remaining buckets", ['rb', '--recursive', 's3://s3cmd-autotest-2', 's3://s3cmd-Autotest-3'],
must_find = [ "Bucket 's3://s3cmd-autotest-2/' removed",
"Bucket 's3://s3cmd-Autotest-3/' removed" ]) |