github/opencas-test-framework
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

User-defined timeout for copy and move

Browse Source 
Signed-off-by: Michal Mielewczyk <michal.mielewczyk@huawei.com>
This commit is contained in:
Michal Mielewczyk 2024-10-03 09:06:06 +02:00
parent b5a6cfbb93
commit cb0ac3ba3a
2 changed files with 8 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
test_tools/fs_utils.py
View File

@ -108,17 +108,18 @@ def copy(source: str,
destination: str, destination: str,
force: bool = False, force: bool = False,
recursive: bool = False, recursive: bool = False,
dereference: bool = False): dereference: bool = False,
timeout: timedelta = timedelta(minutes=30)):
cmd = f"cp{' --force' if force else ''}" \ cmd = f"cp{' --force' if force else ''}" \
f"{' --recursive' if recursive else ''}" \ f"{' --recursive' if recursive else ''}" \
f"{' --dereference' if dereference else ''} " \ f"{' --dereference' if dereference else ''} " \
f"{source} {destination}" f"{source} {destination}"
return TestRun.executor.run_expect_success(cmd) return TestRun.executor.run_expect_success(cmd, timeout)
def move(source, destination, force: bool = False): def move(source, destination, force: bool = False, timeout: timedelta = timedelta(minutes=30)):
cmd = f"mv{' --force' if force else ''} \"{source}\" \"{destination}\"" cmd = f"mv{' --force' if force else ''} \"{source}\" \"{destination}\""
return TestRun.executor.run_expect_success(cmd) return TestRun.executor.run_expect_success(cmd, timeout)
def remove(path: str, force: bool = False, recursive: bool = False, ignore_errors: bool = False): def remove(path: str, force: bool = False, recursive: bool = False, ignore_errors: bool = False):

5
test_utils/filesystem/file.py
View File

@ -54,8 +54,9 @@ class File(FsItem):
destination, destination,
force: bool = False, force: bool = False,
recursive: bool = False, recursive: bool = False,
dereference: bool = False): dereference: bool = False,
fs_utils.copy(str(self), destination, force, recursive, dereference) timeout: timedelta = timedelta(minutes=30)):
fs_utils.copy(str(self), destination, force, recursive, dereference, timeout)
if fs_utils.check_if_directory_exists(destination): if fs_utils.check_if_directory_exists(destination):
path = f"{destination}{'/' if destination[-1] != '/' else ''}{self.name}" path = f"{destination}{'/' if destination[-1] != '/' else ''}{self.name}"
else: else: