File: test_dealloc.py

package info (click to toggle)
uvloop 0.21.0%2Bds1-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, trixie
  • size: 10,308 kB
  • sloc: python: 8,273; ansic: 108; makefile: 42
file content (61 lines) | stat: -rw-r--r-- 1,843 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import asyncio
import subprocess
import sys

from uvloop import _testbase as tb


class TestDealloc(tb.UVTestCase):

    def test_dealloc_1(self):
        # Somewhere between Cython 0.25.2 and 0.26.0 uvloop programs
        # started to trigger the following output:
        #
        #    $ python prog.py
        #    Error in sys.excepthook:
        #
        #    Original exception was:
        #
        # Upon some debugging, it appeared that Handle.__dealloc__ was
        # called at a time where some CPython objects become non-functional,
        # and any exception in __dealloc__ caused CPython to output the
        # above.
        #
        # This regression test starts an event loop in debug mode,
        # lets it run for a brief period of time, and exits the program.
        # This will trigger Handle.__dealloc__, CallbackHandle.__dealloc__,
        # and Loop.__dealloc__ methods.  The test will fail if they produce
        # any unwanted output.

        async def test():
            prog = '''\
import uvloop

async def foo():
    return 42

def main():
    loop = uvloop.new_event_loop()
    loop.set_debug(True)
    loop.run_until_complete(foo())
    # Do not close the loop on purpose: let __dealloc__ methods run.

if __name__ == '__main__':
    main()
            '''

            cmd = sys.executable
            proc = await asyncio.create_subprocess_exec(
                cmd, b'-W', b'ignore', b'-c', prog,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE)

            await proc.wait()
            out = await proc.stdout.read()
            err = await proc.stderr.read()

            return out, err

        out, err = self.loop.run_until_complete(test())
        self.assertEqual(out, b'', 'stdout is not empty')
        self.assertEqual(err, b'', 'stderr is not empty')