Refine tests

Author: sergey-miryanov

Lib/test/test_get_gc_stats.py

@@ -1,3 +1,4 @@
+import os
 import textwrap
 import time
 import unittest
@@ -47,30 +48,7 @@ class TestGetGCStats(unittest.TestCase):
     @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 = '''
+        cls._script = '''
             import concurrent.interpreters as interpreters
             import gc
             import time
@@ -83,8 +61,9 @@ def setUpClass(cls):
                 gc.collect(2)
             """
 
-            interp = interpreters.create()
-            interp.exec(source)
+            if {0}:
+                interp = interpreters.create()
+                interp.exec(source)
 
             gc.collect(0)
             gc.collect(1)
@@ -100,7 +79,8 @@ def setUpClass(cls):
                 objects.append(object())
 
                 time.sleep(0.1)
-                interp.exec(source)
+                if {0}:
+                    interp.exec(source)
                 gc.collect(0)
                 gc.collect(1)
                 gc.collect(2)
@@ -124,7 +104,7 @@ def _collect_gc_stats(self, script:str, all_interpreters:bool):
 
         return before_stats, after_stats
 
-    def _check_gc_state(self, before, after):
+    def _check_gc_stats(self, before, after):
         self.assertIsNotNone(before)
         self.assertIsNotNone(after)
 
@@ -148,41 +128,55 @@ def _check_gc_state(self, before, after):
                                     before["objects_not_transitively_reachable"],
                                     (before, after))
 
-    def _check_main_interpreter_stats(self, before_stats, after_stats):
+    def _check_interpreter_gc_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,))
+        self.assertEqual(before_iids, after_iids)
 
         self.assertEqual(get_generations(before_stats), (0, 1, 2))
         self.assertEqual(get_generations(after_stats), (0, 1, 2))
 
-        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))
+        for iid in after_iids:
+            with self.subTest(f"interpreter id={iid}"):
+                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, self._main_iid),
-                            get_last_item(after_stats, 1, self._main_iid),
-                            get_last_item(after_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))
 
-        for before, after in zip(before_last_items, after_last_items):
-            self._check_gc_state(before, after)
+                for before, after in zip(before_last_items, after_last_items):
+                    self._check_gc_stats(before, after)
+
+    def test_get_gc_stats_fields(self):
+        keys = sorted(("gen", "iid", "ts_start", "ts_stop", "heap_size",
+                "work_to_do", "collections", "object_visits",
+                "collected", "uncollectable", "candidates",
+                "objects_transitively_reachable",
+                "objects_not_transitively_reachable",
+                "duration"))
+        stats = _remote_debugging.get_gc_stats(os.getpid(), all_interpreters=False)
+        self.assertIsInstance(stats, list)
+        for item in stats:
+            self.assertIsInstance(item, dict)
+            self.assertEqual(sorted(item.keys()), keys)
 
     def test_get_gc_stats_for_main_interpreter(self):
-        script = textwrap.dedent(self._only_main_interpreter_script)
+        script = textwrap.dedent(self._script.format(False))
         before_stats, after_stats = self._collect_gc_stats(script, False)
 
-        self._check_main_interpreter_stats(before_stats,after_stats)
+        self._check_interpreter_gc_stats(before_stats,after_stats)
 
     def test_get_gc_stats_for_main_interpreter_if_subinterpreter_exists(self):
-        script = textwrap.dedent(self._subinterpreters_script)
+        script = textwrap.dedent(self._script.format(True))
         before_stats, after_stats = self._collect_gc_stats(script, False)
 
-        self._check_main_interpreter_stats(before_stats,after_stats)
+        self._check_interpreter_gc_stats(before_stats, after_stats)
 
     def test_get_gc_stats_for_all_interpreters(self):
-        script = textwrap.dedent(self._subinterpreters_script)
+        script = textwrap.dedent(self._script.format(True))
         before_stats, after_stats = self._collect_gc_stats(script, True)
 
         before_iids = get_interpreter_identifiers(before_stats)
@@ -192,21 +186,4 @@ def test_get_gc_stats_for_all_interpreters(self):
         self.assertGreater(len(after_iids), 1)
         self.assertEqual(before_iids, after_iids)
 
-        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))
-
-        for iid in after_iids:
-            with self.subTest(f"iid={iid}"):
-                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)
+        self._check_interpreter_gc_stats(before_stats, after_stats)