Function acquire_write_in_thread

tests/test_read_write_unit.py:676–681

()

        real_acquire_tx(blocking=blocking, timeout=timeout)

    def acquire_write_in_thread() -> None:
        writer_tid["id"] = threading.get_ident()
        try:
            lock.acquire_write(timeout=5)
        except RuntimeError as exc:
            errors["writer"] = exc

    lock._acquire_transaction_lock = gated  # ty: ignore[invalid-assignment]
    thread = threading.Thread(target=acquire_write_in_thread)