Merge pull request #45998 from dparmar18/implement_ln_cephfs_shell

cephfs-shell: ln command implementation

Reviewed-by: Nikhilkumar Shelke <nshelke@redhat.com>
Reviewed-by: Venky Shankar <vshankar@redhat.com>
Reviewed-by: Anthony D'Atri <anthony.datri@gmail.com>
Reviewed-by: Rishabh Dave <ridave@redhat.com>

diff --cc doc/man/8/cephfs-shell.rst
Simple merge

diff --cc qa/tasks/cephfs/test_cephfs_shell.py
index ac869fcea6c6c0f9a58d910289cb2af7978947a6,cb15a7a1ab87a29b45b40c64e48739412f9430c2..e658ab5511766b520f63e1f52e087c1f843d5f68
--- 1/qa/tasks/cephfs/test_cephfs_shell.py
--- 2/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

diff --cc src/tools/cephfs/cephfs-shell
Simple merge