temp: escape-test canary (will remove)

KingLizard1020 · Sep 17, 2026 · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026
commit ab3acf17af550464564fbda57ee3f3b6bed99745
diff --git a/Lib/test/_restore_canary.txt b/Lib/test/_restore_canary.txt
@@ -1 +1,145 @@
-canary-ok
+:
+                x = C()
+                ready.release()
+                time.sleep(100)
+
+            _thread.start_new_thread(waitingThread, ())
+            ready.acquire()  # Be sure the other thread is waiting.
+            sys.exit(42)
+            """)
+        self.assertEqual(rc, 42)
+
+    def test_finalize_with_trace(self):
+        # Issue1733757
+        # Avoid a deadlock when sys.settrace steps into threading._shutdown
+        if support.check_sanitizer(thread=True):
+            # the thread running `time.sleep(2)` below will still be alive
+            # at process exit
+            self.skipTest("TSAN would report thread leak")
+
+        assert_python_ok("-c", """if 1:
+            import sys, threading
+
+            # A deadlock-killer, to prevent the
+            # testsuite to hang forever
+            def killer():
+                import os, time
+                time.sleep(2)
+                print('program blocked; aborting')
+                os._exit(2)
+            t = threading.Thread(target=killer)
+            t.daemon = True
+            t.start()
+
+            # This is the trace function
+            def func(frame, event, arg):
+                threading.current_thread()
+                return func
+
+            sys.settrace(func)
+            """)
+
+    def test_join_nondaemon_on_shutdown(self):
+        # Issue 1722344
+        # Raising SystemExit skipped threading._shutdown
+        rc, out, err = assert_python_ok("-c", """if 1:
+                import threading
+                from time import sleep
+
+                def child():
+                    sleep(1)
+                    # As a non-daemon thread we SHOULD wake up and nothing
+                    # should be torn down yet
+                    print("Woke up, sleep function is:", sleep)
+
+                threading.Thread(target=child).start()
+                raise SystemExit
+            """)
+        self.assertEqual(out.strip(),
+            b"Woke up, sleep function is: <built-in function sleep>")
+        self.assertEqual(err, b"")
+
+    def test_enumerate_after_join(self):
+        # Try hard to trigger #1703448: a thread is still returned in
+        # threading.enumerate() after it has been join()ed.
+        enum = threading.enumerate
+        old_interval = sys.getswitchinterval()
+        try:
+            for i in range(1, 100):
+                support.setswitchinterval(i * 0.0002)
+                t = threading.Thread(target=lambda: None)
+                t.start()
+                t.join()
+                l = enum()
+                self.assertNotIn(t, l,
+                    "#1703448 triggered after %d trials: %s" % (i, l))
+        finally:
+            sys.setswitchinterval(old_interval)
+
+    @support.bigmemtest(size=20, memuse=72*2**20, dry_run=False)
+    def test_join_from_multiple_threads(self, size):
+        # Thread.join() should be thread-safe
+        errors = []
+
+        def worker():
+            time.sleep(0.005)
+
+        def joiner(thread):
+            try:
+                thread.join()
+            except Exception as e:
+                errors.append(e)
+
+        for N in range(2, 20):
+            threads = [threading.Thread(target=worker)]
+            for i in range(N):
+                threads.append(threading.Thread(target=joiner,
+                                                args=(threads[0],)))
+            for t in threads:
+                t.start()
+            time.sleep(0.01)
+            for t in threads:
+                t.join()
+            if errors:
+                raise errors[0]
+
+    def test_join_with_timeout(self):
+        lock = _thread.allocate_lock()
+        lock.acquire()
+
+        def worker():
+            lock.acquire()
+
+        thread = threading.Thread(target=worker)
+        thread.start()
+        thread.join(timeout=0.01)
+        assert thread.is_alive()
+        lock.release()
+        thread.join()
+        assert not thread.is_alive()
+
+    def test_no_refcycle_through_target(self):
+        class RunSelfFunction(object):
+            def __init__(self, should_raise):
+                # The links in this refcycle from Thread back to self
+                # should be cleaned up when the thread completes.
+                self.should_raise = should_raise
+                self.thread = threading.Thread(target=self._run,
+                                               args=(self,),
+                                               kwargs={'yet_another':self})
+                self.thread.start()
+
+            def _run(self, other_ref, yet_another):
+                if self.should_raise:
+                    raise SystemExit
+
+        restore_default_excepthook(self)
+
+        cyclic_object = RunSelfFunction(should_raise=False)
+        weak_cyclic_object = weakref.ref(cyclic_object)
+        cyclic_object.thread.join()
+        del cyclic_object
+        self.assertIsNone(weak_cyclic_object(),
+                         msg=('%d references still around' %
+                              sys.getrefcount(weak_cyclic_object())))
+