self = <test_013_SqlHandle_free_shutdown.TestHandleFreeShutdown object at 0x1107ca350>
conn_str = 'Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=Azure@123!;TrustServerCertificate=yes'

    def test_aggressive_dbc_segfault_reproduction(self, conn_str):
        """
        AGGRESSIVE TEST: Try to reproduce DBC handle segfault during shutdown.
    
        This test aggressively attempts to trigger the segfault described in the stack trace
        by creating many DBC handles and forcing Python to shut down while they're still alive.
    
        Current vulnerability: DBC handles (Type 2) are NOT protected during shutdown,
        so they will call SQLFreeHandle during finalization, potentially accessing
        the already-destructed static ENV handle.
    
        Expected with CURRENT CODE: May segfault (this is the bug we're testing for)
        Expected with FIXED CODE: No segfault
        """
        script = textwrap.dedent(f"""
            import sys
            import gc
            from mssql_python import connect
    
            print("=== AGGRESSIVE DBC SEGFAULT TEST ===")
            print("Creating many DBC handles and forcing shutdown...")
    
            # Create many connections without closing them
            # This maximizes the chance of DBC handles being finalized
            # AFTER the static ENV handle has destructed
            connections = []
            for i in range(5):  # Reduced for faster execution
                conn = connect("{conn_str}")
                # Don't even create cursors - just DBC handles
                connections.append(conn)
    
            print(f"Created {{len(connections)}} DBC handles")
            print("Forcing GC to ensure objects are tracked...")
            gc.collect()
    
            # Delete the list but objects are still alive in GC
            del connections
    
            print("WARNING: About to exit with unclosed DBC handles")
            print("If Type 2 (DBC) handles are not protected, this may SEGFAULT")
            print("Stack trace will show: SQLFreeHandle -> SqlHandle::free() -> finalize_garbage")
    
            # Force immediate exit - this triggers finalize_garbage
            sys.exit(0)
        """)
    
>       result = subprocess.run(
            [sys.executable, "-c", script], capture_output=True, text=True, timeout=5
        )

tests/test_013_SqlHandle_free_shutdown.py:87: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:556: in run
    stdout, stderr = process.communicate(input, timeout=timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:1222: in communicate
    stdout, stderr = self._communicate(input, endtime, timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:2155: in _communicate
    self._check_timeout(endtime, orig_timeout, stdout, stderr)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: -9 args: ['/Library/Frameworks/Python.framework/Versions...>
endtime = 2472.695417551, orig_timeout = 5, stdout_seq = [], stderr_seq = []
skip_check_and_raise = False

    def _check_timeout(self, endtime, orig_timeout, stdout_seq, stderr_seq,
                       skip_check_and_raise=False):
        """Convenience for checking if a timeout has expired."""
        if endtime is None:
            return
        if skip_check_and_raise or _time() > endtime:
>           raise TimeoutExpired(
                    self.args, orig_timeout,
                    output=b''.join(stdout_seq) if stdout_seq else None,
                    stderr=b''.join(stderr_seq) if stderr_seq else None)
E           subprocess.TimeoutExpired: Command '['/Library/Frameworks/Python.framework/Versi

self = <test_013_SqlHandle_free_shutdown.TestHandleFreeShutdown object at 0x110798510>
conn_str = 'Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=Azure@123!;TrustServerCertificate=yes'

    def test_force_gc_finalization_order_issue(self, conn_str):
        """
        TEST: Force specific GC finalization order to trigger segfault.
    
        By creating objects in specific order and forcing GC cycles,
        we try to ensure DBC handles are finalized after ENV handle destruction.
    
        Expected with CURRENT CODE: May segfault
        Expected with FIXED CODE: No segfault
        """
        script = textwrap.dedent(f"""
            import sys
            import gc
            import weakref
            from mssql_python import connect
    
            print("=== FORCED GC FINALIZATION ORDER TEST ===")
    
            # Create many connections
            connections = []
            weakrefs = []
    
            for i in range(5):  # Reduced for faster execution
                conn = connect("{conn_str}")
                wr = weakref.ref(conn)
                connections.append(conn)
                weakrefs.append(wr)
    
            print(f"Created {{len(connections)}} connections with weakrefs")
    
            # Force GC to track these objects
            gc.collect()
    
            # Delete strong references
            del connections
    
            # Force GC cycles
            print("Forcing GC cycles...")
            for i in range(2):
                collected = gc.collect()
                print(f"GC cycle {{i+1}}: collected {{collected}} objects")
    
            # Check weakrefs
            alive = sum(1 for wr in weakrefs if wr() is not None)
            print(f"Weakrefs still alive: {{alive}}")
    
            print("Exiting - finalize_garbage will be called")
            print("If DBC handles aren't protected, segfault in SQLFreeHandle")
            sys.exit(0)
        """)
    
>       result = subprocess.run(
            [sys.executable, "-c", script], capture_output=True, text=True, timeout=5
        )

tests/test_013_SqlHandle_free_shutdown.py:207: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:556: in run
    stdout, stderr = process.communicate(input, timeout=timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:1222: in communicate
    stdout, stderr = self._communicate(input, endtime, timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:2155: in _communicate
    self._check_timeout(endtime, orig_timeout, stdout, stderr)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: -9 args: ['/Library/Frameworks/Python.framework/Versions...>
endtime = 2480.105544492, orig_timeout = 5, stdout_seq = [], stderr_seq = []
skip_check_and_raise = False

    def _check_timeout(self, endtime, orig_timeout, stdout_seq, stderr_seq,
                       skip_check_and_raise=False):
        """Convenience for checking if a timeout has expired."""
        if endtime is None:
            return
        if skip_check_and_raise or _time() > endtime:
>           raise TimeoutExpired(
                    self.args, orig_timeout,
                    output=b''.join(stdout_seq) if stdout_seq else None,
                    stderr=b''.join(stderr_seq) if stderr_seq else None)
E           subprocess.TimeoutExpired: Command '['/Library/Frameworks/Python.framework/Versions/3.13/bin/python', '-c', '\nimport sys\nimport gc\nimport weakref\nfrom mssql_python import connect\n\nprint("=== FORCED GC FINALIZATION ORDER TEST ===")\n\n# Create many connections\nconnections = []\nweakrefs = []\n\nfor i in range(5):  # Reduced for faster execution\n    con

self = <test_013_SqlHandle_free_shutdown.TestHandleFreeShutdown object at 0x11073dd90>
conn_str = 'Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=Azure@123!;TrustServerCertificate=yes'

    def test_dbc_handle_cleanup_at_shutdown(self, conn_str):
        """
        Test DBC handle (Type 2) cleanup during Python shutdown.
    
        Scenario:
        1. Create multiple connections (multiple DBC handles)
        2. Close cursors but leave connections open
        3. Let Python shutdown without closing connections
        4. DBC handles' __del__ should skip SQLFreeHandle during shutdown
    
        Expected: No segfault, clean exit
        """
        script = textwrap.dedent(f"""
            import sys
            from mssql_python import connect
    
            # Create multiple connections (DBC handles)
            connections = []
            for i in range(3):
                conn = connect("{conn_str}")
                cursor = conn.cursor()
                cursor.execute(f"SELECT {{i}} AS test_value")
                result = cursor.fetchall()
                cursor.close()  # Close cursor, but keep connection
                connections.append(conn)
                print(f"Connection {{i}}: created and cursor closed")
    
            # Intentionally skip connection cleanup
            # This will trigger SqlHandle::free() for DBC handles during shutdown
            # Type 2 (DBC) handles should be skipped when pythonShuttingDown=true
            print("DBC handle cleanup test: Exiting without explicit connection cleanup")
            sys.exit(0)
        """)
    
>       result = subprocess.run(
            [sys.executable, "-c", script], capture_output=True, text=True, timeout=5
        )

tests/test_013_SqlHandle_free_shutdown.py:292: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:556: in run
    stdout, stderr = process.communicate(input, timeout=timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:1222: in communicate
    stdout, stderr = self._communicate(input, endtime, timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:2155: in _communicate
    self._check_timeout(endtime, orig_timeout, stdout, stderr)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: -9 args: ['/Library/Frameworks/Python.framework/Versions...>
endtime = 2489.974983749, orig_timeout = 5, stdout_seq = [], stderr_seq = []
skip_check_and_raise = False

    def _check_timeout(self, endtime, orig_timeout, stdout_seq, stderr_seq,
                       skip_check_and_raise=False):
        """Convenience for checking if a timeout has expired."""
        if endtime is None:
            return
        if skip_check_and_raise or _time() > endtime:
>           raise TimeoutExpired(
                    self.args, orig_timeout,
                    output=b''.join(stdout_seq) if stdout_seq else None,
                    stderr=b''.join(stderr_seq) if stderr_seq else None)
E           subprocess.TimeoutExpired: Command '['/Library/Frameworks/Python.framework/Versions/3.13/bin/python', '-c', '\nimport sys\nfrom mssql_python import connect\n\n# Create multiple connections (DBC handles)\nconnections = []\nfor i in range(3):\n    conn = connect("Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=Azure@123!;TrustServerCertificate=yes")\n    cursor = conn.cursor()\n    cursor.execute(f"SELECT {i} AS test_value")\n    result = cursor.fetchall()\n    cursor.close()  # Close cursor, but keep connection\n    connections.append(conn)\n    print(f"Connection {i}: created and cursor closed")\n\n# Intentionally skip connection cleanup\n# This will trigger SqlHandle::free() for DBC handles duri

self = <test_013_SqlHandle_free_shutdown.TestHandleFreeShutdown object at 0x1107019d0>
conn_str = 'Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=Azure@123!;TrustServerCertificate=yes'

    def test_env_handle_cleanup_at_shutdown(self, conn_str):
        """
        Test ENV handle (Type 1) cleanup during Python shutdown.
    
        Scenario:
        1. Create and close connections (ENV handle is static singleton)
        2. Let Python shutdown
        3. ENV handle is static and should follow normal C++ destruction
        4. ENV handle should NOT be skipped (no protection needed)
    
        Expected: No segfault, clean exit
        Note: ENV handle is static and destructs via normal C++ mechanisms,
              not during Python GC. This test verifies the overall flow.
        """
        script = textwrap.dedent(f"""
            import sys
            from mssql_python import connect
    
            # Create and properly close connections
            # ENV handle is static singleton shared across all connections
            for i in range(3):
                conn = connect("{conn_str}")
                cursor = conn.cursor()
                cursor.execute(f"SELECT {{i}} AS test_value")
                cursor.fetchall()
                cursor.close()
                conn.close()
                print(f"Connection {{i}}: properly closed")
    
            # ENV handle is static and will destruct via C++ static destruction
            # It does NOT have pythonShuttingDown protection (Type 1 not in check)
            print("ENV handle cleanup test: All connections closed properly")
            sys.exit(0)
        """)
    
>       result = subprocess.run(
            [sys.executable, "-c", script], capture_output=True, text=True, timeout=5
        )

tests/test_013_SqlHandle_free_shutdown.py:340: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:556: in run
    stdout, stderr = process.communicate(input, timeout=timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:1222: in communicate
    stdout, stderr = self._communicate(input, endtime, timeout)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
/Library/Frameworks/Python.framework/Versions/3.13/lib/python3.13/subprocess.py:2155: in _communicate
    self._check_timeout(endtime, orig_timeout, stdout, stderr)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Popen: returncode: -9 args: ['/Library/Frameworks/Python.framework/Versions...>
endtime = 2497.789568981, orig_timeout = 5, stdout_seq = [], stderr_seq = []
skip_check_and_raise = False

    def _check_timeout(self, endtime, orig_timeout, stdout_seq, stderr_seq,
                       skip_check_and_raise=False):
        """Convenience for checking if a timeout has expired."""
        if endtime is None:
            return
        if skip_check_and_raise or _time() > endtime:
>           raise TimeoutExpired(
                    self.args, orig_timeout,
                    output=b''.join(stdout_seq) if stdout_seq else None,
                    stderr=b''.join(stderr_seq) if stderr_seq else None)
E           subprocess.TimeoutExpired: Command '['/Library/Frameworks/Python.framework/Versions/3.13/bin/python', '-c', '\nimport sys\nfrom mssql_python import connect\n\n# Create and properly close connections\n# ENV handle is static singleton shared across all connections\nfor i in range(3):\n    conn = connect("Server=tcp:127.0.0.1,1433;Database=master;Uid=SA;Pwd=Azure@123!;TrustServerCertificate=yes")\n    cursor = conn.cursor()\n    cursor.execute(f"SELECT {i} AS test_value")\n    cursor.fetchall()\n    cursor.close()\n    conn.close()\n    print(f"Connection {i}: properly closed")\n\n# ENV handle is static and will destruct via C++ static destruction\n#