From: Rishabh Dave Date: Fri, 8 Jul 2022 13:56:34 +0000 (+0530) Subject: Merge pull request #45998 from dparmar18/implement_ln_cephfs_shell X-Git-Tag: v18.0.0~541 X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=9f95f96dc0d4c6631ffbd8c0d0c120c91f2b74bc;p=ceph-ci.git Merge pull request #45998 from dparmar18/implement_ln_cephfs_shell cephfs-shell: ln command implementation Reviewed-by: Nikhilkumar Shelke Reviewed-by: Venky Shankar Reviewed-by: Anthony D'Atri Reviewed-by: Rishabh Dave --- 9f95f96dc0d4c6631ffbd8c0d0c120c91f2b74bc diff --cc qa/tasks/cephfs/test_cephfs_shell.py index ac869fcea6c,cb15a7a1ab8..e658ab55117 --- a/qa/tasks/cephfs/test_cephfs_shell.py +++ b/qa/tasks/cephfs/test_cephfs_shell.py @@@ -334,7 -335,113 +335,76 @@@ class TestRmdir(TestCephFSShell) self.mount_a.stat(self.dir_name) + class TestLn(TestCephFSShell): + dir1 = 'test_dir1' + dir2 = 'test_dir2' + dump_id = 11 + s = 'somedata' + dump_file = 'dump11' + + def test_soft_link_without_link_name(self): + self.run_cephfs_shell_cmd(f'mkdir -p {self.dir1}/{self.dir2}') + self.mount_a.write_file(path=f'{self.dir1}/{self.dump_file}', + data=self.s) + self.run_cephfs_shell_script(script=dedent(f''' + cd /{self.dir1}/{self.dir2} + ln -s ../{self.dump_file}''')) + o = self.get_cephfs_shell_cmd_output(f'cat /{self.dir1}/{self.dir2}' + f'/{self.dump_file}') + self.assertEqual(self.s, o) + + def test_soft_link_with_link_name(self): + self.run_cephfs_shell_cmd(f'mkdir -p {self.dir1}/{self.dir2}') + self.mount_a.write_file(path=f'{self.dir1}/{self.dump_file}', + data=self.s) + self.run_cephfs_shell_cmd(f'ln -s /{self.dir1}/{self.dump_file} ' + f'/{self.dir1}/{self.dir2}/') + o = self.get_cephfs_shell_cmd_output(f'cat /{self.dir1}/{self.dir2}' + f'/{self.dump_file}') + self.assertEqual(self.s, o) + + def test_hard_link_without_link_name(self): + self.run_cephfs_shell_cmd(f'mkdir -p {self.dir1}/{self.dir2}') + self.mount_a.write_file(path=f'{self.dir1}/{self.dump_file}', + data=self.s) + self.run_cephfs_shell_script(script=dedent(f''' + cd /{self.dir1}/{self.dir2} + ln ../{self.dump_file}''')) + o = self.get_cephfs_shell_cmd_output(f'cat /{self.dir1}/{self.dir2}' + f'/{self.dump_file}') + self.assertEqual(self.s, o) + + def test_hard_link_with_link_name(self): + self.run_cephfs_shell_cmd(f'mkdir -p {self.dir1}/{self.dir2}') + self.mount_a.write_file(path=f'{self.dir1}/{self.dump_file}', + data=self.s) + self.run_cephfs_shell_cmd(f'ln /{self.dir1}/{self.dump_file} ' + f'/{self.dir1}/{self.dir2}/') + o = self.get_cephfs_shell_cmd_output(f'cat /{self.dir1}/{self.dir2}' + f'/{self.dump_file}') + self.assertEqual(self.s, o) + + def test_hard_link_to_dir_not_allowed(self): + self.run_cephfs_shell_cmd(f'mkdir {self.dir1}') + self.run_cephfs_shell_cmd(f'mkdir {self.dir2}') + r = self.run_cephfs_shell_cmd(f'ln /{self.dir1} /{self.dir2}/', + check_status=False) + self.assertEqual(r.returncode, 3) + + def test_target_exists_in_dir(self): + self.mount_a.write_file(path=f'{self.dump_file}', data=self.s) + r = self.run_cephfs_shell_cmd(f'ln {self.dump_file} {self.dump_file}', + check_status=False) + self.assertEqual(r.returncode, 1) + + def test_incorrect_dir(self): + self.mount_a.write_file(path=f'{self.dump_file}', data=self.s) + r = self.run_cephfs_shell_cmd(f'ln {self.dump_file} /dir1/', + check_status=False) + self.assertEqual(r.returncode, 5) + + class TestGetAndPut(TestCephFSShell): - def test_without_target_dir(self): - """ - Test put and get commands without target path. - """ - tempdir = self.mount_a.client_remote.mkdtemp() - tempdirname = path.basename(tempdir) - files = ('dump1', 'dump2', 'dump3', tempdirname) - - for i, file_ in enumerate(files[: -1]): - size = i + 1 - ofarg = 'of=' + path.join(tempdir, file_) - bsarg = 'bs=' + str(size) + 'M' - self.mount_a.run_shell_payload("dd if=/dev/urandom " - f"{ofarg} {bsarg} count=1") - - self.run_cephfs_shell_cmd('put ' + tempdir) - for file_ in files: - if file_ == tempdirname: - self.mount_a.stat(path.join(self.mount_a.mountpoint, file_)) - else: - self.mount_a.stat(path.join(self.mount_a.mountpoint, - tempdirname, file_)) - - self.mount_a.run_shell_payload(f"rm -rf {tempdir}") - - self.run_cephfs_shell_cmd('get ' + tempdirname) - # NOTE: cwd=None because we want to run it at CWD, not at cephfs mntpt. - pwd = self.mount_a.run_shell('pwd', cwd=None).stdout.getvalue(). \ - strip() - for file_ in files: - if file_ == tempdirname: - self.mount_a.run_shell_payload(f"stat {path.join(pwd, file_)}") - else: - self.mount_a.run_shell_payload( - f"stat " - f"{path.join(pwd, tempdirname, file_)}") - def test_get_with_target_name(self): """ Test that get passes with target name