Загрузить файлы в «venv/Lib/site-packages/greenlet/tests»

venv/Lib/site-packages/greenlet/tests/test_extension_interface.py

@@ -0,0 +1,141 @@
+import sys
+
+import greenlet
+from . import _test_extension
+from . import TestCase
+from .leakcheck import ignores_leakcheck
+
+# pylint:disable=c-extension-no-member
+
+class CAPITests(TestCase):
+    def test_switch(self):
+        self.assertEqual(
+            50, _test_extension.test_switch(greenlet.greenlet(lambda: 50)))
+
+    def test_switch_kwargs(self):
+        def adder(x, y):
+            return x * y
+        g = greenlet.greenlet(adder)
+        self.assertEqual(6, _test_extension.test_switch_kwargs(g, x=3, y=2))
+
+        with self.assertRaisesRegex(TypeError, "argument 1 must be greenlet"):
+            _test_extension.test_switch_kwargs("not a greenlet")
+
+    def test_setparent(self):
+        # pylint:disable=disallowed-name
+        def foo():
+            def bar():
+                greenlet.getcurrent().parent.switch()
+
+                # This final switch should go back to the main greenlet, since
+                # the test_setparent() function in the C extension should have
+                # reparented this greenlet.
+                greenlet.getcurrent().parent.switch()
+                raise AssertionError("Should never have reached this code")
+            child = greenlet.greenlet(bar)
+            child.switch()
+            greenlet.getcurrent().parent.switch(child)
+            greenlet.getcurrent().parent.throw(
+                AssertionError("Should never reach this code"))
+        foo_child = greenlet.greenlet(foo).switch()
+        self.assertEqual(None, _test_extension.test_setparent(foo_child))
+
+    def test_getcurrent(self):
+        _test_extension.test_getcurrent()
+
+    def test_new_greenlet(self):
+        self.assertEqual(-15, _test_extension.test_new_greenlet(lambda: -15))
+
+    def test_raise_greenlet_dead(self):
+        self.assertRaises(
+            greenlet.GreenletExit, _test_extension.test_raise_dead_greenlet)
+
+    def test_raise_greenlet_error(self):
+        self.assertRaises(
+            greenlet.error, _test_extension.test_raise_greenlet_error)
+
+    def test_throw(self):
+        seen = []
+
+        def foo(): # pylint:disable=disallowed-name
+            try:
+                greenlet.getcurrent().parent.switch()
+            except ValueError:
+                seen.append(sys.exc_info()[1])
+            except greenlet.GreenletExit:
+                raise AssertionError
+        g = greenlet.greenlet(foo)
+        g.switch()
+        _test_extension.test_throw(g)
+        self.assertEqual(len(seen), 1)
+        self.assertTrue(
+            isinstance(seen[0], ValueError),
+            "ValueError was not raised in foo()")
+        self.assertEqual(
+            str(seen[0]),
+            'take that sucka!',
+            "message doesn't match")
+
+    def test_non_traceback_param(self):
+        with self.assertRaises(TypeError) as exc:
+            _test_extension.test_throw_exact(
+                greenlet.getcurrent(),
+                Exception,
+                Exception(),
+                self
+            )
+        self.assertEqual(str(exc.exception),
+                         "throw() third argument must be a traceback object")
+
+    def test_instance_of_wrong_type(self):
+        with self.assertRaises(TypeError) as exc:
+            _test_extension.test_throw_exact(
+                greenlet.getcurrent(),
+                Exception(),
+                BaseException(),
+                None,
+            )
+
+        self.assertEqual(str(exc.exception),
+                         "instance exception may not have a separate value")
+
+    def test_not_throwable(self):
+        with self.assertRaises(TypeError) as exc:
+            _test_extension.test_throw_exact(
+                greenlet.getcurrent(),
+                "abc",
+                None,
+                None,
+            )
+        self.assertEqual(str(exc.exception),
+                         "exceptions must be classes, or instances, not str")
+
+    @ignores_leakcheck
+    def test_leaks(self):
+        from . import PY314
+        iters = 100
+        if PY314:
+            expected_refs = [1] * iters
+        else:
+            expected_refs = [2] * iters
+        for name, caller in (
+            ("test_switch",
+             lambda: _test_extension.test_switch(greenlet.greenlet(object))),
+            ("test_switch_kwargs",
+             lambda: _test_extension.test_switch_kwargs(greenlet.greenlet(object))),
+            ("test_new_greenlet",
+             lambda: _test_extension.test_new_greenlet(object)),
+        ):
+            with self.subTest(name):
+                results = [caller() for _ in range(iters)]
+                refs = [
+                    sys.getrefcount(i) - 1 # ignore ref in ``i``
+                    for i
+                    in results
+                ]
+                self.assertEqual(refs, expected_refs)
+
+
+if __name__ == '__main__':
+    import unittest
+    unittest.main()

venv/Lib/site-packages/greenlet/tests/test_gc.py

@@ -0,0 +1,137 @@
+import gc
+
+import weakref
+import sys
+import greenlet
+
+
+from . import TestCase
+from .leakcheck import fails_leakcheck_on_py314_or_less
+# These only work with greenlet gc support
+# which is no longer optional.
+assert greenlet.GREENLET_USE_GC
+
+class TestGC(TestCase):
+    def test_dead_circular_ref(self):
+        o = weakref.ref(greenlet.greenlet(greenlet.getcurrent).switch())
+        gc.collect()
+        if o() is not None:
+            print("O IS NOT NONE.", sys.getrefcount(o()))
+        self.assertIsNone(o())
+        self.assertFalse(gc.garbage, gc.garbage)
+
+    def test_circular_greenlet(self):
+        class circular_greenlet(greenlet.greenlet):
+            self = None
+        o = circular_greenlet()
+        o.self = o
+        o = weakref.ref(o)
+        gc.collect()
+        self.assertIsNone(o())
+        self.assertFalse(gc.garbage, gc.garbage)
+
+    def test_inactive_ref(self):
+        class inactive_greenlet(greenlet.greenlet):
+            def __init__(self):
+                greenlet.greenlet.__init__(self, run=self.run)
+
+            def run(self):
+                pass
+        o = inactive_greenlet()
+        o = weakref.ref(o)
+        gc.collect()
+        self.assertIsNone(o())
+        self.assertFalse(gc.garbage, gc.garbage)
+
+    @fails_leakcheck_on_py314_or_less
+    def test_finalizer_crash(self):
+        # This test is designed to crash when active greenlets
+        # are made garbage collectable, until the underlying
+        # problem is resolved. How does it work:
+        # - order of object creation is important
+        # - array is created first, so it is moved to unreachable first
+        # - we create a cycle between a greenlet and this array
+        # - we create an object that participates in gc, is only
+        #   referenced by a greenlet, and would corrupt gc lists
+        #   on destruction, the easiest is to use an object with
+        #   a finalizer
+        # - because array is the first object in unreachable it is
+        #   cleared first, which causes all references to greenlet
+        #   to disappear and causes greenlet to be destroyed, but since
+        #   it is still live it causes a switch during gc, which causes
+        #   an object with finalizer to be destroyed, which causes stack
+        #   corruption and then a crash
+
+        class object_with_finalizer(object):
+            def __del__(self):
+                pass
+        array = []
+        parent = greenlet.getcurrent()
+        def greenlet_body():
+            greenlet.getcurrent().object = object_with_finalizer()
+            try:
+                parent.switch()
+            except greenlet.GreenletExit:
+                print("Got greenlet exit!")
+            finally:
+                del greenlet.getcurrent().object
+        g = greenlet.greenlet(greenlet_body)
+        g.array = array
+        array.append(g)
+        g.switch()
+        del array
+        del g
+        greenlet.getcurrent()
+        gc.collect()
+
+    def test_crashing_deferred_object(self):
+        if sys.version_info < (3, 15):
+            self.skipTest("Test is 3.15+ only")
+        import doctest
+        def with_doctest():
+            """
+            >>> import gc
+            >>> from greenlet import getcurrent, greenlet, GreenletExit
+            >>> def outer():
+            ...     gc.collect()
+            >>> outer_glet = greenlet(outer)
+            >>> outer_glet.switch()
+            """
+        doctest.run_docstring_examples(with_doctest, dict())
+
+    def test_cycle_in_suspended_frame(self):
+        if sys.version_info < (3, 15):
+            self.skipTest("Test is 3.15+ only")
+        import doctest
+        def with_doctest():
+            """
+            >>> import gc
+            >>> from greenlet import getcurrent, greenlet
+            >>> class Cycle:
+            ...     def __del__(self):
+            ...         print("(Running finalizer)")
+            >>> def collect_it():
+            ...     print("Collecting garbage")
+            ...     gc.collect()
+            >>> def inner():
+            ...     cycle1 = Cycle()
+            ...     cycle2 = Cycle()
+            ...     cycle1.cycle = cycle2
+            ...     cycle2.cycle = cycle1
+            ...     getcurrent().parent.switch()
+            >>> def outer():
+            ...     glet = greenlet(inner)
+            ...     glet.switch()
+            ...     collect_it()
+
+            >>> outer_glet = greenlet(outer)
+            >>> outer_glet.switch()
+            Collecting garbage
+            >>> outer_glet.dead
+            True
+            >>> collect_it()
+            Collecting garbage
+            (Running finalizer)
+            (Running finalizer)
+            """
+        doctest.run_docstring_examples(with_doctest, dict())

venv/Lib/site-packages/greenlet/tests/test_generator.py

@@ -0,0 +1,59 @@
+
+from greenlet import greenlet
+
+from . import TestCase
+
+class genlet(greenlet):
+    parent = None
+    def __init__(self, *args, **kwds):
+        self.args = args
+        self.kwds = kwds
+
+    def run(self):
+        fn, = self.fn
+        fn(*self.args, **self.kwds)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        self.parent = greenlet.getcurrent()
+        result = self.switch()
+        if self:
+            return result
+
+        raise StopIteration
+
+    next = __next__
+
+
+def Yield(value):
+    g = greenlet.getcurrent()
+    while not isinstance(g, genlet):
+        if g is None:
+            raise RuntimeError('yield outside a genlet')
+        g = g.parent
+    g.parent.switch(value)
+
+
+def generator(func):
+    class Generator(genlet):
+        fn = (func,)
+    return Generator
+
+# ____________________________________________________________
+
+
+class GeneratorTests(TestCase):
+    def test_generator(self):
+        seen = []
+
+        def g(n):
+            for i in range(n):
+                seen.append(i)
+                Yield(i)
+        g = generator(g)
+        for _ in range(3):
+            for j in g(5):
+                seen.append(j)
+        self.assertEqual(seen, 3 * [0, 0, 1, 1, 2, 2, 3, 3, 4, 4])

venv/Lib/site-packages/greenlet/tests/test_generator_nested.py

@@ -0,0 +1,168 @@
+
+from greenlet import greenlet
+from . import TestCase
+from .leakcheck import fails_leakcheck
+
+class genlet(greenlet):
+    parent = None
+    def __init__(self, *args, **kwds):
+        self.args = args
+        self.kwds = kwds
+        self.child = None
+
+    def run(self):
+        # Note the function is packed in a tuple
+        # to avoid creating a bound method for it.
+        fn, = self.fn
+        fn(*self.args, **self.kwds)
+
+    def __iter__(self):
+        return self
+
+    def set_child(self, child):
+        self.child = child
+
+    def __next__(self):
+        if self.child:
+            child = self.child
+            while child.child:
+                tmp = child
+                child = child.child
+                tmp.child = None
+
+            result = child.switch()
+        else:
+            self.parent = greenlet.getcurrent()
+            result = self.switch()
+
+        if self:
+            return result
+
+        raise StopIteration
+
+    next = __next__
+
+def Yield(value, level=1):
+    g = greenlet.getcurrent()
+
+    while level != 0:
+        if not isinstance(g, genlet):
+            raise RuntimeError('yield outside a genlet')
+        if level > 1:
+            g.parent.set_child(g)
+        g = g.parent
+        level -= 1
+
+    g.switch(value)
+
+
+def Genlet(func):
+    class TheGenlet(genlet):
+        fn = (func,)
+    return TheGenlet
+
+# ____________________________________________________________
+
+
+def g1(n, seen):
+    for i in range(n):
+        seen.append(i + 1)
+        yield i
+
+
+def g2(n, seen):
+    for i in range(n):
+        seen.append(i + 1)
+        Yield(i)
+
+g2 = Genlet(g2)
+
+
+def nested(i):
+    Yield(i)
+
+
+def g3(n, seen):
+    for i in range(n):
+        seen.append(i + 1)
+        nested(i)
+g3 = Genlet(g3)
+
+
+def a(n):
+    if n == 0:
+        return
+    for ii in ax(n - 1):
+        Yield(ii)
+    Yield(n)
+ax = Genlet(a)
+
+
+def perms(l):
+    if len(l) > 1:
+        for e in l:
+            # No syntactical sugar for generator expressions
+            x = [Yield([e] + p) for p in perms([x for x in l if x != e])]
+            assert x
+    else:
+        Yield(l)
+perms = Genlet(perms)
+
+
+def gr1(n):
+    for ii in range(1, n):
+        Yield(ii)
+        Yield(ii * ii, 2)
+
+gr1 = Genlet(gr1)
+
+
+def gr2(n, seen):
+    for ii in gr1(n):
+        seen.append(ii)
+
+gr2 = Genlet(gr2)
+
+
+class NestedGeneratorTests(TestCase):
+    def test_layered_genlets(self):
+        seen = []
+        for ii in gr2(5, seen):
+            seen.append(ii)
+        self.assertEqual(seen, [1, 1, 2, 4, 3, 9, 4, 16])
+
+    @fails_leakcheck
+    def test_permutations(self):
+        gen_perms = perms(list(range(4)))
+        permutations = list(gen_perms)
+        self.assertEqual(len(permutations), 4 * 3 * 2 * 1)
+        self.assertIn([0, 1, 2, 3], permutations)
+        self.assertIn([3, 2, 1, 0], permutations)
+        res = []
+        for ii in zip(perms(list(range(4))), perms(list(range(3)))):
+            res.append(ii)
+        self.assertEqual(
+            res,
+            [([0, 1, 2, 3], [0, 1, 2]), ([0, 1, 3, 2], [0, 2, 1]),
+             ([0, 2, 1, 3], [1, 0, 2]), ([0, 2, 3, 1], [1, 2, 0]),
+             ([0, 3, 1, 2], [2, 0, 1]), ([0, 3, 2, 1], [2, 1, 0])])
+        # XXX Test to make sure we are working as a generator expression
+
+    def test_genlet_simple(self):
+        for g in g1, g2, g3:
+            seen = []
+            for _ in range(3):
+                for j in g(5, seen):
+                    seen.append(j)
+            self.assertEqual(seen, 3 * [1, 0, 2, 1, 3, 2, 4, 3, 5, 4])
+
+    def test_genlet_bad(self):
+        try:
+            Yield(10)
+        except RuntimeError:
+            pass
+
+    def test_nested_genlets(self):
+        seen = []
+        for ii in ax(5):
+            seen.append(ii)