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 datetime import timezone 

11from shutil import rmtree 

12from typing import Any 

13 

14from src import DEFAULT_REQUEST_TIMEOUT_SEC, 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 # iCloudPy produces date_modified via strptime(..., "%Y-%m-%dT%H:%M:%SZ") — always 

38 # naive UTC with no tzinfo. replace(tzinfo=UTC) is the correct conversion. 

39 remote_file_modified_time = int(item.date_modified.replace(tzinfo=timezone.utc).timestamp()) 

40 local_file_size = os.path.getsize(local_file) 

41 remote_file_size = item.size 

42 

43 if local_file_modified_time == remote_file_modified_time and ( 

44 local_file_size == remote_file_size 

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

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

47 ): 

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

49 return True 

50 

51 LOGGER.debug( 

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

53 + f"remote_modified_time is {remote_file_modified_time}, " 

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

55 ) 

56 return False 

57 

58 

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

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

61 

62 Args: 

63 item: iCloud package item with date_modified and size attributes 

64 local_package_path: Path to the local package directory 

65 

66 Returns: 

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

68 """ 

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

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

71 return False 

72 

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

74 # iCloudPy produces date_modified via strptime(..., "%Y-%m-%dT%H:%M:%SZ") — always 

75 # naive UTC with no tzinfo. replace(tzinfo=UTC) is the correct conversion. 

76 remote_package_modified_time = int(item.date_modified.replace(tzinfo=timezone.utc).timestamp()) 

77 

78 # Only date_modified is comparable here. A package is stored remotely as a zip and 

79 # unpacked locally, so item.size (the zip) and the summed size of the unpacked 

80 # directory are never equal — e.g. one .pxm is 7,821,180 bytes zipped and 

81 # 11,716,516 unpacked. Requiring size equality made this branch unreachable, so 

82 # every package was rmtree'd and re-downloaded on every single sync. 

83 if local_package_modified_time == remote_package_modified_time: 

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

85 return True 

86 

87 LOGGER.info( 

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

89 + f"remote_modified_time is {remote_package_modified_time} " 

90 + f"for package {local_package_path}.", 

91 ) 

92 rmtree(local_package_path) 

93 return False 

94 

95 

96def package_bundle_unchanged(item: Any, local_file: str) -> bool: 

97 """Freshness check for a package stored as a flat single-file bundle. 

98 

99 A flattened package -- an unrecognised-mime package kept as-is, or any 

100 package downloaded with ``flatten_packages: true`` -- lands on disk as a 

101 single file whose byte count is the size of the package-download archive, 

102 NOT ``item.size`` (the package's *logical* size that iCloud reports). So 

103 ``file_exists`` always sees a spurious size mismatch for these and 

104 re-downloads the bundle on every sync. 

105 

106 ``download_file`` stamps the bundle's mtime to ``item.date_modified`` via 

107 ``os.utime``, and iCloud bumps ``date_modified`` whenever a package's 

108 contents change, so the mtime is the reliable change signal. Match on mtime 

109 alone (the same mtime comparison ``file_exists`` uses), skipping the 

110 unusable size check. 

111 

112 Args: 

113 item: iCloud package item with a ``date_modified`` attribute 

114 local_file: Path to the local single-file bundle 

115 

116 Returns: 

117 True if the bundle is present and its mtime matches the remote 

118 (unchanged -- skip the re-download), False otherwise. 

119 """ 

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

121 return False 

122 # iCloudPy produces date_modified as naive UTC (strptime ...Z); replace(tzinfo=UTC) 

123 # matches the convention file_exists/package_exists use so mtimes are TZ-invariant. 

124 return int(os.path.getmtime(local_file)) == int(item.date_modified.replace(tzinfo=timezone.utc).timestamp()) 

125 

126 

127def is_package(item: Any, timeout: int = DEFAULT_REQUEST_TIMEOUT_SEC) -> bool: 

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

129 

130 Args: 

131 item: iCloud item to check 

132 timeout: HTTP read timeout in seconds (default: DEFAULT_REQUEST_TIMEOUT_SEC) 

133 

134 Returns: 

135 True if item is a package, False otherwise 

136 """ 

137 file_is_a_package = False 

138 try: 

139 with item.open(stream=True, timeout=timeout) as response: 

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

141 except Exception as e: 

142 # Enhanced error logging with file context 

143 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException 

144 error_msg = str(e) 

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

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

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

148 else: 

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

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

151 file_is_a_package = False 

152 return file_is_a_package