עi.xdZddlZddlZddlZddlZddlmZGddeZedk(rejyy)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)TestCasecHeZdZdZdZdZdZdZdZdZ dZ d Z d Z y ) 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 i/sites/domain-asset-master/.venv/lib/python3.12/site-packages/greenlet/tests/test_interpreter_shutdown.py_run_shutdown_scriptz,TestInterpreterShutdown._run_shutdown_scriptsW ook2  ^^T; /    &-->>c |jd\}}}|j|dd|d|||jd|y)a8 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") rProcess crashed (rc=): z OK: exiting with active greenletNr assertEqualassertInrrcrrs r)test_active_greenlet_at_shutdown_no_crashzATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crash1sU"66 8  FF Q"6rd$vhvh OP 8&Arc |jd\}}}|j|dd|d|||jd|y)zm 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") rrrz#OK: 10 active greenlets at shutdownNrr"s r*test_multiple_active_greenlets_at_shutdownzBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownIsU "668  FF Q"6rd$vhvh OP ;VDrc |jd\}}}|j|dd|d|||jd|y)zQ 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") rrrz OK: nested greenlets at shutdownNrr"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)zm 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") rrrz$OK: 3 threaded greenlets at shutdownNrr"s r#test_threaded_greenlets_at_shutdownz;TestInterpreterShutdown.test_threaded_greenlets_at_shutdownwsU "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 caughtNrr"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)zl 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") rrrr,zFINALLY: cleanup executedNrr"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)z 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. rrrz#OK: 50 greenlets about to shut downNrr"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)z 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}") rrrzOK: nested to depth 10Nrr"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)z 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") rrrz.OK: greenlet with active exception at shutdownNrr"s r(test_greenlet_with_traceback_at_shutdownz@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdown&sU "668  FF Q"6rd$vhvh OP FOrN) __name__ __module__ __qualname__rr$r&r(r*r-r/r1r3r5rrrrs= ?&B0E.B. Fl#>J;>E@8*Prr__main__) __doc__rrunittestr greenlet.testsrrr6mainr9rrr?sF$ #`Ph`PF  zHMMOr