cls._abc_negative_cache, cls._abc_negative_cache_version)

def dash_R(the_module, test, indirect_test, huntrleaks):

def dash_R(ns, the_module, test_name, test_func):

"""Run a test multiple times, looking for reference leaks.

Returns:

raise Exception("Tracking reference leaks requires a debug build "

"of Python")

# Avoid false positives due to various caches

# filling slowly with random data:

warm_caches()

# Save current values for dash_R_cleanup() to restore.

fs = warnings.filters[:]

ps = copyreg.dispatch_table.copy()

def get_pooled_int(value):

return int_pool.setdefault(value, value)

nwarmup, ntracked, fname = huntrleaks

nwarmup, ntracked, fname = ns.huntrleaks

fname = os.path.join(support.SAVEDCWD, fname)

repcount = nwarmup + ntracked

# Pre-allocate to ensure that the loop doesn't allocate anything new

rep_range = list(range(repcount))

rc_deltas = [0] * repcount

alloc_deltas = [0] * repcount

fd_deltas = [0] * repcount

getallocatedblocks = sys.getallocatedblocks

gettotalrefcount = sys.gettotalrefcount

fd_count = support.fd_count

print("beginning", repcount, "repetitions", file=sys.stderr)

print(("1234567890"*(repcount//10 + 1))[:repcount], file=sys.stderr,

flush=True)

# initialize variables to make pyflakes quiet

rc_before = alloc_before = fd_before = 0

for i in range(repcount):

indirect_test()

alloc_after, rc_after, fd_after = dash_R_cleanup(fs, ps, pic, zdc,

abcs)

print('.', end='', file=sys.stderr, flush=True)

if i >= nwarmup:

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)

if not ns.quiet:

print("beginning", repcount, "repetitions", file=sys.stderr)

print(("1234567890"*(repcount//10 + 1))[:repcount], file=sys.stderr,

flush=True)

for i in rep_range:

test_func()

dash_R_cleanup(fs, ps, pic, zdc, abcs)

# Collect cyclic trash and read memory statistics immediately after.

support.gc_collect()

alloc_after = getallocatedblocks()

rc_after = gettotalrefcount()

fd_after = fd_count()

if not ns.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

print(file=sys.stderr)

if not ns.quiet:

print(file=sys.stderr)

# These checkers return False on success, True on failure

def check_rc_deltas(deltas):

deltas = deltas[nwarmup:]

if checker(deltas):

msg = '%s leaked %s %s, sum=%s' % (

test, deltas, item_name, sum(deltas))

test_name, deltas, item_name, sum(deltas))

print(msg, file=sys.stderr, flush=True)

with open(fname, "a") as refrep:

print(msg, file=refrep)

def dash_R_cleanup(fs, ps, pic, zdc, abcs):

import gc, copyreg

import copyreg

import collections.abc

# Restore some original values.

clear_caches()

# Collect cyclic trash and read memory statistics immediately after.

func1 = sys.getallocatedblocks

func2 = sys.gettotalrefcount

gc.collect()

return func1(), func2(), support.fd_count()

def clear_caches():

import gc

# Clear the warnings registry, so they can be displayed again

for mod in sys.modules.values():

if hasattr(mod, '__warningregistry__'):

for f in typing._cleanups:

f()

gc.collect()

support.gc_collect()

def warm_caches():