From 5be3fc07eb54bedf441f358ccb2ac62152de2a78 Mon Sep 17 00:00:00 2001 From: Thiago da Silva Date: Tue, 21 Jan 2014 14:21:40 -0500 Subject: first libgfapi-python functional tests adding a few functional tests and removing old tests from source code Change-Id: Iefcb091d614f2825592943cfb42847b5865322c6 Signed-off-by: Thiago da Silva --- gluster/gfapi.py | 221 ------------------------------------------------------- 1 file changed, 221 deletions(-) (limited to 'gluster') diff --git a/gluster/gfapi.py b/gluster/gfapi.py index 516a1f0..c2c267c 100644 --- a/gluster/gfapi.py +++ b/gluster/gfapi.py @@ -1,10 +1,6 @@ -#!/usr/bin/python - import ctypes from ctypes.util import find_library import os -import sys -import types from contextlib import contextmanager @@ -277,220 +273,3 @@ class Volume(object): err = ctypes.get_errno() raise OSError(err, os.strerror(err)) return ret - -if __name__ == "__main__": - - def test_create_write(vol, path, data): - mypath = path + ".io" - with vol.creat(mypath, os.O_WRONLY | os.O_EXCL, 0644) as fd: - if not fd: - return False, "creat error" - rc = fd.write(data) - if rc != len(data): - return False, "wrote %d/%d bytes" % (rc, len(data)) - return True, "wrote %d bytes" % rc - - # TBD: this test fails if we do create, open, write, read - def test_open_read(vol, path, data): - mypath = path + ".io" - with vol.open(mypath, os.O_RDONLY) as fd: - if not fd: - return False, "open error" - dlen = len(data) * 2 - buf = fd.read(dlen) - if isinstance(buf, types.IntType): - return False, "read error %d" % buf - if len(buf) != len(data): - return False, "read %d/%d bytes" % (len(buf), len(data)) - return True, "read '%s'" % buf - - def test_lstat(vol, path, data): - mypath = path + ".io" - sb = vol.lstat(mypath) - if isinstance(sb, types.IntType): - return False, "lstat error %d" % sb - if sb.st_size != len(data): - return False, "lstat size is %d, expected %d" % ( - sb.st_size, len(data)) - return True, "lstat got correct size %d" % sb.st_size - - def test_rename(vol, path, data): - opath = path + ".io" - npath = path + ".tmp" - rc = vol.rename(opath, npath) - if rc < 0: - return False, "rename error %d" % rc - try: - with vol.open(opath, os.O_RDWR) as fd: - return False, "old path working (%s) after rename" % fd - except OSError: - pass - else: - return False, "old path working after rename" - - with vol.open(npath, os.O_RDWR) as nfd: - if not isinstance(nfd, File): - return False, "new path not working after rename" - return True, "rename worked" - - def test_unlink(vol, path, data): - mypath = path + ".tmp" - rc = vol.unlink(mypath) - if rc < 0: - return False, "unlink error %d" % rc - - try: - with vol.open(mypath, os.O_RDWR) as fd: - return False, "old path working (%s) after unlink" % fd - except OSError: - pass - else: - return False, "path still usable after unlink" - - return True, "unlink worked" - - def test_mkdir(vol, path, data): - mypath = path + ".dir" - rc = vol.mkdir(mypath) - if rc < 0: - return False, "mkdir error %d" % rc - - return True, "mkdir worked" - - def test_create_in_dir(vol, path, data): - mypath = path + ".dir/probe" - with vol.creat(mypath, os.O_RDWR, 0644) as fd: - if not isinstance(fd, File): - return False, "create (in dir) error" - return True, "create (in dir) worked" - - def test_dir_listing(vol, path, data): - mypath = path + ".dir" - fd = vol.opendir(mypath) - if not isinstance(fd, Dir): - return False, "opendir error %d" % fd - files = [] - while True: - ent = fd.next() - if not isinstance(ent, Dirent): - break - name = ent.d_name[:ent.d_reclen] - files.append(name) - if files != [".", "..", "probe"]: - return False, "wrong directory contents" - return True, "directory listing worked" - - def test_unlink_in_dir(vol, path, data): - mypath = path + ".dir/probe" - rc = vol.unlink(mypath) - if rc < 0: - return False, "unlink (in dir) error %d" % rc - return True, "unlink (in dir) worked" - - def test_rmdir(vol, path, data): - mypath = path + ".dir" - rc = vol.rmdir(mypath) - if rc < 0: - return False, "rmdir error %d" % rc - try: - vol.lstat(mypath) - except OSError: - pass - else: - return False, "dir still there after rmdir" - return True, "rmdir worked" - - def test_setxattr(vol, path, data): - mypath = path + ".xa" - with vol.creat(mypath, os.O_RDWR | os.O_EXCL, 0644) as fd: - if not fd: - return False, "creat (xattr test) error" - - key1, key2 = "hello", "goodbye" - if vol.setxattr(mypath, "trusted.key1", key1, len(key1)) < 0: - return False, "setxattr (key1) error" - if vol.setxattr(mypath, "trusted.key2", key2, len(key2)) < 0: - return False, "setxattr (key2) error" - return True, "setxattr worked" - - def test_getxattr(vol, path, data): - mypath = path + ".xa" - buf = vol.getxattr(mypath, "trusted.key1", 32) - if isinstance(buf, types.IntType): - return False, "getxattr error" - if buf != "hello": - return False, "wrong getxattr value %s" % buf - return True, "getxattr worked" - - def test_listxattr(vol, path, data): - mypath = path + ".xa" - xattrs = vol.listxattr(mypath) - if isinstance(xattrs, types.IntType): - return False, "listxattr error" - if xattrs != ["trusted.key1", "trusted.key2"]: - return False, "wrong listxattr value %s" % repr(xattrs) - return True, "listxattr worked" - - def test_fallocate(vol, path, data): - mypath = path + ".io" - with vol.creat(mypath, os.O_WRONLY | os.O_EXCL, 0644) as fd: - if not fd: - return False, "creat error" - rc = fd.fallocate(0, 0, 1024 * 1024) - if rc != 0: - return False, "fallocate error" - rc = fd.discard(4096, 4096) - if rc != 0: - return False, "discard error" - return True, "fallocate/discard worked" - - test_list = ( - test_create_write, - #test_open_read, - test_lstat, - test_rename, - test_unlink, - test_mkdir, - test_create_in_dir, - test_dir_listing, - test_unlink_in_dir, - test_rmdir, - test_setxattr, - test_getxattr, - test_listxattr, - test_fallocate, - ) - - ok_to_fail = ( - # TBD: this fails opening the new file, even though the file - # did get renamed. Looks like a gfapi bug, not ours. - (test_rename, "new path not working after rename"), - # TBD: similar, call returns error even though it worked - (test_rmdir, "dir still there after rmdir"), - ) - - volid, path = sys.argv[1:3] - data = "fubar" - vol = Volume("localhost", volid) - vol.set_logging("/dev/null", 7) - #vol.set_logging("/dev/stderr",7) - vol.mount() - - failures = 0 - expected = 0 - for t in test_list: - rc, msg = t(vol, path, data) - if rc: - print "PASS: %s" % msg - else: - print "FAIL: %s" % msg - failures += 1 - for otf in ok_to_fail: - if (t == otf[0]) and (msg == otf[1]): - print " (skipping known failure)" - expected += 1 - break # from the *inner* for loop - else: - break # from the *outer* for loop - - print "%d failures (%d expected)" % (failures, expected) -- cgit