
    Ki.                     z    S r SSKrSSKrSSKrSSKrSSKJr   " S S\5      r\S:X  a  \R                  " 5         gg)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                   P    \ rS rSrS rS rS rS rS rS r	S r
S	 rS
 rS rSrg)TestInterpreterShutdown   c                     [         R                  " U5      n[        R                  " [        R
                  SU/SSSSS9nUR                  UR                  UR                  4$ )z
Run a Python script in a subprocess that exercises greenlet
during interpreter shutdown. Returns (returncode, stdout, stderr).
z-cT   F)capture_outputtexttimeoutcheck)	textwrapdedent
subprocessrunsys
executable
returncodestdoutstderr)selfscript_bodyfull_scriptresults       j/var/www/html/dynamic-report/venv/lib/python3.13/site-packages/greenlet/tests/test_interpreter_shutdown.py_run_shutdown_script,TestInterpreterShutdown._run_shutdown_script   sW    
 ook2^^T;/
   &-->>    c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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")
        r   Process crashed (rc=):
z OK: exiting with active greenletNr   assertEqualassertInr   rcr   r   s       r   )test_active_greenlet_at_shutdown_no_crashATestInterpreterShutdown.test_active_greenlet_at_shutdown_no_crash1   sS     "66 8 F 	Q"6rd$vhvh OP8&Ar   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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")
        r   r   r    z#OK: 10 active greenlets at shutdownNr!   r$   s       r   *test_multiple_active_greenlets_at_shutdownBTestInterpreterShutdown.test_multiple_active_greenlets_at_shutdownI   sS    
 "66 8 F 	Q"6rd$vhvh OP;VDr   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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")
        r   r   r    z OK: nested greenlets at shutdownNr!   r$   s       r   !test_nested_greenlets_at_shutdown9TestInterpreterShutdown.test_nested_greenlets_at_shutdown`   sS     "66 8 F  	Q"6rd$vhvh OP8&Ar   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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")
        r   r   r    z$OK: 3 threaded greenlets at shutdownNr!   r$   s       r   #test_threaded_greenlets_at_shutdown;TestInterpreterShutdown.test_threaded_greenlets_at_shutdownw   sS    
 "66 8 F4 	Q"6rd$vhvh OP<fEr   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        U R                  SU5        g)z
When a thread exits normally while holding active greenlets,
GreenletExit IS thrown and cleanup code runs.  This is the
standard cleanup path used in production (e.g. uWSGI worker
threads finishing a request).
a4              import os
            import threading
            import greenlet

            _write = os.write

            def thread_func():
                def worker(_w=_write,
                           _GreenletExit=greenlet.GreenletExit):
                    try:
                        greenlet.getcurrent().parent.switch("suspended")
                    except _GreenletExit:
                        _w(1, b"CLEANUP: GreenletExit caught\n")
                        raise

                g = greenlet.greenlet(worker)
                g.switch()
                # Thread exits with active greenlet -> thread-state
                # cleanup triggers GreenletExit

            t = threading.Thread(target=thread_func)
            t.start()
            t.join()
            print("OK: thread cleanup done")
        r   r   r    OK: thread cleanup donezCLEANUP: GreenletExit caughtNr!   r$   s       r   (test_greenlet_cleanup_during_thread_exit@TestInterpreterShutdown.test_greenlet_cleanup_during_thread_exit   sb     "66 8 F4 	Q"6rd$vhvh OP/84f=r   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        U R                  SU5        g)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")
        r   r   r    r2   zFINALLY: cleanup executedNr!   r$   s       r   %test_finally_block_during_thread_exit=TestInterpreterShutdown.test_finally_block_during_thread_exit   sb    
 "66 8 F, 	Q"6rd$vhvh OP/816:r   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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.
        r   r   r    z#OK: 50 greenlets about to shut downNr!   r$   s       r   ,test_many_greenlets_with_cleanup_at_shutdownDTestInterpreterShutdown.test_many_greenlets_with_cleanup_at_shutdown   sS    
 "66 8 F0 	Q"6rd$vhvh OP;VDr   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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}")
        r   r   r    zOK: nested to depth 10Nr!   r$   s       r   (test_deeply_nested_greenlets_at_shutdown@TestInterpreterShutdown.test_deeply_nested_greenlets_at_shutdown  sS    
 "66 8 F 	Q"6rd$vhvh OP.7r   c           	          U R                  S5      u  pnU R                  USSU SU U 35        U R                  SU5        g)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")
        r   r   r    z.OK: greenlet with active exception at shutdownNr!   r$   s       r   (test_greenlet_with_traceback_at_shutdown@TestInterpreterShutdown.test_greenlet_with_traceback_at_shutdown&  sS    
 "66 8 F  	Q"6rd$vhvh OPFOr    N)__name__
__module____qualname____firstlineno__r   r&   r)   r,   r/   r3   r6   r9   r<   r?   __static_attributes__rA   r   r   r   r      s=    ?&B0E.B. Fl#>J;>E@8*Pr   r   __main__)
__doc__r   r   unittestr   greenlet.testsr   r   rB   mainrA   r   r   <module>rL      sD   $     #`Ph `PF	 zMMO r   