qa/cephfs: allow creating files at custom location

Concerned method: caps_helper.CapsHelper.write_test_file()

When CephFS mountpoint is changed inside a single test method, the test
files created at root are neither accessible nor useful. Therefore, add
an option to create the test files in the directory passed by the user.
This also increases general flexibility of the concerned method.

Signed-off-by: Rishabh Dave <ridave@redhat.com>

diff --git a/qa/tasks/cephfs/caps_helper.py b/qa/tasks/cephfs/caps_helper.py
index c26a2e5108d49bf55f93a0fd2bcc6a8b9bbdd431..397bdb6230502c2a604fd87cec88e34d859ba95e 100644 (file)
--- a/qa/tasks/cephfs/caps_helper.py
+++ b/qa/tasks/cephfs/caps_helper.py
@@ -9,7 +9,7 @@ from teuthology.orchestra.run import Raw
 
 class CapsHelper(CephFSTestCase):
 
-    def write_test_files(self, mounts):
+    def write_test_files(self, mounts, testpath=''):
         """
         Exercising 'r' and 'w' access levels on a file on CephFS mount is
         pretty routine across all tests for caps. Adding to method to write
@@ -20,9 +20,20 @@ class CapsHelper(CephFSTestCase):
         """
         filepaths, filedata = [], 'testdata'
         dirname, filename = 'testdir', 'testfile'
+        # XXX: The reason behind testpath[1:] below is that the testpath is
+        # supposed to contain a path inside CephFS (which might be passed as
+        # an absolute path). os.path.join() deletes all previous path
+        # components when it encounters a path component starting with '/'.
+        # Deleting the first '/' from the string in testpath ensures that
+        # previous path components are not deleted by os.path.join().
+        if testpath:
+            testpath = testpath[1:] if testpath[0] == '/' else testpath
+        # XXX: passing just '/' screw up os.path.join() ahead.
+        if testpath == '/':
+            testpath = ''
 
         for mount_x in mounts:
-            dirpath = os_path_join(mount_x.hostfs_mntpt, dirname)
+            dirpath = os_path_join(mount_x.hostfs_mntpt, testpath, dirname)
             mount_x.run_shell(f'mkdir {dirpath}')
             filepath = os_path_join(dirpath, filename)
             mount_x.write_file(filepath, filedata)
@@ -30,10 +41,10 @@ class CapsHelper(CephFSTestCase):
 
         return filepaths, (filedata,), mounts
 
-    def run_cap_tests(self, filepaths, filedata, mounts, perm):
+    def run_cap_tests(self, filepaths, filedata, mounts, perm, mntpt=None):
         # TODO
         #self.run_mon_cap_tests()
-        self.run_mds_cap_tests(filepaths, filedata, mounts, perm)
+        self.run_mds_cap_tests(filepaths, filedata, mounts, perm, mntpt=mntpt)
 
     def run_mon_cap_tests(self, moncap, keyring):
         keyring_path = self.fs.admin_remote.mktemp(data=keyring)
@@ -54,7 +65,16 @@ class CapsHelper(CephFSTestCase):
                 else:
                     self.assertNotIn('name: ' + fsname, fsls)
 
-    def run_mds_cap_tests(self, filepaths, filedata, mounts, perm):
+    def run_mds_cap_tests(self, filepaths, filedata, mounts, perm, mntpt=None):
+        # XXX: mntpt is path inside cephfs that serves as root for current
+        # mount. Therefore, this path must me deleted from filepaths.
+        # Example -
+        #   orignal path: /mnt/cephfs_x/dir1/dir2/testdir
+        #   cephfs dir serving as root for current mnt: /dir1/dir2
+        #   therefore, final path: /mnt/cephfs_x//testdir
+        if mntpt:
+            filepaths = [x.replace(mntpt, '') for x in filepaths]
+
         self.conduct_pos_test_for_read_caps(filepaths, filedata, mounts)
 
         if perm == 'rw':