Function assert_read_write_lock_state

tests/read_write_helpers.py:12–25

(lock_file: str, mode: Literal["read", "write"], *, available: bool)

def assert_read_write_lock_state(lock_file: str, mode: Literal["read", "write"], *, available: bool) -> None:
    context = multiprocessing.get_context("spawn")
    acquired = context.Value("b", False)
    probe = context.Process(target=_probe_read_write_lock, args=(lock_file, mode, acquired))
    probe.start()
    try:
        probe.join(timeout=5)
        assert not probe.is_alive(), "read-write lock probe did not exit"
        assert (probe.exitcode, acquired.value) == (0, available)
    finally:
        if probe.is_alive():  # pragma: no cover - cleanup for a hung child after the assertion fails
            probe.terminate()
            probe.join(timeout=5)
        probe.close()


def _probe_read_write_lock(lock_file: str, mode: Literal["read", "write"], acquired: Synchronized[bool]) -> None: