Coverage for src/drive_file_existence.py: 100%

1"""File existence checking utilities. 

2 

3This module provides file and package existence checking functionality, 

4separating existence validation logic from sync operations per SRP. 

5""" 

6 

7__author__ = "Mandar Patil (mandarons@pm.me)" 

8 

9import os 

10from pathlib import Path 

11from shutil import rmtree 

12from typing import Any 

13 

14from src import configure_icloudpy_logging, get_logger 

15 

16# Configure icloudpy logging immediately after import 

17configure_icloudpy_logging() 

18 

19LOGGER = get_logger() 

20 

21 

22def file_exists(item: Any, local_file: str) -> bool: 

23 """Check if a file exists locally and is up-to-date. 

24 

25 Args: 

26 item: iCloud file item with date_modified and size attributes 

27 local_file: Path to the local file 

28 

29 Returns: 

30 True if file exists and is up-to-date, False otherwise 

31 """ 

32 if not (item and local_file and os.path.isfile(local_file)): 

33 LOGGER.debug(f"File {local_file} does not exist locally.") 

34 return False 

35 

36 local_file_modified_time = int(os.path.getmtime(local_file)) 

37 remote_file_modified_time = int(item.date_modified.timestamp()) 

38 local_file_size = os.path.getsize(local_file) 

39 remote_file_size = item.size 

40 

41 if local_file_modified_time == remote_file_modified_time and ( 

42 local_file_size == remote_file_size 

43 or (local_file_size == 0 and remote_file_size is None) 

44 or (local_file_size is None and remote_file_size == 0) 

45 ): 

46 LOGGER.debug(f"No changes detected. Skipping the file {local_file} ...") 

47 return True 

48 

49 LOGGER.debug( 

50 f"Changes detected: local_modified_time is {local_file_modified_time}, " 

51 + f"remote_modified_time is {remote_file_modified_time}, " 

52 + f"local_file_size is {local_file_size} and remote_file_size is {remote_file_size}.", 

53 ) 

54 return False 

55 

56 

57def package_exists(item: Any, local_package_path: str) -> bool: 

58 """Check if a package exists locally and is up-to-date. 

59 

60 Args: 

61 item: iCloud package item with date_modified and size attributes 

62 local_package_path: Path to the local package directory 

63 

64 Returns: 

65 True if package exists and is up-to-date, False otherwise 

66 """ 

67 if not (item and local_package_path and os.path.isdir(local_package_path)): 

68 LOGGER.debug(f"Package {local_package_path} does not exist locally.") 

69 return False 

70 

71 local_package_modified_time = int(os.path.getmtime(local_package_path)) 

72 remote_package_modified_time = int(item.date_modified.timestamp()) 

73 local_package_size = sum(f.stat().st_size for f in Path(local_package_path).glob("**/*") if f.is_file()) 

74 remote_package_size = item.size 

75 

76 if local_package_modified_time == remote_package_modified_time and local_package_size == remote_package_size: 

77 LOGGER.debug(f"No changes detected. Skipping the package {local_package_path} ...") 

78 return True 

79 

80 LOGGER.info( 

81 f"Changes detected: local_modified_time is {local_package_modified_time}, " 

82 + f"remote_modified_time is {remote_package_modified_time}, " 

83 + f"local_package_size is {local_package_size} and remote_package_size is {remote_package_size}.", 

84 ) 

85 rmtree(local_package_path) 

86 return False 

87 

88 

89def is_package(item: Any) -> bool: 

90 """Determine if an iCloud item is a package that needs special handling. 

91 

92 Args: 

93 item: iCloud item to check 

94 

95 Returns: 

96 True if item is a package, False otherwise 

97 """ 

98 file_is_a_package = False 

99 try: 

100 with item.open(stream=True) as response: 

101 file_is_a_package = response.url and "/packageDownload?" in response.url 

102 except Exception as e: 

103 # Enhanced error logging with file context 

104 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException 

105 error_msg = str(e) 

106 item_name = getattr(item, "name", "Unknown file") 

107 if "ObjectNotFoundException" in error_msg or "NOT_FOUND" in error_msg: 

108 LOGGER.error(f"File not found in iCloud Drive while checking package type - {item_name}: {error_msg}") 

109 else: 

110 LOGGER.error(f"Failed to check package type for {item_name}: {error_msg}") 

111 # Return False if we can't determine package type due to error 

112 file_is_a_package = False 

113 return file_is_a_package