# Copyright (C) 2020 Red Hat, Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. # pylint: disable=too-many-locals # pylint: disable=too-many-branches,too-many-statements,too-many-function-args from glusto.core import Glusto as g from glustolibs.gluster.exceptions import ExecutionError from glustolibs.gluster.gluster_base_class import (GlusterBaseClass, runs_on) from glustolibs.gluster.glusterfile import (get_fattr, set_fattr, create_link_file, delete_fattr) from glustolibs.gluster.glusterdir import get_dir_contents from glustolibs.gluster.brick_libs import get_all_bricks from glustolibs.gluster.lib_utils import (append_string_to_file) from glustolibs.gluster.dht_test_utils import validate_files_in_dir import glustolibs.gluster.constants as k @runs_on([['distributed', 'distributed-replicated', 'distributed-dispersed', 'distributed-arbiter'], ['glusterfs']]) class TestDhtCustomXattrClass(GlusterBaseClass): def check_custom_xattr_visible(self, xattr_val): """ Check custom xttar from mount point and on bricks. """ # Check custom xattr from mount point for mount_object in self.mounts: for fname in self.files_and_soft_links: attr_val = get_fattr(mount_object.client_system, fname, 'user.foo') self.assertEqual(attr_val, xattr_val, "Custom xattr not found from mount.") g.log.info("Custom xattr found on mount point.") # Check custom xattr on bricks for brick in get_all_bricks(self.mnode, self.volname): node, brick_path = brick.split(':') files_on_bricks = get_dir_contents(node, brick_path) files = [ fname.split('/')[3] for fname in self.list_of_files if fname.split('/')[3] in files_on_bricks] for fname in files: attr_val = get_fattr(node, "{}/{}".format(brick_path, fname), 'user.foo') self.assertEqual(attr_val, xattr_val, "Custom xattr not visible on bricks") g.log.info("Custom xattr found on bricks.") def delete_xattr_user_foo(self, list_of_files): """ Removes xattr user.foo from all the files. """ for fname in list_of_files: ret = delete_fattr(self.client_node, fname, 'user.foo') self.assertTrue(ret, "Unable to remove custom xattr for " "file {}".format(fname)) g.log.info("Successfully removed custom xattr for each file.") def set_xattr_user_foo(self, list_of_files, xattr_val): """ sets xattr user.foo on all the files. """ for fname in list_of_files: ret = set_fattr(self.client_node, fname, 'user.foo', xattr_val) self.assertTrue(ret, "Unable to create custom xattr " "for file {}".format(fname)) g.log.info("Successfully created a custom xattr for all files.") def check_for_trusted_glusterfs_pathinfo(self, list_of_files): """ Check if trusted.glusterfs.pathinfo is visible. """ for fname in list_of_files: ret = get_fattr(self.client_node, fname, 'trusted.glusterfs.pathinfo') self.assertIsNotNone(ret, "pathinfo not visible") g.log.info("Mount point shows pathinfo xattr for " "all files") def check_mount_point_and_bricks_for_xattr(self, list_of_all_files): """ Check xattr on mount point and bricks. """ # Check if xattr is visable from mount point for mount_object in self.mounts: for fname in list_of_all_files: ret = get_fattr(mount_object.client_system, fname, 'user.foo') self.assertIsNone(ret, "Custom attribute visible at mount " "point even after deletion") # Check if xattr is visable from bricks for brick in get_all_bricks(self.mnode, self.volname): node, brick_path = brick.split(':') files_on_bricks = get_dir_contents(node, brick_path) files = [ fname.split('/')[3] for fname in self.list_of_files if fname.split('/')[3] in files_on_bricks] for fname in files: ret = get_fattr(node, "{}/{}".format(brick_path, fname), 'user.foo') self.assertIsNone(ret, "Custom attribute visible on " "brick even after deletion") g.log.info("Custom xattr for file is not visible on " "mount point and bricks") def setUp(self): # Calling GlusterBaseClass setUp self.get_super_method(self, 'setUp')() # Setup Volume and Mount Volume ret = self.setup_volume_and_mount_volume(self.mounts) if not ret: raise ExecutionError("Failed to Setup_Volume and Mount_Volume") g.log.info("Successful in Setup Volume and Mount Volume") def tearDown(self): # Cleanup Volume ret = self.unmount_volume_and_cleanup_volume(self.mounts) if not ret: raise ExecutionError("volume clean up failed") g.log.info("Successful in cleaning up Volume %s", self.volname) self.get_super_method(self, 'tearDown')() def test_dht_custom_xattr(self): """ Test case: 1.Create a gluster volume and start it. 2.Create file and link files. 3.Create a custom xattr for file. 4.Verify that xattr for file is displayed on mount point and bricks 5.Modify custom xattr value and verify that xattr for file is displayed on mount point and bricks 6.Verify that custom xattr is not displayed once you remove it 7.Create a custom xattr for symbolic link. 8.Verify that xattr for symbolic link is displayed on mount point and sub-volume 9.Modify custom xattr value and verify that xattr for symbolic link is displayed on mount point and bricks 10.Verify that custom xattr is not displayed once you remove it. """ # Initializing variables mount_point = self.mounts[0].mountpoint self.client_node = self.mounts[0].client_system self.list_of_files, list_of_softlinks = [], [] list_of_hardlinks = [] for number in range(1, 3): # Create regular files fname = '{0}/regular_file_{1}'.format(mount_point, str(number)) ret = append_string_to_file(self.client_node, fname, 'Sample content for file.') self.assertTrue(ret, "Unable to create regular file " "{}".format(fname)) self.list_of_files.append(fname) # Create hard link for file hardlink = '{0}/link_file_{1}'.format(mount_point, str(number)) ret = create_link_file(self.client_node, fname, hardlink) self.assertTrue(ret, "Unable to create hard link file " "{}".format(hardlink)) list_of_hardlinks.append(hardlink) # Create soft link for file softlink = '{0}/symlink_file_{1}'.format(mount_point, str(number)) ret = create_link_file(self.client_node, fname, softlink, soft=True) self.assertTrue(ret, "Unable to create symlink file " "{}".format(softlink)) list_of_softlinks.append(softlink) self.files_and_soft_links = self.list_of_files + list_of_softlinks # Check if files are created on the right subvol ret = validate_files_in_dir( self.client_node, mount_point, file_type=k.FILETYPE_FILES, test_type=k.TEST_FILE_EXISTS_ON_HASHED_BRICKS) self.assertTrue(ret, "Files not created on correct sub-vols") g.log.info("Files are on correct sub-vols according to " "the hash value") # Set custom xattr on all the regular files self.set_xattr_user_foo(self.list_of_files, 'bar2') # Check if custom xattr is set to all the regular files self.check_custom_xattr_visible('bar2') # Change the custom xattr on all the regular files self.set_xattr_user_foo(self.list_of_files, 'ABC') # Check if xattr is set to all the regular files self.check_custom_xattr_visible('ABC') # Delete Custom xattr from all regular files self.delete_xattr_user_foo(self.list_of_files) # Check mount point and brick for the xattr list_of_all_files = list_of_hardlinks + self.files_and_soft_links self.check_mount_point_and_bricks_for_xattr(list_of_all_files) # Check if pathinfo xattr is visible self.check_for_trusted_glusterfs_pathinfo(self.list_of_files) # Set custom xattr on all the regular files self.set_xattr_user_foo(list_of_softlinks, 'bar2') # Check if custom xattr is set to all the regular files self.check_custom_xattr_visible('bar2') # Change the custom xattr on all the regular files self.set_xattr_user_foo(list_of_softlinks, 'ABC') # Check if xattr is set to all the regular files self.check_custom_xattr_visible('ABC') # Delete Custom xattr from all regular files self.delete_xattr_user_foo(list_of_softlinks) # Check mount point and brick for the xattr self.check_mount_point_and_bricks_for_xattr(list_of_all_files) # Check if pathinfo xattr is visible self.check_for_trusted_glusterfs_pathinfo(list_of_softlinks)