"""testcases for afr/basic """ import sys import time import os import hostutils import managerutils import glusterutils import atfutils import clientutils import serverutils import parser import validate from atfutils import commands write_op = ">" append_op = ">>" mounts = ['mount1'] bricks = ['brick1', 'brick2'] def reset_testenv(): """ * Unmount all mounts specified in the testenv file. * Stop the Volume (Active Volume) * Delete the Volume * Stop Glusterd on all servers * Remove the glusterd dir (/etc/glusterd) * Remove all glusterd logs from all servers """ output = clientutils.umountall() assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status glusterutils.volume_stop("server1", force=True) glusterutils.volume_delete("server1") glusterutils.glusterd_stop_allservers() glusterutils.glusterd_remove_dir_allservers() glusterutils.glusterd_remove_logs_allservers() return 0 def setup_testenv(): """ * Start glusterd on all servers * Peer Probe all servers in the storage pool * Create Bricks on the servers * Create Volume * Start Volume * Mount to the Volume from all mounts specified in testenv """ output = glusterutils.glusterd_start_allservers(force=True) assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status output = glusterutils.peer_probe("server1") assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status output = glusterutils.create_brick_allservers() assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status output = glusterutils.volume_create("server1") assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status output = glusterutils.volume_set("server1", key="diagnostics.client-log-level", value="DEBUG") output = glusterutils.volume_start("server1") assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status output = clientutils.mountall() assert_success_status = atfutils.assert_success(output["exitstatus"]) if assert_success_status is not 0: return assert_success_status return 0 def file_write(): """ Description: Test 'write' operation """ return_status = 1 data = "Hello World. This is testing write file operation" filename = "write.txt" command = ' '.join([commands['echo'], data, write_op, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['ls'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = filename return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_lookup(): """ Description: Test 'lookup' operation """ return_status = 1 filename = "stat.txt" command = ' '.join([commands['touch'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "regular empty file" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_read(): """ Description: Test 'read' operation """ return_status = 1 data = "Hello World. This is testing read file operation" filename = "read.txt" command = ' '.join([commands['echo'], data, write_op, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_append(): """ Description: Test 'append' operation """ return_status = 1 data1 = "Hello World. This is testing append file operation." data2 = "Appending data to file." filename = "append.txt" command = ' '.join([commands['echo'], data1, write_op, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = data1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['echo'], data2, append_op, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = str([data1 + "\n", data2 + "\n"]) return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_truncate(): """ """ return_status = 1 data = "Hello World. This is testing truncate file operation" filename = "truncate.txt" command = ' '.join([commands['echo'], data, write_op, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['truncate'], "-s 0", filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_hardlink(): """ """ return_status = 1 filename1 = "testlink.txt" filename2 = "hardlink_to_testlink.txt" data = "Hello World. This is testing file hard link operation" # Create a file command = ' '.join([commands['echo'], data, write_op, filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status # Check whether the file has been created on mount, bricks command = ' '.join([commands['cat'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status # Create Hardlink command = ' '.join([commands['hardlink'], filename1, filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status # Check whether the hardlink has been created on mount, bricks # Check the Link Count # Cat the data of new hard_link for filename in [filename1, filename2]: command = ' '.join([commands['stat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "Links: 2" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_symlink(): """ """ return_status = 1 filename1 = "testsymlink.txt" filename2 = "symlink_to_testsymlink.txt" data = "Hello World. This is testing file sym link operation" # Create a file command = ' '.join([commands['echo'], data, write_op, filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status # Check whether the file has been created on mount, bricks command = ' '.join([commands['cat'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status # Create Symlink command = ' '.join([commands['symlink'], filename1, filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status # Check whether the symlink has been created on mount, bricks # Check the Sym Link in stat structure # Cat the data of sym_link and output's the original file data command = ' '.join([commands['stat'], filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "symbolic link" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_set_get_fattr(): """ """ return_status = 1 name = "trusted.name" value = "set_get_attr_test" data = "Hello World. This is testing set_get_attr file operation" filename = "set_get_fattr.txt" command = ' '.join([commands['echo'], data, write_op, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['setattr'],"-n", name, "-v", value, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['getattr'], "-n", name, filename]) output = clientutils.execute_on_mount("mount1", command) expected_output = value return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_rename(): """ """ return_status = 1 data = "Hello World. This is testing rename file operation" filename1 = "rename_me.txt" filename2 = "rename_to.txt" command = ' '.join([commands['echo'], data, write_op, filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['ls'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = filename1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['rename'], filename1, filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['ls'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "No such file or directory" return_status = atfutils.validate_output(output, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 1, expected_output, stream="stderr") if return_status is not 0: return return_status command = ' '.join([commands['ls'], filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = filename2 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['cat'], filename2]) output = clientutils.execute_on_mount("mount1", command) expected_output = data return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def file_unlink(): """ """ return_status = 1 data = "Hello World. This is testing unlink file operation" filename1 = "unlink_me.txt" command = ' '.join([commands['echo'], data, write_op, filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['ls'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = filename1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['unlink_force'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['ls'], filename1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "No such file or directory" return_status = atfutils.validate_output(output, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_mkdir(): """ """ return_status = 1 dirname = "dir_mkdir" command = ' '.join([commands['mkdir'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_lookup(): """ """ return_status = 1 dirname = "dir_lookup" command = ' '.join([commands['mkdir'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = "directory" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_read(): """ """ return_status = 1 dirname = "dir_read" subdirlist = [] command = ' '.join([commands['mkdir'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status subdirs = os.path.join(dirname, "{1..4}") command = ' '.join([commands['mkdir'], subdirs]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status for x in range(1,5): subdirlist.append(str(x) + "\n") command = ' '.join([commands['ls'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = str(subdirlist) return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_hardlink(): """ """ return_status = 1 dirname1 = "dir_hlink" dirname2 = "hardlink_to_dir_hlink" command = ' '.join([commands['mkdir'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['hardlink'], dirname1, dirname2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "hard link not allowed for directory" return_status = atfutils.validate_output(output, 1, expected_output, stream="stderr") if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "No such file or directory" return_status = atfutils.validate_output(output, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_symlink(): """ """ return_status = 1 dirname1 = "dir_symlink" dirname2 = "symlink_to_dir_symlink" command = ' '.join([commands['mkdir'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['symlink'], dirname1, dirname2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "symbolic link" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_set_get_fattr(): """ """ return_status = 1 name = "trusted.name" value = "set_get_attr_test" dirname = "set_get_attr" command = ' '.join([commands['mkdir'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['setattr'], "-n", name, "-v", value, dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['getattr'], "-n", name, dirname]) output = clientutils.execute_on_mount("mount1", command) expected_output = value return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_rename(): """ """ return_status = 1 dirname1 = "dir_rename_me" dirname2 = "dir_rename_to" command = ' '.join([commands['mkdir'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['rename'], dirname1, dirname2]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "No such file or directory" return_status = atfutils.validate_output(output, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 1, expected_output, stream="stderr") if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname2]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname2 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0 def dir_unlink(): """ """ return_status = 1 dirname1 = "dir_unlink_me" command = ' '.join([commands['mkdir'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = dirname1 return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['unlink_dir_force'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "" return_status = atfutils.validate_output(output, 0, expected_output) if return_status is not 0: return return_status command = ' '.join([commands['stat'], dirname1]) output = clientutils.execute_on_mount("mount1", command) expected_output = "No such file or directory" return_status = atfutils.validate_output(output, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_on_bricks(bricks, command, 1, expected_output, stream="stderr") if return_status is not 0: return return_status return_status = validate.validate_md5sums(mounts, bricks) if return_status is not 0: return return_status return 0