Function _concurrent_operation_script

tests/test_read_write_fork.py:505–574

()

def _concurrent_operation_script() -> str:
    return textwrap.dedent(
        """
        from __future__ import annotations

        import queue
        import sqlite3
        import sys
        import threading

        from filelock import ReadWriteLock

        lock_path, operation = sys.argv[1:]
        lock = ReadWriteLock(lock_path)
        acquisition_inside_connect = threading.Event()
        continue_acquisition = threading.Event()
        acquire_errors: queue.SimpleQueue[BaseException] = queue.SimpleQueue()
        operation_errors: queue.SimpleQueue[BaseException] = queue.SimpleQueue()
        armed = True

        from typing import TYPE_CHECKING

        if TYPE_CHECKING:
            from _typeshed import Unused

        def audit_hook(event: str, _args: Unused) -> None:
            if armed and event == "sqlite3.connect":
                acquisition_inside_connect.set()
                assert continue_acquisition.wait(5)

        def acquire() -> None:
            try:
                lock.acquire_write()
            except BaseException as error:
                acquire_errors.put(error)

        def operate() -> None:
            try:
                getattr(lock, operation)()
            except BaseException as error:
                operation_errors.put(error)

        sys.addaudithook(audit_hook)
        acquire_thread = threading.Thread(target=acquire)
        acquire_thread.start()
        assert acquisition_inside_connect.wait(5)
        operation_thread = threading.Thread(target=operate)
        operation_thread.start()
        operation_thread.join(0.1)
        assert operation_thread.is_alive()
        armed = False
        continue_acquisition.set()
        acquire_thread.join(5)
        operation_thread.join(5)
        assert not acquire_thread.is_alive()
        assert not operation_thread.is_alive()
        assert acquire_errors.empty()
        assert operation_errors.empty()