gh-92031, test_embed: Improve test for unquickening static code (#92440)

1 files changed, 30 insertions, 10 deletions

diff --git a/Lib/test/test_embed.py b/Lib/test/test_embed.py
index 169ae5c..5ba6e3a 100644
--- a/Lib/test/test_embed.py
+++ b/Lib/test/test_embed.py
@@ -343,19 +343,39 @@ class EmbeddingTests(EmbeddingTestsMixin, unittest.TestCase):
         out, err = self.run_embedded_interpreter("test_repeated_init_exec", code)
         self.assertEqual(out, 'Tests passed\n' * INIT_LOOPS)
 
-    @support.skip_if_pgo_task
     def test_quickened_static_code_gets_unquickened_at_Py_FINALIZE(self):
         # https://github.com/python/cpython/issues/92031
-        code = """if 1:
-            from importlib._bootstrap import _handle_fromlist
-            import dis
-            for name in dis.opmap:
-                # quicken this frozen code object.
-                _handle_fromlist(dis, [name], lambda *args: None)
-        """
+
+        # Do these imports outside of the code string to avoid using
+        # importlib too much from within the code string, so that
+        # _handle_fromlist doesn't get quickened until we intend it to.
+        from dis import _all_opmap
+        resume = _all_opmap["RESUME"]
+        resume_quick = _all_opmap["RESUME_QUICK"]
+        from test.test_dis import QUICKENING_WARMUP_DELAY
+
+        code = textwrap.dedent(f"""\
+            import importlib._bootstrap
+            func = importlib._bootstrap._handle_fromlist
+            code = func.__code__
+
+            # Assert initially unquickened.
+            # Use sets to account for byte order.
+            if set(code._co_code_adaptive[:2]) != set([{resume}, 0]):
+                raise AssertionError()
+
+            for i in range({QUICKENING_WARMUP_DELAY}):
+                func(importlib._bootstrap, ["x"], lambda *args: None)
+
+            # Assert quickening worked
+            if set(code._co_code_adaptive[:2]) != set([{resume_quick}, 0]):
+                raise AssertionError()
+
+            print("Tests passed")
+        """)
         run = self.run_embedded_interpreter
-        for i in range(50):
-            out, err = run("test_repeated_init_exec", code, timeout=60)
+        out, err = run("test_repeated_init_exec", code)
+        self.assertEqual(out, 'Tests passed\n' * INIT_LOOPS)
 
     def test_ucnhash_capi_reset(self):
         # bpo-47182: unicodeobject.c:ucnhash_capi was not reset on shutdown.