def test04_leak_GC(self):
     import sys
     a = "example of private object"
     refcount = sys.getrefcount(a)
     self.obj.set_private(a)
     self.obj = None
     self.assertEqual(refcount, sys.getrefcount(a))

 def check_set_double_new(self,level=5):
     a = UserDict()
     key = 1.0
     inline_tools.inline('a[key] = 123.0;',['a','key'])
     assert_equal(sys.getrefcount(key),4) # should be 3
     assert_equal(sys.getrefcount(a[key]),2)
     assert_equal(a[key],123.0)

    def test_no_refcycle_through_target(self):
        class RunSelfFunction(object):
            def __init__(self, should_raise):
                # The links in this refcycle from Thread back to self
                # should be cleaned up when the thread completes.
                self.should_raise = should_raise
                self.thread = threading.Thread(target=self._run, args=(self,), kwargs={"yet_another": self})
                self.thread.start()

            def _run(self, other_ref, yet_another):
                if self.should_raise:
                    raise SystemExit

        cyclic_object = RunSelfFunction(should_raise=False)
        weak_cyclic_object = weakref.ref(cyclic_object)
        cyclic_object.thread.join()
        del cyclic_object
        self.assertIsNone(
            weak_cyclic_object(), msg=("%d references still around" % sys.getrefcount(weak_cyclic_object()))
        )

        raising_cyclic_object = RunSelfFunction(should_raise=True)
        weak_raising_cyclic_object = weakref.ref(raising_cyclic_object)
        raising_cyclic_object.thread.join()
        del raising_cyclic_object
        self.assertIsNone(
            weak_raising_cyclic_object(),
            msg=("%d references still around" % sys.getrefcount(weak_raising_cyclic_object())),
        )

    def test_memoize_method(self):
        class foo(object):
            def __init__(self):
                self._count = 0

            @memoize
            def wrapped(self, a, b):
                self._count += 1
                return a + b

        instance = foo()
        refcount = sys.getrefcount(instance)
        self.assertEqual(instance._count, 0)
        self.assertEqual(instance.wrapped(1, 1), 2)
        self.assertEqual(instance._count, 1)
        self.assertEqual(instance.wrapped(1, 1), 2)
        self.assertEqual(instance._count, 1)
        self.assertEqual(instance.wrapped(2, 1), 3)
        self.assertEqual(instance._count, 2)
        self.assertEqual(instance.wrapped(1, 2), 3)
        self.assertEqual(instance._count, 3)
        self.assertEqual(instance.wrapped(1, 2), 3)
        self.assertEqual(instance._count, 3)
        self.assertEqual(instance.wrapped(1, 1), 2)
        self.assertEqual(instance._count, 3)

        # Memoization of methods is expected to not keep references to
        # instances, so the refcount shouldn't have changed after executing the
        # memoized method.
        self.assertEqual(refcount, sys.getrefcount(instance))

 def test_set_complex(self):
     a = UserDict()
     key = 1+1j
     inline_tools.inline("a[key] = 1234;",['a','key'])
     assert_equal(sys.getrefcount(key),4) # should be 3
     assert_equal(sys.getrefcount(a[key]),2)
     assert_equal(a[key],1234)

    def test_nrt_gen0_stop_iteration(self):
        """
        Test cleanup on StopIteration
        """
        pygen = nrt_gen0
        cgen = jit(nopython=True)(pygen)

        py_ary = np.arange(1)
        c_ary = py_ary.copy()

        py_iter = pygen(py_ary)
        c_iter = cgen(c_ary)

        py_res = next(py_iter)
        c_res = next(c_iter)

        with self.assertRaises(StopIteration):
            py_res = next(py_iter)

        with self.assertRaises(StopIteration):
            c_res = next(c_iter)

        del py_iter
        del c_iter

        np.testing.assert_equal(py_ary, c_ary)
        self.assertEqual(py_res, c_res)
        # Check reference count
        self.assertEqual(sys.getrefcount(py_ary),
                         sys.getrefcount(c_ary))

    def testBogusObject(self):
        def add(key, obj):
            self.cache[key] = obj

        nones = sys.getrefcount(None)

        key = p64(2)
        # value isn't persistent
        self.assertRaises(TypeError, add, key, 12)

        o = StubObject()
        # o._p_oid == None
        self.assertRaises(TypeError, add, key, o)

        o._p_oid = p64(3)
        self.assertRaises(ValueError, add, key, o)

        o._p_oid = key
        # o._p_jar == None
        self.assertRaises(Exception, add, key, o)

        o._p_jar = self.jar
        self.cache[key] = o
        # make sure it can be added multiple times
        self.cache[key] = o

        # same object, different keys
        self.assertRaises(ValueError, add, p64(0), o)

        if sys.gettrace() is None:
            # 'coverage' keeps track of coverage information in a data
            # structure that adds a new reference to None for each executed
            # line of code, which interferes with this test.  So check it
            # only if we're running without coverage tracing.
            self.assertEqual(sys.getrefcount(None), nones)

 def test_refs(self):
     if hasattr(sys, "getrefcount"):
         for tp in self._types:
             m = memoryview(tp(self._source))
             oldrefcount = sys.getrefcount(m)
             m[1:2]
             self.assertEqual(sys.getrefcount(m), oldrefcount)

    def test_leak_references_on(self):
        model = TesterModel()
        obj_ref = weakref.ref(model.root)
        # Initial refcount is 1 for model.root + the temporary
        self.assertEqual(sys.getrefcount(model.root), 2)

        # Iter increases by 1 do to assignment to iter.user_data
        res, it = model.do_get_iter([0])
        self.assertEqual(id(model.root), it.user_data)
        self.assertEqual(sys.getrefcount(model.root), 3)

        # Verify getting a TreeIter more then once does not further increase
        # the ref count.
        res2, it2 = model.do_get_iter([0])
        self.assertEqual(id(model.root), it2.user_data)
        self.assertEqual(sys.getrefcount(model.root), 3)

        # Deleting the iter does not decrease refcount because references
        # leak by default (they are stored in the held_refs pool)
        del it
        gc.collect()
        self.assertEqual(sys.getrefcount(model.root), 3)

        # Deleting a model should free all held references to user data
        # stored by TreeIters
        del model
        gc.collect()
        self.assertEqual(obj_ref(), None)

 def test_set_double_new(self):
     a = UserDict()
     key = 1.0
     inline_tools.inline("a[key] = 123.0;", ["a", "key"])
     assert_equal(sys.getrefcount(key), 4)  # should be 3
     assert_equal(sys.getrefcount(a[key]), 2)
     assert_equal(a[key], 123.0)

 def test_commit_refcount(self):
     commit = self.repo[COMMIT_SHA]
     start = sys.getrefcount(commit)
     tree = commit.tree
     del tree
     end = sys.getrefcount(commit)
     self.assertEqual(start, end)

 def test04_leak_GC(self):
     a = 'example of private object'
     refcount = sys.getrefcount(a)
     self.obj.set_private(a)
     self.obj = None
     self.assertEqual(refcount, sys.getrefcount(a))
     return

 def testGetSetUpd(self):
     self.assertTrue(self.snes.getUpdate() is None)
     upd = lambda snes, it: None
     refcnt = getrefcount(upd)
     self.snes.setUpdate(upd)
     self.assertEqual(getrefcount(upd), refcnt + 1)
     self.snes.setUpdate(upd)
     self.assertEqual(getrefcount(upd), refcnt + 1)
     self.snes.setUpdate(None)
     self.assertTrue(self.snes.getUpdate() is None)
     self.assertEqual(getrefcount(upd), refcnt)
     self.snes.setUpdate(upd)
     self.assertEqual(getrefcount(upd), refcnt + 1)
     upd2 = lambda snes, it: None
     refcnt2 = getrefcount(upd2)
     self.snes.setUpdate(upd2)
     self.assertEqual(getrefcount(upd),  refcnt)
     self.assertEqual(getrefcount(upd2), refcnt2 + 1)
     tmp = self.snes.getUpdate()[0]
     self.assertTrue(tmp is upd2)
     self.assertEqual(getrefcount(upd2), refcnt2 + 2)
     del tmp
     self.snes.setUpdate(None)
     self.assertTrue(self.snes.getUpdate() is None)
     self.assertEqual(getrefcount(upd2), refcnt2)

def test_to_pandas_zero_copy():
    import gc

    arr = pyarrow.from_pylist(range(10))

    for i in range(10):
        np_arr = arr.to_pandas()
        assert sys.getrefcount(np_arr) == 2
        np_arr = None  # noqa

    assert sys.getrefcount(arr) == 2

    for i in range(10):
        arr = pyarrow.from_pylist(range(10))
        np_arr = arr.to_pandas()
        arr = None
        gc.collect()

        # Ensure base is still valid

        # Because of py.test's assert inspection magic, if you put getrefcount
        # on the line being examined, it will be 1 higher than you expect
        base_refcount = sys.getrefcount(np_arr.base)
        assert base_refcount == 2
        np_arr.sum()

    def test_swap(self):

        def pyfunc(x, y, t):
            """Swap array x and y for t number of times
            """
            for i in range(t):
                x, y = y, x

            return x, y


        cfunc = nrtjit(pyfunc)

        x = np.random.random(100)
        y = np.random.random(100)

        t = 100

        initrefct = sys.getrefcount(x), sys.getrefcount(y)
        expect, got = pyfunc(x, y, t), cfunc(x, y, t)
        self.assertIsNone(got[0].base)
        self.assertIsNone(got[1].base)
        np.testing.assert_equal(expect, got)
        del expect, got
        self.assertEqual(initrefct, (sys.getrefcount(x), sys.getrefcount(y)))

    def test_noargs_with_args_not_instantiated(self):
        # calling a function that doesn't take args with args should fail.
        # Note: difference between this test add ``test_noargs_with_args``
        # below is that here Foo is not instantiated.
        def Foo():
            return "blah"

        code = """
               py::tuple args(2);
               args[0] = 1;
               args[1] = "hello";
               return_val = Foo.call(args);
               """
        try:
            first = sys.getrefcount(Foo)
            inline_tools.inline(code,['Foo'])
        except TypeError:
            second = sys.getrefcount(Foo)
            try:
                inline_tools.inline(code,['Foo'])
            except TypeError:
                third = sys.getrefcount(Foo)

        # first should == second, but the weird refcount error
        assert_equal(second,third)

    def test_refct_mt(self):
        """
        This test exercises the refct in multithreaded code
        """

        def pyfunc(n, inp):
            out = np.empty(inp.size)
            for i in range(out.size):
                out[i] = inp[i] + 1
            # Use swap to trigger many refct ops
            for i in range(n):
                out, inp = inp, out
            return out

        cfunc = nrtjit(pyfunc)
        size = 10
        input = np.arange(size, dtype=np.float)
        expected_refct = sys.getrefcount(input)
        swapct = random.randrange(1000)
        expected = pyfunc(swapct, input)
        np.testing.assert_equal(expected, cfunc(swapct, input))
        # The following checks can discover a reference count error
        del expected
        self.assertEqual(expected_refct, sys.getrefcount(input))

        workers = []
        outputs = []
        swapcts = []

        # Make wrapper to store the output
        def wrapped(n, input, out):
            out[:] = cfunc(n, input)

        # Create worker threads
        for i in range(100):
            out = np.empty(size)
            # All thread shares the same input
            swapct = random.randrange(1000)
            thread = threading.Thread(target=wrapped,
                                      args=(swapct, input, out),
                                      name="worker{0}".format(i))
            workers.append(thread)
            outputs.append(out)
            swapcts.append(swapct)

        # Launch worker threads
        for thread in workers:
            thread.start()

        # Join worker threads
        for thread in workers:
            thread.join()

        # Check result
        for swapct, out in zip(swapcts, outputs):
            np.testing.assert_equal(pyfunc(swapct, input), out)

        del outputs, workers
        # The following checks can discover a reference count error
        self.assertEqual(expected_refct, sys.getrefcount(input))

 def test03_leak_assignment(self) :
     a = "example of private object"
     refcount = sys.getrefcount(a)
     self.obj.set_private(a)
     self.assertEqual(refcount+1, sys.getrefcount(a))
     self.obj.set_private(None)
     self.assertEqual(refcount, sys.getrefcount(a))

    def test_nrt_nested_gen(self):

        def gen0(arr):
            for i in range(arr.size):
                yield arr

        def factory(gen0):
            def gen1(arr):
                out = np.zeros_like(arr)
                for x in gen0(arr):
                    out = out + x
                return out, arr

            return gen1

        py_arr = np.arange(10)
        c_arr = py_arr.copy()
        py_res, py_old = factory(gen0)(py_arr)
        c_gen = jit(nopython=True)(factory(jit(nopython=True)(gen0)))
        c_res, c_old = c_gen(c_arr)

        self.assertIsNot(py_arr, c_arr)
        self.assertIs(py_old, py_arr)
        self.assertIs(c_old, c_arr)

        np.testing.assert_equal(py_res, c_res)

        self.assertEqual(sys.getrefcount(py_res),
                         sys.getrefcount(c_res))

 def test_lookup_commit_refcount(self):
     start = sys.getrefcount(self.repo)
     commit_sha = '5fe808e8953c12735680c257f56600cb0de44b10'
     commit = self.repo[commit_sha]
     del commit
     end = sys.getrefcount(self.repo)
     self.assertEqual(start, end)