Raise AssertionError if the wrapped code creates garbage with cycles.
()
| 63 | |
| 64 | @contextlib.contextmanager |
| 65 | def assert_no_cycle_garbage(): |
| 66 | """Raise AssertionError if the wrapped code creates garbage with cycles.""" |
| 67 | gc.disable() |
| 68 | gc.collect() |
| 69 | gc.set_debug(gc.DEBUG_STATS | gc.DEBUG_SAVEALL) |
| 70 | yield |
| 71 | try: |
| 72 | # We have DEBUG_STATS on which causes gc.collect to write to stderr. |
| 73 | # Capture the output instead of spamming the logs on passing runs. |
| 74 | f = io.StringIO() |
| 75 | old_stderr = sys.stderr |
| 76 | sys.stderr = f |
| 77 | try: |
| 78 | gc.collect() |
| 79 | finally: |
| 80 | sys.stderr = old_stderr |
| 81 | garbage = gc.garbage[:] |
| 82 | # Must clear gc.garbage (the same object, not just replacing it with a |
| 83 | # new list) to avoid warnings at shutdown. |
| 84 | gc.garbage[:] = [] |
| 85 | if len(garbage) == 0: |
| 86 | return |
| 87 | for circular in find_circular_references(garbage): |
| 88 | f.write("\n==========\n Circular \n==========") |
| 89 | for item in circular: |
| 90 | f.write(f"\n {repr(item)}") |
| 91 | for item in circular: |
| 92 | if isinstance(item, types.FrameType): |
| 93 | f.write(f"\nLocals: {item.f_locals}") |
| 94 | f.write(f"\nTraceback: {repr(item)}") |
| 95 | traceback.print_stack(item) |
| 96 | del garbage |
| 97 | raise AssertionError(f.getvalue()) |
| 98 | finally: |
| 99 | gc.set_debug(0) |
| 100 | gc.enable() |
| 101 | |
| 102 | |
| 103 | # GC behavior is cpython-specific |
no test coverage detected