Increase test coverage (#434)

Author: paultiq

tests/test_async_filelock.py

@@ -179,3 +179,92 @@ async def _release(self) -> None:  # type: ignore[override]
     await lock.release()
     assert acquired
     assert released
+
+
+@pytest.mark.parametrize("lock_type", [AsyncFileLock, AsyncSoftFileLock])
+@pytest.mark.asyncio
+async def test_wait_message_logged(
+    lock_type: type[BaseAsyncFileLock], tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+    lock_path = tmp_path / "a"
+    first_lock = lock_type(str(lock_path))
+    second_lock = lock_type(str(lock_path), timeout=0.2)
+
+    # Hold the lock so second_lock has to wait
+    await first_lock.acquire()
+    with pytest.raises(Timeout):
+        await second_lock.acquire()
+    assert any("waiting" in msg for msg in caplog.messages)
+
+
+@pytest.mark.parametrize("lock_type", [AsyncSoftFileLock, AsyncFileLock])
+@pytest.mark.asyncio
+async def test_attempting_to_acquire_branch(
+    lock_type: type[BaseAsyncFileLock], tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+
+    lock = lock_type(str(tmp_path / "a"))
+    await lock.acquire()
+    assert any("Attempting to acquire lock" in m for m in caplog.messages)
+    await lock.release()
+
+
+@pytest.mark.asyncio
+async def test_thread_local_run_in_executor(tmp_path: Path) -> None:  # noqa: RUF029
+    with pytest.raises(ValueError, match="run_in_executor is not supported when thread_local is True"):
+        AsyncSoftFileLock(str(tmp_path / "a"), thread_local=True, run_in_executor=True)
+
+
+@pytest.mark.parametrize("lock_type", [AsyncSoftFileLock, AsyncFileLock])
+@pytest.mark.asyncio
+async def test_attempting_to_acquire(
+    lock_type: type[BaseAsyncFileLock], tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+    lock = lock_type(str(tmp_path / "a.lock"), run_in_executor=False)
+    await lock.acquire(timeout=0.1)
+    assert any("Attempting to acquire lock" in m for m in caplog.messages)
+    await lock.release()
+
+
+@pytest.mark.parametrize("lock_type", [AsyncSoftFileLock, AsyncFileLock])
+@pytest.mark.asyncio
+async def test_attempting_to_release(
+    lock_type: type[BaseAsyncFileLock], tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+    lock = lock_type(str(tmp_path / "a.lock"), run_in_executor=False)
+
+    await lock.acquire(timeout=0.1)  # lock_counter = 1, is_locked = True
+    await lock.acquire(timeout=0.1)  # lock_counter = 2 (reentrant)
+    await lock.release(force=True)
+
+    assert any("Attempting to release lock" in m for m in caplog.messages)
+    assert any("released" in m for m in caplog.messages)
+
+
+@pytest.mark.parametrize("lock_type", [AsyncFileLock, AsyncSoftFileLock])
+@pytest.mark.asyncio
+async def test_release_early_exit_when_unlocked(lock_type: type[BaseAsyncFileLock], tmp_path: Path) -> None:
+    lock = lock_type(str(tmp_path / "a.lock"), run_in_executor=False)
+    assert not lock.is_locked
+    await lock.release()
+    assert not lock.is_locked
+
+
+@pytest.mark.parametrize("lock_type", [AsyncFileLock, AsyncSoftFileLock])
+@pytest.mark.asyncio
+async def test_release_nonzero_counter_exit(
+    lock_type: type[BaseAsyncFileLock], tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+    lock = lock_type(str(tmp_path / "a.lock"), run_in_executor=False)
+    await lock.acquire()
+    await lock.acquire()
+    await lock.release()  # counter goes 2→1
+    assert lock.lock_counter == 1
+    assert lock.is_locked
+    assert not any("Attempting to release" in m for m in caplog.messages)
+    await lock.release()

tests/test_filelock.py

@@ -691,7 +691,8 @@ def __init__(
             my_param: int = 0,
             **kwargs: dict[str, Any],  # noqa: ARG002
         ) -> None:
-            super().__init__(lock_file, timeout, mode, thread_local, blocking=True, is_singleton=True)
+            super().__init__(lock_file, timeout, mode, thread_local, is_singleton=True)
+            self.blocking = True
             self.my_param = my_param
 
     lock_path = tmp_path / "a"
@@ -814,3 +815,24 @@ def __init__(self, file_path: str) -> None:
     lock_path = tmp_path / "a"
     lock = FilePathLock(str(lock_path))
     assert lock.lock_file == str(lock_path) + ".lock"
+
+
+@pytest.mark.parametrize(
+    ("lock_type", "expected_exc"),
+    [
+        (SoftFileLock, TimeoutError),
+        (FileLock, TimeoutError) if sys.platform == "win32" else (FileLock, PermissionError),
+    ],
+)
+def test_mtime_zero_exit_branch(
+    lock_type: type[BaseFileLock], expected_exc: type[BaseException], tmp_path: Path
+) -> None:
+    p = tmp_path / "z.lock"
+    p.touch()
+    Path(p).chmod(0o444)
+    os.utime(p, (0, 0))
+
+    lock = lock_type(str(p))
+
+    with pytest.raises(expected_exc):
+        lock.acquire(timeout=0)