1 files changed, 84 insertions, 0 deletions
diff --git a/Lib/test/test_sys.py b/Lib/test/test_sys.py
index ae2a1c8..f1f1524 100644
--- a/Lib/test/test_sys.py
+++ b/Lib/test/test_sys.py
@@ -237,6 +237,90 @@ class SysModuleTest(unittest.TestCase):
             is sys._getframe().f_code
         )
 
+    # sys._current_frames() is a CPython-only gimmick.
+    def test_current_frames(self):
+        have_threads = True
+        try:
+            import thread
+        except ImportError:
+            have_threads = False
+
+        if have_threads:
+            self.current_frames_with_threads()
+        else:
+            self.current_frames_without_threads()
+
+    # Test sys._current_frames() in a WITH_THREADS build.
+    def current_frames_with_threads(self):
+        import threading, thread
+        import traceback
+
+        # Spawn a thread that blocks at a known place.  Then the main
+        # thread does sys._current_frames(), and verifies that the frames
+        # returned make sense.
+        entered_g = threading.Event()
+        leave_g = threading.Event()
+        thread_info = []  # the thread's id
+
+        def f123():
+            g456()
+
+        def g456():
+            thread_info.append(thread.get_ident())
+            entered_g.set()
+            leave_g.wait()
+
+        t = threading.Thread(target=f123)
+        t.start()
+        entered_g.wait()
+
+        # At this point, t has finished its entered_g.set(), although it's
+        # impossible to guess whether it's still on that line or has moved on
+        # to its leave_g.wait().
+        self.assertEqual(len(thread_info), 1)
+        thread_id = thread_info[0]
+
+        d = sys._current_frames()
+
+        main_id = thread.get_ident()
+        self.assert_(main_id in d)
+        self.assert_(thread_id in d)
+
+        # Verify that the captured main-thread frame is _this_ frame.
+        frame = d.pop(main_id)
+        self.assert_(frame is sys._getframe())
+
+        # Verify that the captured thread frame is blocked in g456, called
+        # from f123.  This is a litte tricky, since various bits of
+        # threading.py are also in the thread's call stack.
+        frame = d.pop(thread_id)
+        stack = traceback.extract_stack(frame)
+        for i, (filename, lineno, funcname, sourceline) in enumerate(stack):
+            if funcname == "f123":
+                break
+        else:
+            self.fail("didn't find f123() on thread's call stack")
+
+        self.assertEqual(sourceline, "g456()")
+
+        # And the next record must be for g456().
+        filename, lineno, funcname, sourceline = stack[i+1]
+        self.assertEqual(funcname, "g456")
+        self.assert_(sourceline in ["leave_g.wait()", "entered_g.set()"])
+
+        # Reap the spawned thread.
+        leave_g.set()
+        t.join()
+
+    # Test sys._current_frames() when thread support doesn't exist.
+    def current_frames_without_threads(self):
+        # Not much happens here:  there is only one thread, with artificial
+        # "thread id" 0.
+        d = sys._current_frames()
+        self.assertEqual(len(d), 1)
+        self.assert_(0 in d)
+        self.assert_(d[0] is sys._getframe())
+
     def test_attributes(self):
         self.assert_(isinstance(sys.api_version, int))
         self.assert_(isinstance(sys.argv, list))