V.j^xdZddlZddlZddlZddlZddlmZGddeZedk(rejyy)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)TestCaseceZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZdZdZdZdZdZdZy)TestInterpreterShutdownctj|}tjtj d|gdddd}|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_scriptresults jC:\xampp\htdocs\tradingbinance\backend\.venv\Lib\site-packages\greenlet/tests/test_interpreter_shutdown.py_run_shutdown_scriptz,TestInterpreterShutdown._run_shutdown_scriptsW ook2  ^^T; /    &-->>c |jd\}}}|j|dd|d|||jd|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") rProcess crashed (rc=): z OK: exiting with active greenletr assertEqualassertInrrcrrs r)test_active_greenlet_at_shutdown_no_crashzATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crash0sU"66 8  FF Q"6rd$vhvh OP 8&Arc |jd\}}}|j|dd|d|||jd|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") rrrz#OK: 10 active greenlets at shutdownrr"s r*test_multiple_active_greenlets_at_shutdownzBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownHsU"668  FF Q"6rd$vhvh OP ;VDrc |jd\}}}|j|dd|d|||jd|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") rrrz OK: nested greenlets at shutdownrr"s r!test_nested_greenlets_at_shutdownz9TestInterpreterShutdown.test_nested_greenlets_at_shutdown^sU"668  FF Q"6rd$vhvh OP 8&Arc |jd\}}}|j|dd|d|||jd|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") rrrz$OK: 3 threaded greenlets at shutdownrr"s r#test_threaded_greenlets_at_shutdownz;TestInterpreterShutdown.test_threaded_greenlets_at_shutdowntsU"668  FF4 Q"6rd$vhvh OP  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 caughtrr"s r(test_greenlet_cleanup_during_thread_exitz@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exitsd "668  FF4 Q"6rd$vhvh OP /8 4f=rc |jd\}}}|j|dd|d|||jd||jd|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") rrrr,zFINALLY: cleanup executedrr"s r%test_finally_block_during_thread_exitz=TestInterpreterShutdown.test_finally_block_during_thread_exitsd"668  FF, Q"6rd$vhvh OP /8 16:rc |jd\}}}|j|dd|d|||jd|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. rrrz#OK: 50 greenlets about to shut downrr"s r,test_many_greenlets_with_cleanup_at_shutdownzDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdownsU"668  FF0 Q"6rd$vhvh OP ;VDrc |jd\}}}|j|dd|d|||jd|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}") rrrzOK: nested to depth 10rr"s r(test_deeply_nested_greenlets_at_shutdownz@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdownsU"66 8  FF Q"6rd$vhvh OP .7rc |jd\}}}|j|dd|d|||jd|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") rrrz.OK: greenlet with active exception at shutdownrr"s r(test_greenlet_with_traceback_at_shutdownz@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdownsU"668  FF Q"6rd$vhvh OP FOrc |jd\}}}|j|dd|d|||jd||jd|dy)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)rr"s r&test_getcurrent_during_atexit_no_crashz>TestInterpreterShutdown.test_getcurrent_during_atexit_no_crash.sm "668  FF& Q"6rd$vhvh OP -v6 >J Krc |jd\}}}|j|dd|d|||jd|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") rrrOK: registeredrr"s r$test_gettrace_during_atexit_no_crashzrc |jd\}}}|j|dd|d|||jd||jd|dy)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") rrrz)OK: 5 active greenlets, atexit registeredr8r9rr"s r3test_getcurrent_with_active_greenlets_during_atexitzKTestInterpreterShutdown.test_getcurrent_with_active_greenlets_during_atexittsm "668  FF8 Q"6rd$vhvh OP A6J >J Krc |jd\}}}|j|dd|d|||jd||jd|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") rrrr7z!OK: created greenlet successfullyrr"s r1test_greenlet_construction_during_atexit_no_crashzITestInterpreterShutdown.test_greenlet_construction_during_atexit_no_crashsd"668  FF& Q"6rd$vhvh OP -v6 96Brc |jd\}}}|j|dd|d|||jd|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") rrr*OK: 10 active greenlets, atexit registeredrr"s r>test_greenlet_construction_with_active_greenlets_during_atexitzVTestInterpreterShutdown.test_greenlet_construction_with_active_greenlets_during_atexitsU "668  FF4 Q"6rd$vhvh OP BFKrc |jd\}}}|j|dd|d|||jd|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") rrrz.OK: cross-thread setup done, atexit registeredrr"s rCtest_greenlet_construction_with_cross_thread_deleteme_during_atexitz[TestInterpreterShutdown.test_greenlet_construction_with_cross_thread_deleteme_during_atexitsV"66"8 " FFF Q"6rd$vhvh OP FOrc |jd\}}}|j|dd|d|||jd||jd|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") rrrzOK: deferred cycle createdGUARDED: getcurrent=Nonerr"s r3test_getcurrent_returns_none_during_gc_finalizationzKTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalizationsd"668  FF8 Q"6rd$vhvh OP 2F; 0&9rc |jd\}}}|j|dd|d|||jd||jd|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") rrrz'OK: 10 active greenlets, cycle deferredrKrr"s rItest_getcurrent_returns_none_during_gc_finalization_with_active_greenletszaTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_with_active_greenlets>se"66$8 $ FFJ Q"6rd$vhvh OP ?H 0&9rc |jd\}}}|j|dd|d|||jd||jd|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") rrrz)OK: cross-thread cleanup + cycle deferredrKrr"s r@test_getcurrent_returns_none_during_gc_finalization_cross_threadzXTestInterpreterShutdown.test_getcurrent_returns_none_during_gc_finalization_cross_threadkse"66*8 * FFV Q"6rd$vhvh OP A6J 0&9rc |jd\}}}|j|dd|d|||jd||jd|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") rrrz,OK: atexit registered before greenlet import UNGUARDEDrr"s r0test_getcurrent_returns_none_during_atexit_phasezHTestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phasesc"668  FF0 Q"6rd$vhvh OP DfM k6*rc |jd\}}}|j|dd|d|||jd||jd|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") rrrrFrRrr"s rFtest_getcurrent_returns_none_during_atexit_phase_with_active_greenletsz^TestInterpreterShutdown.test_getcurrent_returns_none_during_atexit_phase_with_active_greenletssa!668  FF< Q"6rd$vhvh OP BFK k6*rc |jd\}}}|j|dd|d|||jd||jd|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() rrrDestructor runningz)RuntimeError: greenlet is being finalizedrr"s r5test_api_getcurrent_no_system_error_at_module_gc_timezMTestInterpreterShutdown.test_api_getcurrent_no_system_error_at_module_gc_timesd"66 8  FF Q"6rd$vhvh OP *F3 A6Jrc |jd\}}}|j|dd|d|||jd||jd|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() rrrrWzResult 1rr"s r&test_switch_no_error_at_module_gc_timez>TestInterpreterShutdown.test_switch_no_error_at_module_gc_timesc"668  FF0 Q"6rd$vhvh OP *F3 j&)rN)__name__ __module__ __qualname__rr$r&r(r*r-r/r1r3r5r:r=r@rBrDrGrIrLrNrPrSrUrXrZrrrrs ?&B0E,B,FT">H;:E>8(P8K<0(0(%KNC6 LD+P|&:P+:Z4:V+B"+HK2*rr__main__) __doc__rrunittestr greenlet.testsrrr[mainr^rrrdsD" #V *hV *r zHMMOr