:jzSrSSKrSSKrSSKrSSKrSSKJr "SS\5r\S:Xa\R"5 gg)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\rSrSrSrSrSrSrSrSr Sr S r S r S r S rS rSrSrSrSrSrSrSrSrSrSrSrSrSrg)TestInterpreterShutdownc[R"U5n[R"[R SU/SSSSS9nUR URUR4$)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_scriptresults y/root/GenerationalWealth/GenerationalWealth/venv/lib/python3.13/site-packages/greenlet/tests/test_interpreter_shutdown.py_run_shutdown_script,TestInterpreterShutdown._run_shutdown_scriptsW ook2  ^^T; /    &-->>c URS5upnURUSSUSUU35 URSU5 g)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 assertEqualassertInrrcrrs r)test_active_greenlet_at_shutdown_no_crashATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crash0sS"66 8  F Q"6rd$vhvh OP 8&Arc URS5upnURUSSUSUU35 URSU5 g)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") rrr z#OK: 10 active greenlets at shutdownr!r$s r*test_multiple_active_greenlets_at_shutdownBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownHsS"668  F Q"6rd$vhvh OP ;VDrc URS5upnURUSSUSUU35 URSU5 g)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") rrr z OK: nested greenlets at shutdownr!r$s r!test_nested_greenlets_at_shutdown9TestInterpreterShutdown.test_nested_greenlets_at_shutdown^sS"668  F Q"6rd$vhvh OP 8&Arc URS5upnURUSSUSUU35 URSU5 g)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") rrr z$OK: 3 threaded greenlets at shutdownr!r$s r#test_threaded_greenlets_at_shutdown;TestInterpreterShutdown.test_threaded_greenlets_at_shutdowntsS"668  F4 Q"6rd$vhvh OP  thread-state # cleanup triggers GreenletExit t = threading.Thread(target=thread_func) t.start() t.join() print("OK: thread cleanup done") rrr OK: thread cleanup donezCLEANUP: GreenletExit caughtr!r$s r(test_greenlet_cleanup_during_thread_exit@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exitsb "668  F4 Q"6rd$vhvh OP /8 4f=rc URS5upnURUSSUSUU35 URSU5 URSU5 g)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") rrr r2zFINALLY: cleanup executedr!r$s r%test_finally_block_during_thread_exit=TestInterpreterShutdown.test_finally_block_during_thread_exitsb"668  F, Q"6rd$vhvh OP /8 16:rc URS5upnURUSSUSUU35 URSU5 g)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. rrr z#OK: 50 greenlets about to shut downr!r$s r,test_many_greenlets_with_cleanup_at_shutdownDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdownsS"668  F0 Q"6rd$vhvh OP ;VDrc URS5upnURUSSUSUU35 URSU5 g)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}") rrr zOK: nested to depth 10r!r$s r(test_deeply_nested_greenlets_at_shutdown@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdownsS"66 8  F Q"6rd$vhvh OP .7rc URS5upnURUSSUSUU35 URSU5 g)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") rrr z.OK: greenlet with active exception at shutdownr!r$s r(test_greenlet_with_traceback_at_shutdown@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdownsS"668  F Q"6rd$vhvh OP FOrc URS5upnURUSSUSUU35 URSU5 URSUS5 g)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") rrr OK: atexit registered&OK: getcurrent returned valid greenletigetcurrent() should return a valid greenlet when called before greenlet's cleanup handler (LIFO ordering)r!r$s r&test_getcurrent_during_atexit_no_crash>TestInterpreterShutdown.test_getcurrent_during_atexit_no_crash.sk "668  F& Q"6rd$vhvh OP -v6 >J Krc URS5upnURUSSUSUU35 URSU5 g)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") rrr OK: registeredr!r$s r$test_gettrace_during_atexit_no_crashJ Krc URS5upnURUSSUSUU35 URSU5 URSU5 g)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") rrr rBz!OK: created greenlet successfullyr!r$s r1test_greenlet_construction_during_atexit_no_crashITestInterpreterShutdown.test_greenlet_construction_during_atexit_no_crashsb"668  F& Q"6rd$vhvh OP -v6 96Brc URS5upnURUSSUSUU35 URSU5 g)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") rrr *OK: 10 active greenlets, atexit registeredr!r$s r>test_greenlet_construction_with_active_greenlets_during_atexitVTestInterpreterShutdown.test_greenlet_construction_with_active_greenlets_during_atexitsS "668  F4 Q"6rd$vhvh OP BFKrc URS5upnURUSSUSUU35 URSU5 g)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") rrr z.OK: cross-thread setup done, atexit registeredr!r$s rCtest_greenlet_construction_with_cross_thread_deleteme_during_atexit[TestInterpreterShutdown.test_greenlet_construction_with_cross_thread_deleteme_during_atexitsT"66"8 " FF Q"6rd$vhvh OP FOrc URS5upnURUSSUSUU35 URSU5 URSUS5 g)Nax import gc import os import greenlet gc.disable() class CleanupChecker: def __del__(self): try: 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") # 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") rrr zOK: deferred cycle createdGUARDED: getcurrent=Nonetgetcurrent() must return None during GC finalization; returned a live object instead (missing Py_IsFinalizing guard)r!r$s r3test_getcurrent_returns_none_during_gc_finalizationKTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalizationsk"668  F6 Q"6rd$vhvh OP 2F; 0&W Xrc URS5upnURUSSUSUU35 URSU5 URSUS5 g)Na import gc import os import greenlet gc.disable() class CleanupChecker: def __del__(self): try: 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") # 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") rrr z'OK: 10 active greenlets, cycle deferredr]r^r!r$s rItest_getcurrent_returns_none_during_gc_finalization_with_active_greenletsaTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_with_active_greenlets?sl"66#8 # FH Q"6rd$vhvh OP ?H 0&W Xrc URS5upnURUSSUSUU35 URSU5 URSUS5 g)Na{ import gc import os import threading import greenlet gc.disable() class CleanupChecker: def __del__(self): try: 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") # 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") rrr z)OK: cross-thread cleanup + cycle deferredr]r^r!r$s r@test_getcurrent_returns_none_during_gc_finalization_cross_threadXTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_cross_threadmsl"66)8 ) FT Q"6rd$vhvh OP A6J 0&W Xrc URS5upnURUSSUSUU35 URSU5 URSU5 g)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") rrr z,OK: atexit registered before greenlet import UNGUARDEDr!r$s r0test_getcurrent_returns_none_during_atexit_phaseHTestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phasesa"668  F0 Q"6rd$vhvh OP DfM k6*rc URS5upnURUSSUSUU35 URSU5 URSU5 g)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") rrr rVrhr!r$s rFtest_getcurrent_returns_none_during_atexit_phase_with_active_greenlets^TestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phase_with_active_greenletss_!668  F< Q"6rd$vhvh OP BFK k6*rc URS5upnURUSSUSUU35 URSU5 URSU5 g)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() rrr Destructor runningz)RuntimeError: greenlet is being finalizedr!r$s r5test_api_getcurrent_no_system_error_at_module_gc_timeMTestInterpreterShutdown.test_api_getcurrent_no_system_error_at_module_gc_timesb"66 8  F Q"6rd$vhvh OP *F3 A6Jrc URS5upnURUSSUSUU35 URSU5 URSU5 g)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() rrr rozResult 1r!r$s r&test_switch_no_error_at_module_gc_time>TestInterpreterShutdown.test_switch_no_error_at_module_gc_timesa"668  F0 Q"6rd$vhvh OP *F3 j&)rN)__name__ __module__ __qualname____firstlineno__rr&r)r,r/r3r6r9r<r?rErIrMrPrSrWrZr_rbrerirlrprs__static_attributes__rurrrrs ?&B0E,B,FT">H;:E>8(P8K<0(0(%KNC6 LD+P|'XR,X\5XX+B"+HK2*rr__main__) __doc__rrunittestr greenlet.testsrrrvmainrurrrsB" #Y *hY *x z MMOr