reimplement getdump for purepython implementation of the abc module libpyabc py run a test multiple times looking for reference leaks returns false if the test didn t leak references true if we detected refleaks this code is hackish and inelegant but it seems to do the job avoid false positives due to various caches filling slowly with random data save current values for dashrcleanup to restore private attribute that mypy doesn t know about bpo31217 integer pool to get a single integer object for the same value the pool is used to prevent false alarm when checking for memory block leaks fill the pool with values in 1000 1000 which are the most common reference memory block file descriptor differences preallocate to ensure that the loop doesn t allocate anything new initialize variables to make pyflakes quiet read memory statistics immediately after the garbage collection also readjust the reference counts and alloc blocks by ignoring any strings that might have been interned during testfunc these strings will be deallocated at runtime shutdown these checkers return false on success true on failure checker for reference counters and memory blocks bpo30776 try to ignore false positives 3 0 0 0 1 0 8 8 1 expected leaks 5 5 6 10 1 1 ignore warmup runs restore some original values clear abc registries restoring previously saved abc registries ignore deprecation warning for collections abc bytestring clear caches clear type cache at the end previous function calls can modify types char cache unicode cache int cache reimplement _get_dump for pure python implementation of the abc module lib _py_abc py run a test multiple times looking for reference leaks returns false if the test didn t leak references true if we detected refleaks this code is hackish and inelegant but it seems to do the job avoid false positives due to various caches filling slowly with random data save current values for dash_r_cleanup to restore run unmodified on platforms without zipimport support private attribute that mypy doesn t know about type ignore attr defined bpo 31217 integer pool to get a single integer object for the same value the pool is used to prevent false alarm when checking for memory block leaks fill the pool with values in 1000 1000 which are the most common reference memory block file descriptor differences pre allocate to ensure that the loop doesn t allocate anything new initialize variables to make pyflakes quiet read memory statistics immediately after the garbage collection also readjust the reference counts and alloc blocks by ignoring any strings that might have been interned during test_func these strings will be deallocated at runtime shutdown these checkers return false on success true on failure checker for reference counters and memory blocks bpo 30776 try to ignore false positives 3 0 0 0 1 0 8 8 1 expected leaks 5 5 6 10 1 1 ignore warmup runs restore some original values run unmodified on platforms without zipimport support clear abc registries restoring previously saved abc registries ignore deprecation warning for collections abc bytestring clear caches clear type cache at the end previous function calls can modify types char cache unicode cache int cache
import sys
import warnings
from inspect import isabstract
from typing import Any

from test import support
from test.support import os_helper

from .runtests import HuntRefleak
from .utils import clear_caches

try:
    from _abc import _get_dump
except ImportError:
    import weakref

    def _get_dump(cls):
        
        
        registry_weakrefs = set(weakref.ref(obj) for obj in cls._abc_registry)
        return (registry_weakrefs, cls._abc_cache,
                cls._abc_negative_cache, cls._abc_negative_cache_version)


def runtest_refleak(test_name, test_func,
                    hunt_refleak: HuntRefleak,
                    quiet: bool):
    
    
    import copyreg
    import collections.abc

    if not hasattr(sys, 'gettotalrefcount'):
        raise Exception("Tracking reference leaks requires a debug build "
                        "of Python")

    
    
    warm_caches()

    
    fs = warnings.filters[:]
    ps = copyreg.dispatch_table.copy()
    pic = sys.path_importer_cache.copy()
    zdc: dict[str, Any] | None
    try:
        import zipimport
    except ImportError:
        zdc = None 
    else:
        
        zdc = zipimport._zip_directory_cache.copy()  
    abcs = {}
    for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]:
        if not isabstract(abc):
            continue
        for obj in abc.__subclasses__() + [abc]:
            abcs[obj] = _get_dump(obj)[0]

    
    
    
    
    int_pool = {value: value for value in range(-1000, 1000)}
    def get_pooled_int(value):
        return int_pool.setdefault(value, value)

    warmups = hunt_refleak.warmups
    runs = hunt_refleak.runs
    filename = hunt_refleak.filename
    repcount = warmups + runs

    
    rep_range = list(range(repcount))
    rc_deltas = [0] * repcount
    alloc_deltas = [0] * repcount
    fd_deltas = [0] * repcount
    getallocatedblocks = sys.getallocatedblocks
    gettotalrefcount = sys.gettotalrefcount
    getunicodeinternedsize = sys.getunicodeinternedsize
    fd_count = os_helper.fd_count
    
    rc_before = alloc_before = fd_before = interned_before = 0

    if not quiet:
        print("beginning", repcount, "repetitions", file=sys.stderr)
        print(("1234567890"*(repcount//10 + 1))[:repcount], file=sys.stderr,
              flush=True)

    results = None
    dash_R_cleanup(fs, ps, pic, zdc, abcs)
    support.gc_collect()

    for i in rep_range:
        results = test_func()

        dash_R_cleanup(fs, ps, pic, zdc, abcs)
        support.gc_collect()

        
        
        
        
        interned_after = getunicodeinternedsize()
        alloc_after = getallocatedblocks() - interned_after
        rc_after = gettotalrefcount() - interned_after * 2
        fd_after = fd_count()

        if not quiet:
            print('.', end='', file=sys.stderr, flush=True)

        rc_deltas[i] = get_pooled_int(rc_after - rc_before)
        alloc_deltas[i] = get_pooled_int(alloc_after - alloc_before)
        fd_deltas[i] = get_pooled_int(fd_after - fd_before)

        alloc_before = alloc_after
        rc_before = rc_after
        fd_before = fd_after
        interned_before = interned_after

    if not quiet:
        print(file=sys.stderr)

    
    def check_rc_deltas(deltas):
        
        
        
        
        
        
        
        
        
        
        
        
        return all(delta >= 1 for delta in deltas)

    def check_fd_deltas(deltas):
        return any(deltas)

    failed = False
    for deltas, item_name, checker in [
        (rc_deltas, 'references', check_rc_deltas),
        (alloc_deltas, 'memory blocks', check_rc_deltas),
        (fd_deltas, 'file descriptors', check_fd_deltas)
    ]:
        
        deltas = deltas[warmups:]
        if checker(deltas):
            msg = '%s leaked %s %s, sum=%s' % (
                test_name, deltas, item_name, sum(deltas))
            print(msg, file=sys.stderr, flush=True)
            with open(filename, "a", encoding="utf-8") as refrep:
                print(msg, file=refrep)
                refrep.flush()
            failed = True
    return (failed, results)


def dash_R_cleanup(fs, ps, pic, zdc, abcs):
    import copyreg
    import collections.abc

    
    warnings.filters[:] = fs
    copyreg.dispatch_table.clear()
    copyreg.dispatch_table.update(ps)
    sys.path_importer_cache.clear()
    sys.path_importer_cache.update(pic)
    try:
        import zipimport
    except ImportError:
        pass 
    else:
        zipimport._zip_directory_cache.clear()
        zipimport._zip_directory_cache.update(zdc)

    
    
    abs_classes = [getattr(collections.abc, a) for a in collections.abc.__all__]
    abs_classes = filter(isabstract, abs_classes)
    for abc in abs_classes:
        for obj in abc.__subclasses__() + [abc]:
            for ref in abcs.get(obj, set()):
                if ref() is not None:
                    obj.register(ref())
            obj._abc_caches_clear()

    
    clear_caches()

    
    sys._clear_type_cache()


def warm_caches():
    
    s = bytes(range(256))
    for i in range(256):
        s[i:i+1]
    
    [chr(i) for i in range(256)]
    
    list(range(-5, 257))