Return free, held, or error without creating a missing lock file.
(path, record_lock)
| 74 | |
| 75 | |
| 76 | def lock_status(path, record_lock): |
| 77 | """Return free, held, or error without creating a missing lock file.""" |
| 78 | if not path.exists(): |
| 79 | return "free" |
| 80 | flags = os.O_RDWR | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) |
| 81 | try: |
| 82 | fd = os.open(str(path), flags) |
| 83 | except OSError: |
| 84 | return "error" |
| 85 | try: |
| 86 | status = os.fstat(fd) |
| 87 | if ( |
| 88 | not stat.S_ISREG(status.st_mode) |
| 89 | or status.st_uid != os.geteuid() |
| 90 | or status.st_nlink != 1 |
| 91 | or stat.S_IMODE(status.st_mode) != 0o600 |
| 92 | ): |
| 93 | return "error" |
| 94 | try: |
| 95 | if record_lock: |
| 96 | fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| 97 | fcntl.lockf(fd, fcntl.LOCK_UN) |
| 98 | else: |
| 99 | fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| 100 | fcntl.flock(fd, fcntl.LOCK_UN) |
| 101 | return "free" |
| 102 | except BlockingIOError: |
| 103 | return "held" |
| 104 | except OSError: |
| 105 | return "error" |
| 106 | finally: |
| 107 | os.close(fd) |
| 108 | |
| 109 | |
| 110 | def process_gone_or_zombie(pid): |
no test coverage detected