def test___del__(self):
     self.assertFalse(self.info_exists("varname"))
     v = Variable(self.root, "sample string", "varname")
     self.assertTrue(self.info_exists("varname"))
     del v
     gc_collect()
     self.assertFalse(self.info_exists("varname"))

    def test_leak_fast_process_del_killed(self):
        # Issue #12650: on Unix, if Popen.__del__() was called before the
        # process exited, and the process got killed by a signal, it would never
        # be removed from subprocess._active, which triggered a FD and memory
        # leak.
        # spawn a Popen, delete its reference and kill it
        p = subprocess.Popen([sys.executable, "-c",
                              'import time;'
                              'time.sleep(3)'],
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)
        self.addCleanup(p.stdout.close)
        self.addCleanup(p.stderr.close)
        ident = id(p)
        pid = p.pid
        del p
        test_support.gc_collect()
        os.kill(pid, signal.SIGKILL)
        # check that p is in the active processes list
        self.assertIn(ident, [id(o) for o in subprocess._active])

        # let some time for the process to exit, and create a new Popen: this
        # should trigger the wait() of p
        time.sleep(0.2)
        with self.assertRaises(EnvironmentError) as c:
            with subprocess.Popen(['nonexisting_i_hope'],
                                  stdout=subprocess.PIPE,
                                  stderr=subprocess.PIPE) as proc:
                pass
        # p should have been wait()ed on, and removed from the _active list
        self.assertRaises(OSError, os.waitpid, pid, 0)
        self.assertNotIn(ident, [id(o) for o in subprocess._active])

    def test__count(self):
        # Test the _count() function.
        orig = thread._count()
        mut = thread.allocate_lock()
        mut.acquire()
        started = []

        def task():
            started.append(None)
            mut.acquire()
            mut.release()

        thread.start_new_thread(task, ())
        while not started:
            time.sleep(0.01)
        self.assertEqual(thread._count(), orig + 1)
        # Allow the task to finish.
        mut.release()
        # The only reliable way to be sure that the thread ended from the
        # interpreter's point of view is to wait for the function object to be
        # destroyed.
        done = []
        wr = weakref.ref(task, lambda _: done.append(None))
        del task
        while not done:
            time.sleep(0.01)
            test_support.gc_collect()
        self.assertEqual(thread._count(), orig)

    def test_callbacks_on_callback(self):
        # Set up weakref callbacks *on* weakref callbacks.
        alist = []
        def safe_callback(ignore):
            alist.append("safe_callback called")

        class C(object):
            def cb(self, ignore):
                alist.append("cb called")

        c, d = C(), C()
        c.other = d
        d.other = c
        callback = c.cb
        c.wr = weakref.ref(d, callback)     # this won't trigger
        d.wr = weakref.ref(callback, d.cb)  # ditto
        external_wr = weakref.ref(callback, safe_callback)  # but this will
        self.assertTrue(external_wr() is callback)

        # The weakrefs attached to c and d should get cleared, so that
        # C.cb is never called.  But external_wr isn't part of the cyclic
        # trash, and no cyclic trash is reachable from it, so safe_callback
        # should get invoked when the bound method object callback (c.cb)
        # -- which is itself a callback, and also part of the cyclic trash --
        # gets reclaimed at the end of gc.

        del callback, c, d, C
        self.assertEqual(alist, [])  # del isn't enough to clean up cycles
        gc_collect()
        self.assertEqual(alist, ["safe_callback called"])
        self.assertEqual(external_wr(), None)

        del alist[:]
        gc_collect()
        self.assertEqual(alist, [])

    def test_class_to_test_weakness(self):
        # regrtest for bug 1522, adapted from test code submitted by Matt Brinkley

        # We wish to demonstrate that the proxy created by a Python class and
        # its PyType are GC'd when no longer in use, and therefore that the
        # Jython type system does not keep a PyType alive gratuitously. We do
        # this by holding weak references, then checking they're dead.

        # Get to steady state by creating >1 Dog and then GC-ing homeless ones.
        battersea = []
        for i in range(2):
            battersea.extend(run_script(DOG, ('Dog', 'dog', 'breed')))
        test_support.gc_collect()

        # This is the steady-state, GC number of Dog objects alive after GC:
        start_size = len(survivors(battersea)) # probably = 1

        # Add more Dogs and GC the homeless ones again.
        for i in range(5):
            battersea.extend(run_script(DOG, ('Dog', 'dog', 'breed')))
        test_support.gc_collect()
        #print "\nDogs home  =", battersea
        #print "\nDogs alive =", survivors(battersea)

        # Post-GC number of Dogs should be as before
        self.assertEqual(start_size, len(survivors(battersea)))

 def test_many(self):
     # mktemp can choose many usable file names (stochastic)
     extant = range(TEST_FILES)
     for i in extant:
         extant[i] = self.do_create(pre="aa")
     del extant
     test_support.gc_collect()

 def test_init(self):
     # Issue 3634
     # <weakref to class>.__init__() doesn't check errors correctly
     r = weakref.ref(Exception)
     self.assertRaises(TypeError, r.__init__, 0, 0, 0, 0, 0)
     # No exception should be raised here
     gc_collect()

 def test_weakref(self):
     f = self.thetype(int, base=16)
     p = proxy(f)
     self.assertEqual(f.func, p.func)
     f = None
     test_support.gc_collect()
     self.assertRaises(ReferenceError, getattr, p, 'func')

 def _test_close_open_io(self, io_func, nb_workers=5):
     def worker():
         funcs = itertools.cycle((
             lambda: io_func(),
             lambda: self._close_and_reopen_file(),
         ))
         for f in funcs:
             if not self.do_continue:
                 break
             try:
                 f()
             except (IOError, ValueError):
                 pass
     self._create_file()
     self._run_workers(worker, nb_workers)
     # make sure that all files can be closed now
     del self.all_files
     gc_collect()
     if test_support.verbose:
         # Useful verbose statistics when tuning this test to take
         # less time to run but still ensuring that its still useful.
         #
         # the percent of close calls that raised an error
         percent = 100. - 100.*self.close_success_count/self.close_count
         print self.close_count, ('%.4f ' % percent),

 def test_weak_keys(self):
     #
     #  This exercises d.copy(), d.items(), d[] = v, d[], del d[],
     #  len(d), in d.
     #
     dict, objects = self.make_weak_keyed_dict()
     for o in objects:
         self.assertTrue(weakref.getweakrefcount(o) == 1,
                      "wrong number of weak references to %r!" % o)
         self.assertTrue(o.arg is dict[o],
                      "wrong object returned by weak dict!")
     items1 = dict.items()
     items2 = dict.copy().items()
     self.assertTrue(set(items1) == set(items2),
                  "cloning of weak-keyed dictionary did not work!")
     del items1, items2
     self.assertTrue(len(dict) == self.COUNT)
     del objects[0]
     gc_collect()
     self.assertTrue(len(dict) == (self.COUNT - 1),
                  "deleting object did not cause dictionary update")
     del objects, o
     gc_collect()
     self.assertTrue(len(dict) == 0,
                  "deleting the keys did not clear the dictionary")
     o = Object(42)
     dict[o] = "What is the meaning of the universe?"
     self.assertIn(o, dict)
     self.assertNotIn(34, dict)

 def test_weakref(self):
     s = array.array(self.typecode, self.example)
     p = proxy(s)
     self.assertEqual(p.tostring(), s.tostring())
     s = None
     test_support.gc_collect()
     self.assertRaises(ReferenceError, len, p)

 def tearDown(self):
     test_support.gc_collect()
     for name in self.files:
         os.unlink(name)
     for i in xrange(2):
         if os.path.exists(test_support.TESTFN):
             os.rmdir(test_support.TESTFN)

    def testTruncateOnWindows(self):
        def bug801631():
            # SF bug <http://www.python.org/sf/801631>
            # "file.truncate fault on windows"
            f = _FileIO(TESTFN, 'w')
            f.write(bytes(range(11)))
            f.close()

            f = _FileIO(TESTFN,'r+')
            data = f.read(5)
            if data != bytes(range(5)):
                self.fail("Read on file opened for update failed %r" % data)
            if f.tell() != 5:
                self.fail("File pos after read wrong %d" % f.tell())

            f.truncate()
            if f.tell() != 5:
                self.fail("File pos after ftruncate wrong %d" % f.tell())

            f.close()
            size = os.path.getsize(TESTFN)
            if size != 5:
                self.fail("File size after ftruncate wrong %d" % size)

        try:
            bug801631()
        finally:
            gc_collect()
            os.unlink(TESTFN)

    def check_gc_during_creation(self, makeref):
        if test_support.check_impl_detail():
            import gc
            thresholds = gc.get_threshold()
            gc.set_threshold(1, 1, 1)
        gc_collect()
        class A:
            pass

        def callback(*args):
            pass

        referenced = A()

        a = A()
        a.a = a
        a.wr = makeref(referenced)

        try:
            # now make sure the object and the ref get labeled as
            # cyclic trash:
            a = A()
            weakref.ref(referenced, callback)

        finally:
            if test_support.check_impl_detail():
                gc.set_threshold(*thresholds)

 def test_weakref(self):
     d = deque('gallahad')
     p = weakref.proxy(d)
     self.assertEqual(str(p), str(d))
     d = None
     test_support.gc_collect()
     self.assertRaises(ReferenceError, str, p)

    def tearDown(self):
        # wait on the server thread to terminate
        test_support.gc_collect() # to close the active connections
        self.evt.wait(10)

        # disable traceback reporting
        SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = False

    def test_weak_destroy_and_mutate_while_iterating(self):
        # Issue #7105: iterators shouldn't crash when a key is implicitly removed
        items = [SomeClass(c) for c in string.ascii_letters]
        s = WeakSet(items)
        @contextlib.contextmanager
        def testcontext():
            try:
                it = iter(s)
                next(it)
                # Schedule an item for removal and recreate it
                u = SomeClass(str(items.pop()))
                test_support.gc_collect()      # just in case
                yield u
            finally:
                it = None           # should commit all removals

        test_support.gc_collect()

        with testcontext() as u:
            self.assertNotIn(u, s)
        with testcontext() as u:
            self.assertRaises(KeyError, s.remove, u)
        self.assertNotIn(u, s)
        with testcontext() as u:
            s.add(u)
        self.assertIn(u, s)
        t = s.copy()
        with testcontext() as u:
            s.update(t)
        self.assertEqual(len(s), len(t))
        with testcontext() as u:
            s.clear()
        self.assertEqual(len(s), 0)

    def test_callback_in_cycle_4(self):
        # Like test_callback_in_cycle_3, except c2 and c1 have different
        # classes.  c2's class (C) isn't reachable from c1 then, so protecting
        # objects reachable from the dying object (c1) isn't enough to stop
        # c2's class (C) from getting tp_clear'ed before c2.cb is invoked.
        # The result was a segfault (C.__mro__ was NULL when the callback
        # tried to look up self.me).

        class C(object):
            def cb(self, ignore):
                self.me
                self.c1
                self.wr

        class D:
            pass

        c1, c2 = D(), C()

        c2.me = c2
        c2.c1 = c1
        c2.wr = weakref.ref(c1, c2.cb)

        del c1, c2, C, D
        gc_collect()

    def test_callback_in_cycle_2(self):
        # This is just like test_callback_in_cycle_1, except that II is an
        # old-style class.  The symptom is different then:  an instance of an
        # old-style class looks in its own __dict__ first.  'J' happens to
        # get cleared from I.__dict__ before 'wr', and 'J' was never in II's
        # __dict__, so the attribute isn't found.  The difference is that
        # the old-style II doesn't have a NULL __mro__ (it doesn't have any
        # __mro__), so no segfault occurs.  Instead it got:
        #    test_callback_in_cycle_2 (__main__.ReferencesTestCase) ...
        #    Exception exceptions.AttributeError:
        #   "II instance has no attribute 'J'" in <bound method II.acallback
        #       of <?.II instance at 0x00B9B4B8>> ignored

        class J(object):
            pass

        class II:
            def acallback(self, ignore):
                self.J

        I = II()
        I.J = J
        I.wr = weakref.ref(J, I.acallback)

        del I, J, II
        gc_collect()

def cleanup_test_droppings(testname, verbose):
    import shutil

    # Try to clean up junk commonly left behind.  While tests shouldn't leave
    # any files or directories behind, when a test fails that can be tedious
    # for it to arrange.  The consequences can be especially nasty on Windows,
    # since if a test leaves a file open, it cannot be deleted by name (while
    # there's nothing we can do about that here either, we can display the
    # name of the offending test, which is a real help).
    for name in (test_support.TESTFN,
                 "db_home",
                ):
        if not os.path.exists(name):
            continue

        # work around tests depending on refcounting files,
        # but this doesn't work with respect to Windows
        test_support.gc_collect()

        if os.path.isdir(name):
            kind, nuker = "directory", shutil.rmtree
        elif os.path.isfile(name):
            kind, nuker = "file", os.unlink
        else:
            raise SystemError("os.path says %r exists but is neither "
                              "directory nor file" % name)

        if verbose:
            print "%r left behind %s %r" % (testname, kind, name)
        try:
            nuker(name)
        except Exception, msg:
            print >> sys.stderr, ("%r left behind %s %r and it couldn't be "
                "removed: %s" % (testname, kind, name, msg))