()
| 814 | |
| 815 | |
| 816 | def _idle_fork_script() -> str: |
| 817 | return textwrap.dedent( |
| 818 | """ |
| 819 | from __future__ import annotations |
| 820 | |
| 821 | import os |
| 822 | import sys |
| 823 | |
| 824 | from filelock import ReadWriteLock |
| 825 | |
| 826 | lock_path = sys.argv[1] |
| 827 | inherited_lock = ReadWriteLock(lock_path, is_singleton=False) |
| 828 | child_pid = os.fork() |
| 829 | if child_pid == 0: |
| 830 | try: |
| 831 | inherited_lock.acquire_read() |
| 832 | except RuntimeError as error: |
| 833 | assert "was invalidated by fork()" in str(error) |
| 834 | else: |
| 835 | raise AssertionError("inherited idle lock acquired") |
| 836 | if sys.implementation.name == "pypy": |
| 837 | try: |
| 838 | ReadWriteLock(lock_path, is_singleton=False) |
| 839 | except RuntimeError as error: |
| 840 | assert str(error) == ( |
| 841 | "ReadWriteLock is unavailable in a PyPy fork child; exec or exit before using it" |
| 842 | ) |
| 843 | else: |
| 844 | raise AssertionError("PyPy child opened SQLite after fork") |
| 845 | else: |
| 846 | with ReadWriteLock(lock_path, is_singleton=False).write_lock(): |
| 847 | pass |
| 848 | sys.exit(0) |
| 849 | _, status = os.waitpid(child_pid, 0) |
| 850 | assert os.waitstatus_to_exitcode(status) == 0 |
| 851 | """ |
| 852 | ) |
| 853 | |
| 854 | |
| 855 | def _pypy_fork_script() -> str: # pragma: pypy cover |
no outgoing calls
no test coverage detected