"""testcases for replicate/self-heal """ import sys import time import hostutils import managerutils import glusterutils import atfutils import clientutils import serverutils import parser import pdb def reset_testenv(): return_status = 1 if clientutils.umountall(): return return_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(): """ """ return_status = 1 if glusterutils.glusterd_start_allservers(force=True): return return_status if glusterutils.peer_probe("server1"): return return_status if glusterutils.create_brick_allservers(): return return_status if glusterutils.volume_create("server1"): return 1 glusterutils.volume_set("server1", key="diagnostics.client-log-level", value="DEBUG") if glusterutils.volume_start("server1"): return 1 if clientutils.mountall(): return 1 return 0 def test001(): """ Description: Test for failure when the entry does not exist """ return_status = 1 if reset_testenv(): return return_status if setup_testenv(): return return_status return_status = clientutils.execute_on_mount("mount1", "ls abcd") if return_status: test_status = 0 else: test_status = 1 return test_status def test002(): """ Description: Test for estale when the fs is stale """ return_status = 1 if reset_testenv(): return return_status if setup_testenv(): return return_status clientutils.execute_on_mount("mount1", "touch file") serverutils.execute_on_brick("brick1", "setfattr -n trusted.gfid -v 0sBfz5vAdHTEK1GZ99qjqTIg== file") return_status = clientutils.execute_on_mount("mount1", "find file | xargs stat") if return_status: test_status = 0 else: test_status = 1 return test_status