1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
from __future__ import annotations
import os
import socket
import sys
from errno import EIO, ENOSYS
from typing import TYPE_CHECKING
import pytest
from filelock import SoftFileLock, UnixFileLock
if TYPE_CHECKING:
from pathlib import Path
from unittest.mock import MagicMock
from pytest_mock import MockerFixture
unix_only = pytest.mark.skipif(sys.platform == "win32", reason="unix-only flock fallback")
_ENOSYS_SIDE_EFFECT = OSError(ENOSYS, "Function not implemented")
_FLOCK_PATCH_TARGET = "filelock._unix.fcntl.flock"
@unix_only
def test_fallback_emits_warning(tmp_path: Path, mocker: MockerFixture) -> None:
mocker.patch(_FLOCK_PATCH_TARGET, side_effect=_ENOSYS_SIDE_EFFECT)
lock = UnixFileLock(tmp_path / "test.lock")
with pytest.warns(UserWarning, match="flock not supported on this filesystem, falling back to SoftFileLock"):
lock.acquire()
lock.release()
@unix_only
@pytest.mark.filterwarnings("default::UserWarning")
def test_fallback_swaps_to_soft(tmp_path: Path, mocker: MockerFixture) -> None:
mocker.patch(_FLOCK_PATCH_TARGET, side_effect=_ENOSYS_SIDE_EFFECT)
lock = UnixFileLock(tmp_path / "test.lock")
with lock:
assert lock.is_locked
assert isinstance(lock, SoftFileLock)
@unix_only
@pytest.mark.filterwarnings("default::UserWarning")
def test_fallback_writes_pid_and_hostname(tmp_path: Path, mocker: MockerFixture) -> None:
lock_path = tmp_path / "test.lock"
mocker.patch(_FLOCK_PATCH_TARGET, side_effect=_ENOSYS_SIDE_EFFECT)
lock = UnixFileLock(lock_path)
with lock:
content = lock_path.read_text(encoding="utf-8")
assert content == f"{os.getpid()}\n{socket.gethostname()}\n"
@unix_only
@pytest.mark.filterwarnings("default::UserWarning")
def test_fallback_release_unlinks_file(tmp_path: Path, mocker: MockerFixture) -> None:
lock_path = tmp_path / "test.lock"
mocker.patch(_FLOCK_PATCH_TARGET, side_effect=_ENOSYS_SIDE_EFFECT)
lock = UnixFileLock(lock_path)
lock.acquire()
assert lock_path.exists()
lock.release()
assert not lock_path.exists()
@unix_only
@pytest.mark.filterwarnings("default::UserWarning")
def test_fallback_subsequent_acquire_skips_flock(tmp_path: Path, mocker: MockerFixture) -> None:
flock_mock: MagicMock = mocker.patch(_FLOCK_PATCH_TARGET, side_effect=_ENOSYS_SIDE_EFFECT)
lock = UnixFileLock(tmp_path / "test.lock")
lock.acquire()
lock.release()
flock_mock.reset_mock()
lock.acquire()
lock.release()
flock_mock.assert_not_called()
@unix_only
@pytest.mark.filterwarnings("default::UserWarning")
def test_fallback_reentrant_locking(tmp_path: Path, mocker: MockerFixture) -> None:
mocker.patch(_FLOCK_PATCH_TARGET, side_effect=_ENOSYS_SIDE_EFFECT)
lock = UnixFileLock(tmp_path / "test.lock")
with lock:
with lock:
assert lock.is_locked
assert lock.is_locked
assert not lock.is_locked
@unix_only
def test_release_suppresses_eio_on_close(tmp_path: Path, mocker: MockerFixture) -> None:
lock = UnixFileLock(tmp_path / "test.lock")
lock.acquire()
real_close = os.close
fd_to_fail = lock._context.lock_file_fd
def _close_eio(fd: int) -> None:
real_close(fd)
if fd == fd_to_fail:
raise OSError(EIO, "Input/output error")
mocker.patch("filelock._unix.os.close", side_effect=_close_eio)
lock.release()
assert not lock.is_locked
|
