+ qi/Rt^RIt^RIt^RIt^RIt^RIHt!RR]4t]R8Xd]P!4R#R#)a Tests for greenlet behavior during interpreter shutdown (Py_FinalizeEx). Prior to the safe finalization fix, active greenlets being deallocated during interpreter shutdown could trigger SIGSEGV or SIGABRT on Python < 3.11, because green_dealloc attempted to throw GreenletExit via g_switch() into a partially-torn-down interpreter. The fix adds _Py_IsFinalizing() guards (on Python < 3.11 only) that call murder_in_place() instead of g_switch() when the interpreter is shutting down, avoiding the crash at the cost of not running cleanup code inside the greenlet. These tests verify: 1. No crashes on ANY Python version (the core safety guarantee). 2. GreenletExit cleanup code runs correctly during normal thread exit (the standard production path, e.g. uWSGI worker threads). N)TestCasec\a]tRt^toRtRtRtRtRtRt Rt Rt R t R t R tVtR #) TestInterpreterShutdownc \P!V4p\P!\P RV.RR^RR7pVP VPVP3#)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&& ڑC:\Users\petid\OneDrive\Desktop\Maestro\MAESTRO CLAUDE\MAESTRO\maestro-backend\venv\Lib\site-packages\greenlet/tests/test_interpreter_shutdown.py_run_shutdown_script,TestInterpreterShutdown._run_shutdown_scriptsW ook2  ^^T; /    &-->>c  VPR4wrpVPV^RV RV V 24VPRV4R#)a An active (suspended) greenlet that is deallocated during interpreter shutdown should not crash the process. Before the fix, this would SIGSEGV on Python < 3.11 because _green_dealloc_kill_started_non_main_greenlet tried to call g_switch() during Py_FinalizeEx. aT 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") Process crashed (rc=): z OK: exiting with active greenletNr assertEqualassertInrrcrrs& r)test_active_greenlet_at_shutdown_no_crashATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crash1sS"66 8  F Q"6rd$vhvh OP 8&Arc  VPR4wrpVPV^RV RV V 24VPRV4R#)zU Multiple suspended greenlets at shutdown should all be cleaned up without crashing. a 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") rrz#OK: 10 active greenlets at shutdownNrr!s& r*test_multiple_active_greenlets_at_shutdownBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownIsS "668  F Q"6rd$vhvh OP ;VDrc  VPR4wrpVPV^RV RV V 24VPRV4R#)zA Nested (chained parent) greenlets at shutdown should not crash. a 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") rrz OK: nested greenlets at shutdownNrr!s& r!test_nested_greenlets_at_shutdown9TestInterpreterShutdown.test_nested_greenlets_at_shutdown`sS"668  F Q"6rd$vhvh OP 8&Arc  VPR4wrpVPV^RV RV V 24VPRV4R#)zU Greenlets in worker threads that are still referenced at shutdown should not crash. a 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") rrz$OK: 3 threaded greenlets at shutdownNrr!s& r#test_threaded_greenlets_at_shutdown;TestInterpreterShutdown.test_threaded_greenlets_at_shutdownwsS "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") rrOK: thread cleanup donezCLEANUP: GreenletExit caughtNrr!s& r(test_greenlet_cleanup_during_thread_exit@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exitsb"668  F4 Q"6rd$vhvh OP /8 4f=rc  VPR4wrpVPV^RV RV V 24VPRV4VPRV4R#)zT try/finally blocks in active greenlets run correctly when the owning thread exits. aR 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/zFINALLY: cleanup executedNrr!s& r%test_finally_block_during_thread_exit=TestInterpreterShutdown.test_finally_block_during_thread_exitsb "668  F, Q"6rd$vhvh OP /8 16:rc  VPR4wrpVPV^RV RV V 24VPRV4R#)zx Stress test: many active greenlets with cleanup code at shutdown. Ensures no crashes regardless of deallocation order. a 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. rrz#OK: 50 greenlets about to shut downNrr!s& r,test_many_greenlets_with_cleanup_at_shutdownDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdownsS "668  F0 Q"6rd$vhvh OP ;VDrc  VPR4wrpVPV^RV RV V 24VPRV4R#)zk Deeply nested greenlet parent chains at shutdown. Tests that the deallocation order doesn't cause issues. a 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}") rrzOK: nested to depth 10Nrr!s& r(test_deeply_nested_greenlets_at_shutdown@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdownsS "66 8  F Q"6rd$vhvh OP .7rc  VPR4wrpVPV^RV RV V 24VPRV4R#)zo A greenlet that has an active exception context when it's suspended should not crash during shutdown cleanup. a 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") rrz.OK: greenlet with active exception at shutdownNrr!s& r(test_greenlet_with_traceback_at_shutdown@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdown&sS "668  F Q"6rd$vhvh OP FOrN)__name__ __module__ __qualname____firstlineno__rr#r&r)r,r0r3r6r9r<__static_attributes____classdictcell__) __classdict__s@rrrsF ?&B0E.B. Fl#>J;>E@8*PPrr__main__) __doc__rr unittestr greenlet.testsrrr?mainr>rrrKsD$ #`Ph`PF  z MMOr