|
25 | 25 | from pkgutil import resolve_name |
26 | 26 | from queue import Queue |
27 | 27 | from stat import S_IMODE, S_IWGRP, S_IWOTH, S_IWUSR, filemode |
28 | | -from types import CodeType, FrameType, TracebackType |
| 28 | +from types import CodeType, FrameType, SimpleNamespace, TracebackType |
29 | 29 | from typing import TYPE_CHECKING, Any, Final, Literal, cast, get_args |
30 | 30 | from uuid import uuid4 |
31 | 31 | from weakref import WeakValueDictionary |
|
64 | 64 |
|
65 | 65 | if TYPE_CHECKING: |
66 | 66 | from collections.abc import Callable, Generator, Iterator |
| 67 | + from unittest.mock import Mock |
67 | 68 |
|
68 | 69 | from _typeshed import Unused |
69 | 70 | from pytest_mock import MockerFixture |
@@ -729,48 +730,52 @@ def test_default_poll_interval(lock_type: type[BaseFileLock], tmp_path: Path) -> |
729 | 730 |
|
730 | 731 |
|
731 | 732 | @pytest.fixture |
732 | | -def steady_clock(mocker: MockerFixture) -> None: |
| 733 | +def steady_clock(mocker: MockerFixture) -> Mock: |
733 | 734 | # These tests assert the sleep the lock asked for. The lock caps that sleep at the time left before its timeout, so |
734 | 735 | # a stall on a loaded machine between the start and the first poll would shorten it. A clock that advances 10 ms |
735 | 736 | # per reading keeps the first poll inside the timeout regardless of load. |
736 | | - mocker.patch("filelock._api.time.perf_counter", side_effect=(reading / 100 for reading in count())) |
| 737 | + sleep_mock: Final = mocker.create_autospec(time.sleep) |
| 738 | + mocker.patch( |
| 739 | + "filelock._api.time", |
| 740 | + new=SimpleNamespace( |
| 741 | + perf_counter=mocker.create_autospec(time.perf_counter, side_effect=(reading / 100 for reading in count())), |
| 742 | + sleep=sleep_mock, |
| 743 | + ), |
| 744 | + ) |
| 745 | + return sleep_mock |
737 | 746 |
|
738 | 747 |
|
739 | | -@pytest.mark.usefixtures("steady_clock") |
740 | 748 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |
741 | 749 | def test_poll_interval_used_by_context_manager( |
742 | | - lock_type: type[BaseFileLock], tmp_path: Path, mocker: MockerFixture |
| 750 | + lock_type: type[BaseFileLock], tmp_path: Path, steady_clock: Mock |
743 | 751 | ) -> None: |
744 | 752 | lock_path = tmp_path / "a" |
745 | 753 | lock_1 = lock_type(str(lock_path)) |
746 | 754 | lock_2 = lock_type(str(lock_path), timeout=0.2, poll_interval=0.05) |
747 | 755 |
|
748 | 756 | lock_1.acquire() |
749 | | - sleep_mock = mocker.patch("filelock._api.time.sleep") |
750 | 757 | with pytest.raises(Timeout): |
751 | 758 | lock_2.acquire() |
752 | | - sleep_mock.assert_any_call(0.05) |
| 759 | + steady_clock.assert_any_call(0.05) |
753 | 760 |
|
754 | | - sleep_mock.reset_mock() |
| 761 | + steady_clock.reset_mock() |
755 | 762 | lock_2.poll_interval = 0.1 |
756 | 763 | with pytest.raises(Timeout): |
757 | 764 | lock_2.acquire() |
758 | | - sleep_mock.assert_any_call(0.1) |
| 765 | + steady_clock.assert_any_call(0.1) |
759 | 766 | lock_1.release() |
760 | 767 |
|
761 | 768 |
|
762 | | -@pytest.mark.usefixtures("steady_clock") |
763 | 769 | @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock]) |
764 | | -def test_poll_interval_acquire_override(lock_type: type[BaseFileLock], tmp_path: Path, mocker: MockerFixture) -> None: |
| 770 | +def test_poll_interval_acquire_override(lock_type: type[BaseFileLock], tmp_path: Path, steady_clock: Mock) -> None: |
765 | 771 | lock_path = tmp_path / "a" |
766 | 772 | lock_1 = lock_type(str(lock_path)) |
767 | 773 | lock_2 = lock_type(str(lock_path), timeout=0.2, poll_interval=0.05) |
768 | 774 |
|
769 | 775 | lock_1.acquire() |
770 | | - sleep_mock = mocker.patch("filelock._api.time.sleep") |
771 | 776 | with pytest.raises(Timeout): |
772 | 777 | lock_2.acquire(poll_interval=0.15) |
773 | | - sleep_mock.assert_any_call(0.15) |
| 778 | + steady_clock.assert_any_call(0.15) |
774 | 779 | lock_1.release() |
775 | 780 |
|
776 | 781 |
|
@@ -1969,7 +1974,10 @@ def cancel_after_two() -> bool: |
1969 | 1974 | call_count += 1 |
1970 | 1975 | return call_count >= 2 |
1971 | 1976 |
|
1972 | | - mocker.patch("filelock._api.time.sleep") |
| 1977 | + mocker.patch( |
| 1978 | + "filelock._api.time", |
| 1979 | + new=SimpleNamespace(perf_counter=time.perf_counter, sleep=mocker.create_autospec(time.sleep)), |
| 1980 | + ) |
1973 | 1981 | with pytest.raises(Timeout): |
1974 | 1982 | lock_2.acquire(timeout=10, cancel_check=cancel_after_two) |
1975 | 1983 | assert call_count == 2 |
@@ -3536,7 +3544,8 @@ def test_lock_descriptor_blocking_retries_until_free( |
3536 | 3544 | assert lock_descriptor(holder, blocking=False) is True |
3537 | 3545 | # Drive the real blocking loop: the first attempt sees contention, the mocked sleep frees the holder, and the |
3538 | 3546 | # second attempt wins. Only the clock is mocked, so a single sleep call proves exactly one retry happened. |
3539 | | - sleep = mocker.patch("filelock._descriptor.time.sleep", side_effect=lambda _: unlock_descriptor(holder)) |
| 3547 | + sleep: Final = mocker.create_autospec(time.sleep, side_effect=lambda _: unlock_descriptor(holder)) |
| 3548 | + mocker.patch("filelock._descriptor.time", new=SimpleNamespace(sleep=sleep)) |
3540 | 3549 | try: |
3541 | 3550 | assert lock_descriptor(fd, blocking=True, poll_interval=poll_interval) is True |
3542 | 3551 | sleep.assert_called_once_with(poll_interval) |
|
0 commit comments