Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_util.py: 35%

1from __future__ import annotations 

2 

3import os 

4import stat 

5import sys 

6from errno import EACCES, EISDIR 

7from pathlib import Path 

8 

9 

10def raise_on_not_writable_file(filename: str) -> None: 

11 """ 

12 Raise an exception if attempting to open the file for writing would fail. 

13 

14 This is done so files that will never be writable can be separated from files that are writable but currently 

15 locked. 

16 

17 :param filename: file to check 

18 

19 :raises OSError: as if the file was opened for writing. 

20 

21 """ 

22 try: # use stat to do exists + can write to check without race condition 

23 file_stat = os.stat(filename) # noqa: PTH116 

24 except OSError: 

25 return # swallow does not exist or other errors 

26 

27 if file_stat.st_mtime != 0: # if os.stat returns but modification is zero that's an invalid os.stat - ignore it 

28 if not (file_stat.st_mode & stat.S_IWUSR): 

29 raise PermissionError(EACCES, "Permission denied", filename) 

30 

31 if stat.S_ISDIR(file_stat.st_mode): 

32 if sys.platform == "win32": # pragma: win32 cover 

33 # On Windows, this is PermissionError 

34 raise PermissionError(EACCES, "Permission denied", filename) 

35 else: # pragma: win32 no cover # noqa: RET506 

36 # On linux / macOS, this is IsADirectoryError 

37 raise IsADirectoryError(EISDIR, "Is a directory", filename) 

38 

39 

40def ensure_directory_exists(filename: Path | str) -> None: 

41 """ 

42 Ensure the directory containing the file exists (create it if necessary). 

43 

44 :param filename: file. 

45 

46 """ 

47 Path(filename).parent.mkdir(parents=True, exist_ok=True) 

48 

49 

50def break_lock_file(lock_file: str, mtime_before: float, ino_before: int) -> None: 

51 """ 

52 Atomically break a stale lock file that was judged stale at modification time *mtime_before*. 

53 

54 The file is renamed to a process-private name before being unlinked, so two processes breaking the same lock 

55 cannot delete each other's work (only one rename of a given inode succeeds; the loser gets ``OSError``). After the 

56 rename the file is re-checked: a newer modification time, or a different inode than *ino_before*, means a peer 

57 recreated the lock between the stale decision and the rename, so we grabbed a live file and must abort, leaving the 

58 renamed file in place rather than rolling back (a rollback rename is itself racy — same trade-off as the soft 

59 read/write marker break). The inode check matters because filesystems with coarse modification-time granularity 

60 (NFS, FAT) can give a same-second recreation the old mtime, so mtime alone would not catch it and a live lock would 

61 be unlinked; the inode is the reliable identity, mirroring the token re-check in the soft read/write marker break. 

62 ``lstat`` is used so a hostile symlink swapped in after the decision is not followed. 

63 

64 :param lock_file: path to the lock file to break. 

65 :param mtime_before: modification time observed when the lock was judged stale. 

66 :param ino_before: inode number observed when the lock was judged stale. 

67 

68 :raises OSError: if the rename fails (e.g. the file vanished or is not owned in a sticky directory). 

69 

70 """ 

71 break_path = f"{lock_file}.break.{os.getpid()}" 

72 Path(lock_file).rename(break_path) 

73 try: 

74 st_after = os.lstat(break_path) 

75 except OSError: 

76 return 

77 if st_after.st_mtime > mtime_before or st_after.st_ino != ino_before: 

78 return 

79 Path(break_path).unlink() 

80 

81 

82__all__ = [ 

83 "break_lock_file", 

84 "ensure_directory_exists", 

85 "raise_on_not_writable_file", 

86]