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 is_package(item: Any, timeout: int = DEFAULT_REQUEST_TIMEOUT_SEC) -> bool: 

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

98 

99 Args: 

100 item: iCloud item to check 

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

102 

103 Returns: 

104 True if item is a package, False otherwise 

105 """ 

106 file_is_a_package = False 

107 try: 

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

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

110 except Exception as e: 

111 # Enhanced error logging with file context 

112 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException 

113 error_msg = str(e) 

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

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

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

117 else: 

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

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

120 file_is_a_package = False 

121 return file_is_a_package