file_lock.py

import fcntl
import time
from contextlib import contextmanager


@contextmanager
def file_lock(lock_path, timeout=5):
    with open(lock_path, "w") as f:
        start = time.monotonic()

        while True:
            try:
                fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if time.monotonic() - start > timeout:
                    raise TimeoutError
                time.sleep(0.1)

        yield
        
        

test.py

import time
import multiprocessing as mp
import pytest
from file_lock import file_lock


def worker(lock_path, q):
    with file_lock(lock_path):
        start = time.time()
        time.sleep(0.5)
        end = time.time()
        q.put((start, end))


def test_lock_serializes(tmp_path):
    lock = tmp_path / "lock"
    q = mp.Queue()

    # start two processes competing for the same lock
    p1 = mp.Process(target=worker, args=(lock, q))
    p2 = mp.Process(target=worker, args=(lock, q))

    p1.start()
    time.sleep(0.05)  # ensure overlap
    p2.start()

    p1.join()
    p2.join()

    r1, r2 = q.get(), q.get()

    # verify critical sections did not overlap
    assert r1[1] <= r2[0] or r2[1] <= r1[0]


def test_timeout(tmp_path):
    lock = tmp_path / "lock"

    def holder():
        with file_lock(lock):
            time.sleep(1)

    p = mp.Process(target=holder)
    p.start()
    time.sleep(0.1)

    # acquiring lock should fail due to timeout
    with pytest.raises(TimeoutError):
        with file_lock(lock, timeout=0.1):
            pass

    p.join()