= i/dZddlZddlZddlZddlZddlmZGddeZedkrej dSdS)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)TestCasecJeZdZdZdZdZdZdZdZdZ dZ d Z d Z d S) TestInterpreterShutdownctj|}tjtjd|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 C:\Users\Dell Inspiron 16\Desktop\tws\AgrotaPowerBi\back-agrota-powerbi\mcp-client-agrota\venv\Lib\site-packages\greenlet/tests/test_interpreter_shutdown.py_run_shutdown_scriptz,TestInterpreterShutdown._run_shutdown_scriptsW ok22  ^T; /     &->>c |d\}}}||dd|d|||d|dS)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_crash1sp"66 8   FF Q Or O Ov Ov O OPPP 8&AAAAArc |d\}}}||dd|d|||d|dS)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_shutdownIsp "668   FF Q Or O Ov Ov O OPPP ;VDDDDDrc |d\}}}||dd|d|||d|dS)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`sp"668   FF Q Or O Ov Ov O OPPP 8&AAAAArc |d\}}}||dd|d|||d|dS)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_shutdownwsp "668   FF4 Q Or O Ov Ov O OPPP  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_exits"668   FF4 Q Or O Ov Ov O OPPP /888 4f=====rc |d\}}}||dd|d|||d||d|dS)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_exits "668   FF, Q Or O Ov Ov O OPPP /888 16:::::rc |d\}}}||dd|d|||d|dS)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_shutdownsp "668   FF0 Q Or O Ov Ov O OPPP ;VDDDDDrc |d\}}}||dd|d|||d|dS)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_shutdownsp "66 8   FF Q Or O Ov Ov O OPPP .77777rc |d\}}}||dd|d|||d|dS)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&sp "668   FF Q Or O Ov Ov O OPPP FOOOOOrN) __name__ __module__ __qualname__rr$r&r(r*r-r/r1r3r5rrrrs ? ? ?&BBB0EEE.BBB. F F Fl#>#>#>J;;;>EEE@888*PPPPPrr__main__) __doc__rrunittestr greenlet.testsrrr6mainr9rrr?s$ ######`P`P`P`P`Ph`P`P`PF  zHMOOOOOr