📁 File Manager Pro
v10.0.3 | PHP: 8.2.31
Server: LiteSpeed
2026-06-28 06:55:35
📂
/ (Root)
/
opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
greenlet
/
tests
/
__pycache__
📍 /opt/hc_python/lib/python3.12/site-packages/greenlet/tests/__pycache__
🔄 Refresh
✏️
Editing: test_interpreter_shutdown.cpython-312.pyc
Read Only
� }�*j)� � �x � d Z ddlZddlZddlZddlZddlmZ G d� de� Zedk( r ej � yy)u: Tests for greenlet behavior during interpreter shutdown (Py_FinalizeEx). These tests are organized into four groups: A. Core safety (smoke): no crashes with active greenlets at shutdown. B. Cleanup semantics: GreenletExit / finally still works during normal thread exit (the standard production path). C. Atexit "still works" tests: getcurrent() / greenlet construction during atexit handlers registered AFTER greenlet import (i.e. BEFORE greenlet's cleanup handler in LIFO order) must return valid objects — verifies the guards don't over-block. D. TDD-certified regression tests: getcurrent() must return None when called AFTER greenlet's cleanup (GC finalization phase or late atexit phase). These tests fail on greenlet 3.3.2 and pass with the fix across Python 3.10-3.14. � N)�TestCasec � � e Zd Zd� Zd� Zd� Zd� Zd� Zd� Zd� Z d� Z d � Z d � Z d � Z d � Zd � Zd� Zd� Zd� Zd� Zd� Zd� Zd� Zd� Zd� Zd� Zd� Zy)�TestInterpreterShutdownc �� � t j |� }t j t j d|gdddd�� }|j |j |j fS )z� Run a Python script in a subprocess that exercises greenlet during interpreter shutdown. Returns (returncode, stdout, stderr). z-cT� F)�capture_output�text�timeout�check) �textwrap�dedent� subprocess�run�sys� executable� returncode�stdout�stderr)�self� script_body� full_script�results �Y/opt/hc_python/lib64/python3.12/site-packages/greenlet/tests/test_interpreter_shutdown.py�_run_shutdown_scriptz,TestInterpreterShutdown._run_shutdown_script sW � � �o�o�k�2� �� � � �^�^�T�;� /� ���� �� � � �&�-�-�� � �>�>� c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )NaT import greenlet def worker(): greenlet.getcurrent().parent.switch("from worker") return "done" g = greenlet.greenlet(worker) result = g.switch() assert result == "from worker", result print("OK: exiting with active greenlet") r �Process crashed (rc=�): z OK: exiting with active greenlet�r � assertEqual�assertIn�r �rcr r s r �)test_active_greenlet_at_shutdown_no_crashzATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crash0 sU � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �8�&�Ar c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import greenlet def worker(name): greenlet.getcurrent().parent.switch(f"hello from {name}") return "done" greenlets = [] for i in range(10): g = greenlet.greenlet(worker) result = g.switch(f"g{i}") greenlets.append(g) print(f"OK: {len(greenlets)} active greenlets at shutdown") r r r z#OK: 10 active greenlets at shutdownr r" s r �*test_multiple_active_greenlets_at_shutdownzBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownH sU � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �;�V�Dr c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import greenlet def inner(): greenlet.getcurrent().parent.switch("inner done") def outer(): g_inner = greenlet.greenlet(inner) g_inner.switch() greenlet.getcurrent().parent.switch("outer done") g = greenlet.greenlet(outer) result = g.switch() assert result == "outer done", result print("OK: nested greenlets at shutdown") r r r z OK: nested greenlets at shutdownr r" s r �!test_nested_greenlets_at_shutdownz9TestInterpreterShutdown.test_nested_greenlets_at_shutdown^ sU � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �8�&�Ar c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import greenlet import threading results = [] def thread_worker(): def greenlet_func(): greenlet.getcurrent().parent.switch("from thread greenlet") return "done" g = greenlet.greenlet(greenlet_func) val = g.switch() results.append((g, val)) threads = [] for _ in range(3): t = threading.Thread(target=thread_worker) t.start() threads.append(t) for t in threads: t.join() print(f"OK: {len(results)} threaded greenlets at shutdown") r r r z$OK: 3 threaded greenlets at shutdownr r" s r �#test_threaded_greenlets_at_shutdownz;TestInterpreterShutdown.test_threaded_greenlets_at_shutdownt sU � � "� 6� 6� 8 � ���F�F�4 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �<�f�Er c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na4 import os import threading import greenlet _write = os.write def thread_func(): def worker(_w=_write, _GreenletExit=greenlet.GreenletExit): try: greenlet.getcurrent().parent.switch("suspended") except _GreenletExit: _w(1, b"CLEANUP: GreenletExit caught\n") raise g = greenlet.greenlet(worker) g.switch() # Thread exits with active greenlet -> thread-state # cleanup triggers GreenletExit t = threading.Thread(target=thread_func) t.start() t.join() print("OK: thread cleanup done") r r r �OK: thread cleanup donez CLEANUP: GreenletExit caughtr r" s r �(test_greenlet_cleanup_during_thread_exitz@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exit� sd � � "� 6� 6� 8 � ���F�F�4 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �/��8� � � �4�f�=r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )NaR import os import threading import greenlet _write = os.write def thread_func(): def worker(_w=_write): try: greenlet.getcurrent().parent.switch("suspended") finally: _w(1, b"FINALLY: cleanup executed\n") g = greenlet.greenlet(worker) g.switch() t = threading.Thread(target=thread_func) t.start() t.join() print("OK: thread cleanup done") r r r r, zFINALLY: cleanup executedr r" s r �%test_finally_block_during_thread_exitz=TestInterpreterShutdown.test_finally_block_during_thread_exit� sd � � "� 6� 6� 8 � ���F�F�, ����Q�"6�r�d�$�v�h�v�h� O�P� � � �/��8� � � �1�6�:r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import sys import greenlet cleanup_count = 0 def worker(idx): global cleanup_count try: greenlet.getcurrent().parent.switch(f"ready-{idx}") except greenlet.GreenletExit: cleanup_count += 1 raise greenlets = [] for i in range(50): g = greenlet.greenlet(worker) result = g.switch(i) greenlets.append(g) print(f"OK: {len(greenlets)} greenlets about to shut down") # Note: we can't easily print cleanup_count during shutdown # since it happens after the main module's code runs. r r r z#OK: 50 greenlets about to shut downr r" s r �,test_many_greenlets_with_cleanup_at_shutdownzDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdown� sU � � "� 6� 6� 8 � ���F�F�0 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �;�V�Dr c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import greenlet def level(depth, max_depth): if depth < max_depth: g = greenlet.greenlet(level) g.switch(depth + 1, max_depth) greenlet.getcurrent().parent.switch(f"depth-{depth}") g = greenlet.greenlet(level) result = g.switch(0, 10) print(f"OK: nested to depth 10, got {result}") r r r zOK: nested to depth 10r r" s r �(test_deeply_nested_greenlets_at_shutdownz@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdown� sU � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �.��7r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na import greenlet def worker(): try: raise ValueError("test error") except ValueError: # Suspend while an exception is active on the stack greenlet.getcurrent().parent.switch("suspended with exc") return "done" g = greenlet.greenlet(worker) result = g.switch() assert result == "suspended with exc" print("OK: greenlet with active exception at shutdown") r r r z.OK: greenlet with active exception at shutdownr r" s r �(test_greenlet_with_traceback_at_shutdownz@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdown sU � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �F��Or c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|d� y )Na� import atexit import greenlet def call_getcurrent_at_exit(): try: g = greenlet.getcurrent() if g is not None and type(g).__name__ == 'greenlet': print(f"OK: getcurrent returned valid greenlet") elif g is None: print("FAIL: getcurrent returned None (over-blocked)") else: print(f"FAIL: unexpected {g!r}") except Exception as e: print(f"OK: getcurrent raised {type(e).__name__}: {e}") atexit.register(call_getcurrent_at_exit) print("OK: atexit registered") r r r �OK: atexit registered�&OK: getcurrent returned valid greenlet�igetcurrent() should return a valid greenlet when called before greenlet's cleanup handler (LIFO ordering)r r" s r �&test_getcurrent_during_atexit_no_crashz>TestInterpreterShutdown.test_getcurrent_during_atexit_no_crash. sm � � "� 6� 6� 8 � ���F�F�& ����Q�"6�r�d�$�v�h�v�h� O�P� � � �-�v�6� � � �>��J� Kr c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import atexit import greenlet def check_at_exit(): try: result = greenlet.gettrace() print(f"OK: gettrace returned {result!r}") except Exception as e: print(f"OK: gettrace raised {type(e).__name__}: {e}") atexit.register(check_at_exit) print("OK: registered") r r r �OK: registeredr r" s r �$test_gettrace_during_atexit_no_crashz<TestInterpreterShutdown.test_gettrace_during_atexit_no_crashL �U � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �&��/r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import atexit import greenlet def check_at_exit(): try: greenlet.settrace(lambda *args: None) print("OK: settrace succeeded") except Exception as e: print(f"OK: settrace raised {type(e).__name__}: {e}") atexit.register(check_at_exit) print("OK: registered") r r r r< r r" s r �$test_settrace_during_atexit_no_crashz<TestInterpreterShutdown.test_settrace_during_atexit_no_crash` r> r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|d� y )Na� import atexit import greenlet def worker(): greenlet.getcurrent().parent.switch("ready") greenlets = [] for i in range(5): g = greenlet.greenlet(worker) result = g.switch() greenlets.append(g) def check_at_exit(): try: g = greenlet.getcurrent() if g is not None and type(g).__name__ == 'greenlet': print(f"OK: getcurrent returned valid greenlet") elif g is None: print("FAIL: getcurrent returned None (over-blocked)") else: print(f"FAIL: unexpected {g!r}") except Exception as e: print(f"OK: getcurrent raised {type(e).__name__}: {e}") atexit.register(check_at_exit) print(f"OK: {len(greenlets)} active greenlets, atexit registered") r r r z)OK: 5 active greenlets, atexit registeredr8 r9 r r" s r �3test_getcurrent_with_active_greenlets_during_atexitzKTestInterpreterShutdown.test_getcurrent_with_active_greenlets_during_atexitt sm � � "� 6� 6� 8 � ���F�F�8 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �A�6�J� � � �>��J� Kr c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import atexit import greenlet def create_greenlets_at_exit(): try: def noop(): pass g = greenlet.greenlet(noop) if g is not None: print(f"OK: created greenlet successfully") else: print("FAIL: greenlet() returned None") except Exception as e: print(f"OK: construction raised {type(e).__name__}: {e}") atexit.register(create_greenlets_at_exit) print("OK: atexit registered") r r r r7 z!OK: created greenlet successfullyr r" s r �1test_greenlet_construction_during_atexit_no_crashzITestInterpreterShutdown.test_greenlet_construction_during_atexit_no_crash� sd � � "� 6� 6� 8 � ���F�F�& ����Q�"6�r�d�$�v�h�v�h� O�P� � � �-�v�6� � � �9�6�Br c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Naf import atexit import greenlet def worker(): greenlet.getcurrent().parent.switch("ready") greenlets = [] for i in range(10): g = greenlet.greenlet(worker) g.switch() greenlets.append(g) def create_at_exit(): try: new_greenlets = [] for i in range(5): g = greenlet.greenlet(lambda: None) new_greenlets.append(g) print(f"OK: created {len(new_greenlets)} greenlets at exit") except Exception as e: print(f"OK: raised {type(e).__name__}: {e}") atexit.register(create_at_exit) print(f"OK: {len(greenlets)} active greenlets, atexit registered") r r r �*OK: 10 active greenlets, atexit registeredr r" s r �>test_greenlet_construction_with_active_greenlets_during_atexitzVTestInterpreterShutdown.test_greenlet_construction_with_active_greenlets_during_atexit� sU � � "� 6� 6� 8 � ���F�F�4 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �B�F�Kr c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� y )Na� import atexit import greenlet import threading cross_thread_refs = [] def thread_worker(): # Create greenlets in this thread def gl_body(): greenlet.getcurrent().parent.switch("ready") for _ in range(20): g = greenlet.greenlet(gl_body) g.switch() cross_thread_refs.append(g) t = threading.Thread(target=thread_worker) t.start() t.join() # Dropping these references in the main thread # causes them to be added to the main thread's # deleteme list (deferred cross-thread dealloc). cross_thread_refs.clear() def create_at_exit(): try: g = greenlet.greenlet(lambda: None) print(f"OK: created greenlet at exit {g!r}") except Exception as e: print(f"OK: raised {type(e).__name__}: {e}") atexit.register(create_at_exit) print("OK: cross-thread setup done, atexit registered") r r r z.OK: cross-thread setup done, atexit registeredr r" s r �Ctest_greenlet_construction_with_cross_thread_deleteme_during_atexitz[TestInterpreterShutdown.test_greenlet_construction_with_cross_thread_deleteme_during_atexit� sV � � "� 6� 6� "8 � " ���F�F�F ����Q�"6�r�d�$�v�h�v�h� O�P� � � �F��Or c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import gc import os import greenlet gc.disable() class CleanupChecker: def __del__(self): try: try: greenlet.getcurrent() except RuntimeError: os.write(1, b"GUARDED: getcurrent=None\n") else: os.write(1, b"UNGUARDED: getcurrent=" + type(cur).__name__.encode() + b"\n") except Exception as e: os.write(1, b"EXCEPTION: " + str(e).encode() + b"\n") # Reference cycle: only collected during Py_FinalizeEx GC pass a = CleanupChecker() b = {"ref": a} a._cycle = b del a, b print("OK: deferred cycle created") r r r zOK: deferred cycle created�GUARDED: getcurrent=Noner r" s r �3test_getcurrent_returns_none_during_gc_finalizationzKTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization sd � � "� 6� 6� 8 � ���F�F�8 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �2�F�;� � � �0�&�9r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import gc import os import greenlet gc.disable() class CleanupChecker: def __del__(self): try: try: greenlet.getcurrent() except RuntimeError: os.write(1, b"GUARDED: getcurrent=None\n") else: os.write(1, b"UNGUARDED: getcurrent=" + type(cur).__name__.encode() + b"\n") except Exception as e: os.write(1, b"EXCEPTION: " + str(e).encode() + b"\n") # Create active greenlets greenlets = [] for _ in range(10): def worker(): greenlet.getcurrent().parent.switch("suspended") g = greenlet.greenlet(worker) g.switch() greenlets.append(g) # Reference cycle deferred to Py_FinalizeEx a = CleanupChecker() b = {"ref": a} a._cycle = b del a, b print(f"OK: {len(greenlets)} active greenlets, cycle deferred") r r r z'OK: 10 active greenlets, cycle deferredrK r r" s r �Itest_getcurrent_returns_none_during_gc_finalization_with_active_greenletszaTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_with_active_greenlets> se � � "� 6� 6� $8 � $ ���F�F�J ����Q�"6�r�d�$�v�h�v�h� O�P� � � �?��H� � � �0�&�9r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import gc import os import threading import greenlet gc.disable() class CleanupChecker: def __del__(self): try: try: greenlet.getcurrent() except RuntimeError: os.write(1, b"GUARDED: getcurrent=None\n") else: os.write(1, b"UNGUARDED: getcurrent=" + type(cur).__name__.encode() + b"\n") except Exception as e: os.write(1, b"EXCEPTION: " + str(e).encode() + b"\n") # Create cross-thread greenlet references cross_refs = [] def thread_fn(): for _ in range(20): def body(): greenlet.getcurrent().parent.switch("x") g = greenlet.greenlet(body) g.switch() cross_refs.append(g) t = threading.Thread(target=thread_fn) t.start() t.join() cross_refs.clear() # Reference cycle deferred to Py_FinalizeEx a = CleanupChecker() b = {"ref": a} a._cycle = b del a, b print("OK: cross-thread cleanup + cycle deferred") r r r z)OK: cross-thread cleanup + cycle deferredrK r r" s r �@test_getcurrent_returns_none_during_gc_finalization_cross_threadzXTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_cross_threadk se � � "� 6� 6� *8 � * ���F�F�V ����Q�"6�r�d�$�v�h�v�h� O�P� � � �A�6�J� � � �0�&�9r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Nu� import atexit import os def late_checker(): try: import greenlet cur = greenlet.getcurrent() if cur is None: os.write(1, b"GUARDED: getcurrent=None\n") else: os.write(1, b"UNGUARDED: getcurrent=" + type(cur).__name__.encode() + b"\n") except Exception as e: os.write(1, b"EXCEPTION: " + str(e).encode() + b"\n") # Register BEFORE importing greenlet. LIFO order: # greenlet's handler (registered at import) runs FIRST, # late_checker runs SECOND — seeing the flag already set. atexit.register(late_checker) import greenlet print("OK: atexit registered before greenlet import") r r r z,OK: atexit registered before greenlet import� UNGUARDEDr r" s r �0test_getcurrent_returns_none_during_atexit_phasezHTestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phase� sc � � "� 6� 6� 8 � ���F�F�0 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �D�f�M� � � �k�6�*r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import atexit import os def late_checker(): try: import greenlet cur = greenlet.getcurrent() if cur is None: os.write(1, b"GUARDED: getcurrent=None\n") else: os.write(1, b"UNGUARDED: getcurrent=" + type(cur).__name__.encode() + b"\n") except Exception as e: os.write(1, b"EXCEPTION: " + str(e).encode() + b"\n") atexit.register(late_checker) import greenlet greenlets = [] for _ in range(10): def worker(): greenlet.getcurrent().parent.switch("parked") g = greenlet.greenlet(worker) g.switch() greenlets.append(g) print(f"OK: {len(greenlets)} active greenlets, atexit registered") r r r rF rR r r" s r �Ftest_getcurrent_returns_none_during_atexit_phase_with_active_greenletsz^TestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phase_with_active_greenlets� sa � � !� 6� 6� 8 � ���F�F�< ����Q�"6�r�d�$�v�h�v�h� O�P� � � �B�F�K� � � �k�6�*r c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import greenlet from greenlet.tests import _test_extension class WithDel: # must cache the method we want, because by the time we # run, module globals may have been cleaned up. def __del__(self, gc=_test_extension.getcurrent_api): print('Destructor running') gc() # Should print an unraisable RuntimeException greenlet._greenlet.with_del = WithDel() r r r �Destructor runningz)RuntimeError: greenlet is being finalizedr r" s r �5test_api_getcurrent_no_system_error_at_module_gc_timezMTestInterpreterShutdown.test_api_getcurrent_no_system_error_at_module_gc_time� sd � � "� 6� 6� 8 � ���F�F� ����Q�"6�r�d�$�v�h�v�h� O�P� � � �*�F�3� � � �A�6�Jr c � � | j d� \ }}}| j |dd|� d|� |� �� | j d|� | j d|� y )Na� import greenlet from greenlet.tests import _test_extension gs = [] # must cache the objects we want, because by the time we # run, module globals may have been cleaned up. def do_it(gs=gs): print('current', gs) gs[0].parent.switch(1) gs.append(greenlet.greenlet(do_it)) gs.append(greenlet.greenlet(do_it)) gs[1].switch() class WithDel: def __del__(self, gs=gs): print('Destructor running') r = gs[0].switch() print('Result', r) greenlet._greenlet.with_del = WithDel() r r r rW zResult 1r r" s r �&test_switch_no_error_at_module_gc_timez>TestInterpreterShutdown.test_switch_no_error_at_module_gc_time sc � � "� 6� 6� 8 � ���F�F�0 ����Q�"6�r�d�$�v�h�v�h� O�P� � � �*�F�3� � � �j�&�)r N)�__name__� __module__� __qualname__r r$ r& r( r* r- r/ r1 r3 r5 r: r= r@ rB rD rG rI rL rN rP rS rU rX rZ � r r r r s� � � ?�&B�0E�,B�, F�T">�H;�: E�>8�(P�8 K�<0�(0�(%K�NC�6 L�D+P�|&:�P+:�Z4:�V +�B"+�HK�2 *r r �__main__) �__doc__r r �unittestr �greenlet.testsr r r[ �mainr^ r r �<module>rd sD ���" � � � � #�V *�h� V *�r �z���H�M�M�O� r
💾 Save Changes
❌ Cancel