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 pathlib import Path 

12from shutil import rmtree 

13from typing import Any 

14 

15from src import DEFAULT_REQUEST_TIMEOUT_SEC, configure_icloudpy_logging, get_logger 

16 

17# Configure icloudpy logging immediately after import 

18configure_icloudpy_logging() 

19 

20LOGGER = get_logger() 

21 

22 

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

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

25 

26 Args: 

27 item: iCloud file item with date_modified and size attributes 

28 local_file: Path to the local file 

29 

30 Returns: 

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

32 """ 

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

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

35 return False 

36 

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

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

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

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

41 local_file_size = os.path.getsize(local_file) 

42 remote_file_size = item.size 

43 

44 if local_file_modified_time == remote_file_modified_time and ( 

45 local_file_size == remote_file_size 

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

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

48 ): 

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

50 return True 

51 

52 LOGGER.debug( 

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

54 + f"remote_modified_time is {remote_file_modified_time}, " 

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

56 ) 

57 return False 

58 

59 

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

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

62 

63 Args: 

64 item: iCloud package item with date_modified and size attributes 

65 local_package_path: Path to the local package directory 

66 

67 Returns: 

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

69 """ 

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

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

72 return False 

73 

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

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

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

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

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

79 remote_package_size = item.size 

80 

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

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

83 return True 

84 

85 LOGGER.info( 

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

87 + f"remote_modified_time is {remote_package_modified_time}, " 

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

89 ) 

90 rmtree(local_package_path) 

91 return False 

92 

93 

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

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

96 

97 Args: 

98 item: iCloud item to check 

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

100 

101 Returns: 

102 True if item is a package, False otherwise 

103 """ 

104 file_is_a_package = False 

105 try: 

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

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

108 except Exception as e: 

109 # Enhanced error logging with file context 

110 # This catches all exceptions including iCloudPy errors like ObjectNotFoundException 

111 error_msg = str(e) 

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

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

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

115 else: 

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

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

118 file_is_a_package = False 

119 return file_is_a_package