Try to fix tests

sergey-miryanov · sergey-miryanov · commit 6a47bca6ca7f · 2026-04-12T00:36:05.000+05:00
diff --git a/Lib/test/test_get_gc_stats.py b/Lib/test/test_get_gc_stats.py
@@ -1,27 +1,21 @@
-import gc
-import json
-import os
-import subprocess
-import sys
 import textwrap
+import time
 import unittest
 
 from test.support import (
-    import_helper,
-    SHORT_TIMEOUT,
     requires_gil_enabled,
     requires_remote_subprocess_debugging,
 )
-
-PROCESS_VM_READV_SUPPORTED = False
+from test.test_profiling.test_sampling_profiler.helpers import test_subprocess
 
 try:
-    from _remote_debugging import PROCESS_VM_READV_SUPPORTED
+    import _remote_debugging  # noqa: F401
 except ImportError:
     raise unittest.SkipTest(
         "Test only runs when _remote_debugging is available"
     )
 
+
 def get_interpreter_identifiers(gc_stats: tuple[dict[str, str|int|float]]) -> tuple[str,...]:
     return tuple(sorted({s["iid"] for s in gc_stats}))
 
@@ -45,72 +39,90 @@ def get_last_item(gc_stats: tuple[dict[str, str|int|float]],
 
     return item
 
-skip_if_not_supported = unittest.skipIf(
-    (
-        sys.platform != "darwin"
-        and sys.platform != "linux"
-        and sys.platform != "win32"
-    ),
-    "Test only runs on Linux, Windows and MacOS",
-)
-
 
 @requires_gil_enabled()
 @requires_remote_subprocess_debugging()
 class TestGetGCStats(unittest.TestCase):
 
-    def _run_child_process(self, all_interpreters):
-        # Run the test in a subprocess to avoid side effects
-        script = textwrap.dedent(f"""\
-            import json
-            import os
-            import sys
-            import _remote_debugging
-            try:
-                from _remote_debugging import PROCESS_VM_READV_SUPPORTED
-                supported = True
-            except ImportError:
-                supported = False
-
-            if supported:
-                pid = int(sys.argv[1])
-                gc_stats = _remote_debugging.get_gc_stats(pid, all_interpreters={all_interpreters})
-                print(json.dumps(gc_stats, indent=1))
-            else:
-                print(json.dumps(dict([("error", "not supported")])))
-            """)
-
-        gc.collect(0)
-        gc.collect(1)
-        gc.collect(2)
-
-        result = subprocess.run(
-            [sys.executable, "-c", script, str(os.getpid())],
-            capture_output=True,
-            text=True,
-            timeout=SHORT_TIMEOUT,
-        )
-        self.assertEqual(
-            result.returncode, 0,
-            f"stdout: {result.stdout}\nstderr: {result.stderr}"
-        )
-        data = json.loads(result.stdout)
-        if isinstance(data, dict) and "error" in data:
-            if sys.platform == "linux":
-                self.skipTest("Testing on Linux requires process_vm_readv support")
-            else:
-                self.assertTrue(False, f"Unexpected error: {data}")
-        return data
-
-    def _run_in_interpreter(self, interp):
-        source = f"""if True:
-        import gc
-
-        gc.collect(0)
-        gc.collect(1)
-        gc.collect(2)
-        """
-        interp.exec(source)
+    @classmethod
+    def setUpClass(cls):
+        cls._main_iid = 0 # main interpreter ID
+        cls._only_main_interpreter_script = '''
+            import gc
+            import time
+
+            gc.collect(0)
+            gc.collect(1)
+            gc.collect(2)
+
+            _test_sock.sendall(b"working")
+
+            objects = []
+            while True:
+                if len(objects) > 100:
+                    objects = []
+
+                # objects that GC will visit should increase
+                objects.append(object())
+
+                time.sleep(0.1)
+                gc.collect(0)
+                gc.collect(1)
+                gc.collect(2)
+            '''
+        cls._subinterpreters_script = '''
+            import concurrent.interpreters as interpreters
+            import gc
+            import time
+
+            source = """if True:
+                import gc
+
+                gc.collect(0)
+                gc.collect(1)
+                gc.collect(2)
+            """
+
+            interp = interpreters.create()
+            interp.exec(source)
+
+            gc.collect(0)
+            gc.collect(1)
+            gc.collect(2)
+
+            _test_sock.sendall(b"working")
+            objects = []
+            while True:
+                if len(objects) > 100:
+                    objects = []
+
+                # objects that GC will visit should increase
+                objects.append(object())
+
+                time.sleep(0.1)
+                interp.exec(source)
+                gc.collect(0)
+                gc.collect(1)
+                gc.collect(2)
+            '''
+
+    def _collect_gc_stats(self, script:str, all_interpreters:bool):
+        get_gc_stats = _remote_debugging.get_gc_stats
+        with (
+            test_subprocess(script, wait_for_working=True) as subproc
+        ):
+            before_stats = get_gc_stats(subproc.process.pid,
+                                        all_interpreters=all_interpreters)
+            before = get_last_item(before_stats, 2, self._main_iid)
+            for _ in range(10):
+                time.sleep(0.5)
+                after_stats = get_gc_stats(subproc.process.pid,
+                                           all_interpreters=all_interpreters)
+                after = get_last_item(after_stats, 2, self._main_iid)
+                if after["ts_stop"] > before["ts_stop"]:
+                    break
+
+        return before_stats, after_stats
 
     def _check_gc_state(self, before, after):
         self.assertIsNotNone(before)
@@ -136,54 +148,49 @@ def _check_gc_state(self, before, after):
                                     before["objects_not_transitively_reachable"],
                                     (before, after))
 
-    @skip_if_not_supported
-    @unittest.skipIf(
-        sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
-        "Test only runs on Linux with process_vm_readv support",
-    )
-    def test_get_gc_stats_for_main_interpreter(self):
-        before_stats = self._run_child_process(False)
-        after_stats = self._run_child_process(False)
-
+    def _check_main_interpreter_stats(self, before_stats, after_stats):
         before_iids = get_interpreter_identifiers(before_stats)
         after_iids = get_interpreter_identifiers(after_stats)
 
         self.assertEqual(before_iids, (0,))
         self.assertEqual(after_iids, (0,))
 
-        before_gens = get_generations(before_stats)
-        after_gens = get_generations(after_stats)
+        self.assertEqual(get_generations(before_stats), (0, 1, 2))
+        self.assertEqual(get_generations(after_stats), (0, 1, 2))
 
-        self.assertEqual(before_gens, (0, 1, 2))
-        self.assertEqual(after_gens, (0, 1, 2))
-
-        iid = 0 # main interpreter ID
-        before_last_items = (get_last_item(before_stats, 0, iid),
-                             get_last_item(before_stats, 1, iid),
-                             get_last_item(before_stats, 2, iid))
+        before_last_items = (get_last_item(before_stats, 0, self._main_iid),
+                             get_last_item(before_stats, 1, self._main_iid),
+                             get_last_item(before_stats, 2, self._main_iid))
 
-        after_last_items = (get_last_item(after_stats, 0, iid),
-                            get_last_item(after_stats, 1, iid),
-                            get_last_item(after_stats, 2, iid))
+        after_last_items = (get_last_item(after_stats, 0, self._main_iid),
+                            get_last_item(after_stats, 1, self._main_iid),
+                            get_last_item(after_stats, 2, self._main_iid))
 
         for before, after in zip(before_last_items, after_last_items):
             self._check_gc_state(before, after)
 
-    def test_get_gc_stats_for_all_interpreters(self):
-        interpreters = import_helper.import_module("concurrent.interpreters")
-        interp = interpreters.create()
+    def test_get_gc_stats_for_main_interpreter(self):
+        script = textwrap.dedent(self._only_main_interpreter_script)
+        before_stats, after_stats = self._collect_gc_stats(script, False)
+
+        self._check_main_interpreter_stats(before_stats,after_stats)
 
-        self._run_in_interpreter(interp) # ensure that subinterpeter have GC stats
-        before_stats = self._run_child_process(True)
-        self._run_in_interpreter(interp) # ensure that GC stats in subinterpreter changed
-        after_stats = self._run_child_process(True)
-        interp.close()
+    def test_get_gc_stats_for_main_interpreter_if_subinterpreter_exists(self):
+        script = textwrap.dedent(self._subinterpreters_script)
+        before_stats, after_stats = self._collect_gc_stats(script, False)
+
+        self._check_main_interpreter_stats(before_stats,after_stats)
+
+    def test_get_gc_stats_for_all_interpreters(self):
+        script = textwrap.dedent(self._subinterpreters_script)
+        before_stats, after_stats = self._collect_gc_stats(script, True)
 
         before_iids = get_interpreter_identifiers(before_stats)
         after_iids = get_interpreter_identifiers(after_stats)
 
-        self.assertEqual(before_iids, (0, interp.id))
-        self.assertEqual(after_iids, (0, interp.id))
+        self.assertGreater(len(before_iids), 1)
+        self.assertGreater(len(after_iids), 1)
+        self.assertEqual(before_iids, after_iids)
 
         before_gens = get_generations(before_stats)
         after_gens = get_generations(after_stats)
@@ -203,37 +210,3 @@ def test_get_gc_stats_for_all_interpreters(self):
 
                 for before, after in zip(before_last_items, after_last_items):
                     self._check_gc_state(before, after)
-
-    def test_get_gc_stats_for_main_interpreter_if_subinterpreter_exists(self):
-        interpreters = import_helper.import_module("concurrent.interpreters")
-        interp = interpreters.create()
-
-        self._run_in_interpreter(interp) # ensure that subinterpeter have GC stats
-        before_stats = self._run_child_process(False)
-        self._run_in_interpreter(interp) # ensure that GC stats in subinterpreter changed
-        after_stats = self._run_child_process(False)
-        interp.close()
-
-        before_iids = get_interpreter_identifiers(before_stats)
-        after_iids = get_interpreter_identifiers(after_stats)
-
-        self.assertEqual(before_iids, (0, ))
-        self.assertEqual(after_iids, (0, ))
-
-        before_gens = get_generations(before_stats)
-        after_gens = get_generations(after_stats)
-
-        self.assertEqual(before_gens, (0, 1, 2))
-        self.assertEqual(after_gens, (0, 1, 2))
-
-        iid = 0 # main interpreter ID
-        before_last_items = (get_last_item(before_stats, 0, iid),
-                             get_last_item(before_stats, 1, iid),
-                             get_last_item(before_stats, 2, iid))
-
-        after_last_items = (get_last_item(after_stats, 0, iid),
-                            get_last_item(after_stats, 1, iid),
-                            get_last_item(after_stats, 2, iid))
-
-        for before, after in zip(before_last_items, after_last_items):
-            self._check_gc_state(before, after)
