Run a test multiple times, looking for reference leaks. Returns: False if the test didn't leak references; True if we detected refleaks.
(test_name, test_func,
hunt_refleak: HuntRefleak,
quiet: bool)
| 50 | |
| 51 | |
| 52 | def runtest_refleak(test_name, test_func, |
| 53 | hunt_refleak: HuntRefleak, |
| 54 | quiet: bool): |
| 55 | """Run a test multiple times, looking for reference leaks. |
| 56 | |
| 57 | Returns: |
| 58 | False if the test didn't leak references; True if we detected refleaks. |
| 59 | """ |
| 60 | # This code is hackish and inelegant, but it seems to do the job. |
| 61 | import copyreg |
| 62 | import collections.abc |
| 63 | |
| 64 | if not hasattr(sys, 'gettotalrefcount'): |
| 65 | raise Exception("Tracking reference leaks requires a debug build " |
| 66 | "of Python") |
| 67 | |
| 68 | # Avoid false positives due to various caches |
| 69 | # filling slowly with random data: |
| 70 | warm_caches() |
| 71 | |
| 72 | # Save current values for dash_R_cleanup() to restore. |
| 73 | fs = warnings.filters[:] |
| 74 | ps = copyreg.dispatch_table.copy() |
| 75 | pic = sys.path_importer_cache.copy() |
| 76 | zdc: dict[str, Any] | None |
| 77 | # Linecache holds a cache with the source of interactive code snippets |
| 78 | # (e.g. code typed in the REPL). This cache is not cleared by |
| 79 | # linecache.clearcache(). We need to save and restore it to avoid false |
| 80 | # positives. |
| 81 | linecache_data = linecache.cache.copy(), linecache._interactive_cache.copy() # type: ignore[attr-defined] |
| 82 | try: |
| 83 | import zipimport |
| 84 | except ImportError: |
| 85 | zdc = None # Run unmodified on platforms without zipimport support |
| 86 | else: |
| 87 | # private attribute that mypy doesn't know about: |
| 88 | zdc = zipimport._zip_directory_cache.copy() # type: ignore[attr-defined] |
| 89 | abcs = {} |
| 90 | for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]: |
| 91 | if not isabstract(abc): |
| 92 | continue |
| 93 | for obj in abc.__subclasses__() + [abc]: |
| 94 | abcs[obj] = _get_dump(obj)[0] |
| 95 | |
| 96 | # bpo-31217: Integer pool to get a single integer object for the same |
| 97 | # value. The pool is used to prevent false alarm when checking for memory |
| 98 | # block leaks. Fill the pool with values in -1000..1000 which are the most |
| 99 | # common (reference, memory block, file descriptor) differences. |
| 100 | int_pool = {value: value for value in range(-1000, 1000)} |
| 101 | def get_pooled_int(value): |
| 102 | return int_pool.setdefault(value, value) |
| 103 | |
| 104 | warmups = hunt_refleak.warmups |
| 105 | runs = hunt_refleak.runs |
| 106 | filename = hunt_refleak.filename |
| 107 | repcount = warmups + runs |
| 108 | |
| 109 | # Pre-allocate to ensure that the loop doesn't allocate anything new |
no test coverage detected