|
10 | 10 | # License for the specific language governing permissions and limitations |
11 | 11 | # under the License. |
12 | 12 |
|
| 13 | +import ast |
13 | 14 | import os |
14 | 15 | import re |
15 | 16 |
|
@@ -74,7 +75,7 @@ def assert_no_duplicated_setup(logical_line, filename): |
74 | 75 |
|
75 | 76 |
|
76 | 77 | @core.flake8ext |
77 | | -def assert_use_of_client_aliases(logical_line, filename): |
| 78 | +def assert_use_of_client_aliases(logical_line): |
78 | 79 | """Ensure we use $service_client instead of $sdk_connection.service. |
79 | 80 |
|
80 | 81 | O402 |
@@ -106,3 +107,73 @@ def assert_use_of_client_aliases(logical_line, filename): |
106 | 107 | f"O402: {match.group(1)} is already a mock: there's no need to " |
107 | 108 | f"assign a new mock.Mock instance.", |
108 | 109 | ) |
| 110 | + |
| 111 | + |
| 112 | +class SDKProxyFindChecker(ast.NodeVisitor): |
| 113 | + """NodeVisitor to find ``*_client.find_*`` statements.""" |
| 114 | + |
| 115 | + def __init__(self): |
| 116 | + self.error = False |
| 117 | + |
| 118 | + def visit_Call(self, node): |
| 119 | + # No need to keep visiting the AST if we already found something. |
| 120 | + if self.error: |
| 121 | + return |
| 122 | + |
| 123 | + self.generic_visit(node) |
| 124 | + |
| 125 | + if not ( |
| 126 | + isinstance(node.func, ast.Attribute) |
| 127 | + and node.func.attr.startswith('find_') # and |
| 128 | + # isinstance(node.func.value, ast.Attribute) and |
| 129 | + # node.func.value.attr.endswith('_client') |
| 130 | + ): |
| 131 | + # print(f'skipping: got {node.func}') |
| 132 | + return |
| 133 | + |
| 134 | + if not ( |
| 135 | + ( |
| 136 | + # handle calls like 'identity_client.find_project' |
| 137 | + isinstance(node.func.value, ast.Name) |
| 138 | + and node.func.value.id.endswith('client') |
| 139 | + ) |
| 140 | + or ( |
| 141 | + # handle calls like 'self.app.client_manager.image.find_image' |
| 142 | + isinstance(node.func.value, ast.Attribute) |
| 143 | + and node.func.value.attr |
| 144 | + in ('identity', 'network', 'image', 'compute') |
| 145 | + ) |
| 146 | + ): |
| 147 | + return |
| 148 | + |
| 149 | + if not any(kw.arg == 'ignore_missing' for kw in node.keywords): |
| 150 | + self.error = True |
| 151 | + |
| 152 | + |
| 153 | +@core.flake8ext |
| 154 | +def assert_find_ignore_missing_kwargs(logical_line, filename): |
| 155 | + """Ensure ignore_missing is always used for ``find_*`` SDK proxy calls. |
| 156 | +
|
| 157 | + Okay: self.compute_client.find_server(foo, ignore_missing=True) |
| 158 | + Okay: self.image_client.find_server(foo, ignore_missing=False) |
| 159 | + Okay: self.volume_client.volumes.find(name='foo') |
| 160 | + O403: self.network_client.find_network(parsed_args.network) |
| 161 | + O403: self.compute_client.find_flavor(flavor_id, get_extra_specs=True) |
| 162 | + """ |
| 163 | + if 'tests' in filename: |
| 164 | + return |
| 165 | + |
| 166 | + checker = SDKProxyFindChecker() |
| 167 | + try: |
| 168 | + parsed_logical_line = ast.parse(logical_line) |
| 169 | + except SyntaxError: |
| 170 | + # let flake8 catch this itself |
| 171 | + # https://github.com/PyCQA/flake8/issues/1948 |
| 172 | + return |
| 173 | + checker.visit(parsed_logical_line) |
| 174 | + if checker.error: |
| 175 | + yield ( |
| 176 | + 0, |
| 177 | + 'O403: Calls to find_* proxy methods must explicitly set ' |
| 178 | + 'ignore_missing', |
| 179 | + ) |
0 commit comments