o L)j)@sXdZddlZddlZddlZddlZddlmZGdddeZedkr*e dSdS)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@seZdZddZddZddZddZd d Zd d Zd dZ ddZ ddZ ddZ ddZ ddZddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zd/d0Zd1S)2TestInterpreterShutdowncCs6t|}tjtjd|gddddd}|j|j|jfS)z Run a Python script in a subprocess that exercises greenlet during interpreter shutdown. Returns (returncode, stdout, stderr). z-cTF)capture_outputtexttimeoutcheck) textwrapdedent subprocessrunsys executable returncodestdoutstderr)self script_body full_scriptresultr/var/www/html/finance-dev.cargoinsureonline.com/_shared/backend-venv/lib/python3.10/site-packages/greenlet/tests/test_interpreter_shutdown.py_run_shutdown_scripts  z,TestInterpreterShutdown._run_shutdown_scriptc C>|d\}}}||dd|d|||d|dS)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") rProcess crashed (rc=): z OK: exiting with active greenletr assertEqualassertInrrcrrrrr)test_active_greenlet_at_shutdown_no_crash0s  zATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crashc Cr)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") rrrz#OK: 10 active greenlets at shutdownrrrrr*test_multiple_active_greenlets_at_shutdownHszBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownc Cr)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") rrrz OK: nested greenlets at shutdownrrrrr!test_nested_greenlets_at_shutdown^sz9TestInterpreterShutdown.test_nested_greenlets_at_shutdownc Cr)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") rrrz$OK: 3 threaded greenlets at shutdownrrrrr#test_threaded_greenlets_at_shutdowntsz;TestInterpreterShutdown.test_threaded_greenlets_at_shutdownc CJ|d\}}}||dd|d|||d||d|dS)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") rrrOK: thread cleanup donezCLEANUP: GreenletExit caughtrrrrr(test_greenlet_cleanup_during_thread_exits z@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exitc Cr%)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") rrrr&zFINALLY: cleanup executedrrrrr%test_finally_block_during_thread_exits z=TestInterpreterShutdown.test_finally_block_during_thread_exitc Cr)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. rrrz#OK: 50 greenlets about to shut downrrrrr,test_many_greenlets_with_cleanup_at_shutdownszDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdownc Cr)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}") rrrzOK: nested to depth 10rrrrr(test_deeply_nested_greenlets_at_shutdowns z@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdownc Cr)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") rrrz.OK: greenlet with active exception at shutdownrrrrr(test_greenlet_with_traceback_at_shutdownsz@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdownc CL|d\}}}||dd|d|||d||d|ddS)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") rrrOK: atexit registered&OK: getcurrent returned valid greenletigetcurrent() should return a valid greenlet when called before greenlet's cleanup handler (LIFO ordering)rrrrr&test_getcurrent_during_atexit_no_crash.s  z>TestInterpreterShutdown.test_getcurrent_during_atexit_no_crashc Cr)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") rrrOK: registeredrrrrr$test_gettrace_during_atexit_no_crashLztest_greenlet_construction_with_active_greenlets_during_atexitszVTestInterpreterShutdown.test_greenlet_construction_with_active_greenlets_during_atexitc Cr)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") rrrz.OK: cross-thread setup done, atexit registeredrrrrrCtest_greenlet_construction_with_cross_thread_deleteme_during_atexits#z[TestInterpreterShutdown.test_greenlet_construction_with_cross_thread_deleteme_during_atexitc Cr%)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") rrrzOK: deferred cycle createdGUARDED: getcurrent=Nonerrrrr3test_getcurrent_returns_none_during_gc_finalizations zKTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalizationc Cr%)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") rrrz'OK: 10 active greenlets, cycle deferredr:rrrrrItest_getcurrent_returns_none_during_gc_finalization_with_active_greenlets>s% zaTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_with_active_greenletsc Cr%)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") rrrz)OK: cross-thread cleanup + cycle deferredr:rrrrr@test_getcurrent_returns_none_during_gc_finalization_cross_threadks+ zXTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_cross_threadc Cr%)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") rrrz,OK: atexit registered before greenlet import UNGUARDEDrrrrr0test_getcurrent_returns_none_during_atexit_phases zHTestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phasec Cr%)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") rrrr7r>rrrrrFtest_getcurrent_returns_none_during_atexit_phase_with_active_greenletss z^TestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phase_with_active_greenletsc CsJ|d\}}}||dd|d|||d||d|dS)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() rrrDestructor runningz)RuntimeError: greenlet is being finalizedrrrrr5test_api_getcurrent_no_system_error_at_module_gc_times zMTestInterpreterShutdown.test_api_getcurrent_no_system_error_at_module_gc_timec Cr%)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() rrrrAzResult 1rrrrr&test_switch_no_error_at_module_gc_times z>TestInterpreterShutdown.test_switch_no_error_at_module_gc_timeN)__name__ __module__ __qualname__rr!r"r#r$r'r(r)r*r+r0r2r4r5r6r8r9r;r<r=r?r@rBrCrrrrrs2*$'">(-K!$ r__main__) __doc__r r unittestr greenlet.testsrrrDmainrrrrs