self.mount_a.stat(self.dir_name)
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(f"dd if=/dev/urandom {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)
- pwd = self.get_cephfs_shell_cmd_output('!pwd')
- 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 {path.join(pwd, tempdirname, file_)}")
-
def test_get_with_target_name(self):
"""
Test that get passes with target name