flake8-async/tests/autofix_files/async91x_autofix.py at 94f2a5ae891d571bd14b18c1a2d1ea15703779f3 · jakkdl/flake8-async · GitHub

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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
# AUTOFIX
from __future__ import annotations

"""Docstring for file

So we make sure that import is added after it.
"""
# isort: skip_file
# ARG --enable=ASYNC910,ASYNC911

from typing import Any

import trio


def bar() -> Any: ...


async def foo() -> Any:
    await foo()


async def foo1():  # ASYNC910: 0, "exit", Statement("function definition", lineno)
    bar()
    await trio.lowlevel.checkpoint()


async def foo_return():
    bar()
    await trio.lowlevel.checkpoint()
    return  # ASYNC910: 4, "return", Statement("function definition", lineno-2)


async def foo_yield():  # ASYNC911: 0, "exit", Statement("yield", lineno+2)
    bar()
    await trio.lowlevel.checkpoint()
    yield  # ASYNC911: 4, "yield", Statement("function definition", lineno-2)
    await trio.lowlevel.checkpoint()


async def foo_if():
    if bar():
        await trio.lowlevel.checkpoint()
        return  # ASYNC910: 8, "return", Statement("function definition", lineno-2)
    elif bar():
        await trio.lowlevel.checkpoint()
        return  # ASYNC910: 8, "return", Statement("function definition", lineno-4)
    else:
        await trio.lowlevel.checkpoint()
        return  # ASYNC910: 8, "return", Statement("function definition", lineno-6)


async def foo_while():
    await foo()
    while True:
        await trio.lowlevel.checkpoint()
        yield  # ASYNC911: 8, "yield", Statement("yield", lineno)


async def foo_while2():
    await foo()
    while True:
        yield
        await foo()


async def foo_while3():
    await foo()
    while True:
        if bar():
            return
        await foo()


# check that multiple checkpoints don't get inserted
async def foo_while4():
    while True:
        if bar():
            await trio.lowlevel.checkpoint()
            yield  # ASYNC911: 12, "yield", Statement("yield", lineno)  # ASYNC911: 12, "yield", Statement("yield", lineno+2)  # ASYNC911: 12, "yield", Statement("function definition", lineno-3)
        if bar():
            await trio.lowlevel.checkpoint()
            yield  # ASYNC911: 12, "yield", Statement("yield", lineno)  # ASYNC911: 12, "yield", Statement("yield", lineno-2)  # ASYNC911: 12, "yield", Statement("function definition", lineno-5) # ASYNC911: 12, "yield", Statement("yield", lineno-2)
            # this warns about the yield on lineno-2 twice, since it can arrive here from it in two different ways


# check state management of nested loops
async def foo_nested_while():
    while True:
        await trio.lowlevel.checkpoint()
        yield  # ASYNC911: 8, "yield", Statement("function definition", lineno-2)
        while True:
            await trio.lowlevel.checkpoint()
            yield  # ASYNC911: 12, "yield", Statement("yield", lineno-2)
            while True:
                await trio.lowlevel.checkpoint()
                yield  # ASYNC911: 16, "yield", Statement("yield", lineno-2)  # ASYNC911: 16, "yield", Statement("yield", lineno)


async def foo_while_nested_func():
    while True:
        await trio.lowlevel.checkpoint()
        yield  # ASYNC911: 8, "yield", Statement("function definition", lineno-2) # ASYNC911: 8, "yield", Statement("yield", lineno)

        async def bar():
            while bar():
                ...
            await foo()


# Code coverage: visitors run when inside a sync function that has an async function.
# When sync funcs don't contain an async func the body is not visited.
def sync_func():
    async def async_func(): ...

    try:
        ...
    except:
        ...
    if bar() and bar():
        ...
    while ...:
        if bar():
            continue
        break
    [... for i in range(5)]
    return


# TODO: issue 240
async def livelocks():
    while True:
        ...


# this will autofix 910 by adding a checkpoint outside the loop
async def no_checkpoint():  # ASYNC910: 0, "exit", Statement("function definition", lineno)
    while bar():
        try:
            await trio.sleep("1")  # type: ignore[arg-type]
        except ValueError:
            ...
    await trio.lowlevel.checkpoint()