(self)
| 31 | fail_release: bool = False |
| 32 | |
| 33 | def _acquire(self) -> None: # pragma: needs fork |
| 34 | self.descriptor = self._context.lock_file_fd = os.open(self.lock_file, os.O_CREAT | os.O_RDWR, 0o600) |
| 35 | if self.acquire_error is not None: |
| 36 | raise self.acquire_error |
| 37 | |
| 38 | def _release(self) -> None: # pragma: needs fork |
| 39 | if self.fail_release: |
nothing calls this directly
no outgoing calls
no test coverage detected