From 3c7bcfb082b56a47d4598091ba0eaddf58002941 Mon Sep 17 00:00:00 2001 From: themylogin Date: Wed, 8 May 2024 18:16:44 +0200 Subject: [PATCH] Simplify `File` artifact implementation (cherry picked from commit 0fae0d467059b127968096e9ba9d4410556c62f7) --- ixdiagnose/artifacts/items/file.py | 31 ++++++++++-------------------- 1 file changed, 10 insertions(+), 21 deletions(-) diff --git a/ixdiagnose/artifacts/items/file.py b/ixdiagnose/artifacts/items/file.py index 54268f0..257b6be 100644 --- a/ixdiagnose/artifacts/items/file.py +++ b/ixdiagnose/artifacts/items/file.py @@ -1,5 +1,5 @@ -import contextlib import os +import pathlib import shutil from ixdiagnose.utils.io import truncate_file @@ -13,20 +13,19 @@ class File(Item): def __init__(self, name: str, max_size: Optional[int] = None, truncate: Optional[bool] = True): super().__init__(name, max_size) self.truncate: bool = truncate - self.file_descriptor: Optional[int] = None - - def initialize_context(self, item_path: str): - self.file_descriptor = os.open(item_path, os.O_RDONLY | os.O_NOFOLLOW) def size(self, item_path: str) -> int: - return os.fstat(self.file_descriptor).st_size + try: + return pathlib.Path(item_path).lstat().st_size + except Exception: + return 0 def copy_impl(self, item_path: str, destination_path: str) -> list: - shutil.copy2(self.get_fd_path(), destination_path) - return [item_path] - - def get_fd_path(self) -> str: - return os.path.join('/proc/self/fd', str(self.file_descriptor)) + try: + shutil.copy2(item_path, destination_path) + return [item_path] + except Exception: + return [] def to_be_copied_checks(self, item_path: str) -> Tuple[bool, Optional[str]]: to_copy, error = (True, None) if os.path.isfile(item_path) else (False, f'{item_path!r} is not a file') @@ -34,17 +33,7 @@ def to_be_copied_checks(self, item_path: str) -> Tuple[bool, Optional[str]]: return self.size_check(item_path) return to_copy, error - def close_descriptor(self): - if self.file_descriptor: - with contextlib.suppress(OSError): - os.close(self.file_descriptor) - self.file_descriptor = None - - def __del__(self): - self.close_descriptor() - def post_copy_hook(self, destination_path: str): - self.close_descriptor() if self.max_size is None: return