codekingpro/portable-devtools
114k
1# testing/fixtures/mypy.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9from __future__ import annotations
10
11import inspect
12import os
13from pathlib import Path
14import re
15import shutil
16import sys
17import tempfile
18
19from .base import TestBase
20from .. import config
21from ..assertions import eq_
22from ... import util
23
24
25@config.add_to_marker.mypy
26class MypyTest(TestBase):
27 __requires__ = ("no_sqlalchemy2_stubs",)
28
29 @config.fixture(scope="function")
30 def per_func_cachedir(self):
31 yield from self._cachedir()
32
33 @config.fixture(scope="class")
34 def cachedir(self):
35 yield from self._cachedir()
36
37 def _cachedir(self):
38 # as of mypy 0.971 i think we need to keep mypy_path empty
39 mypy_path = ""
40
41 with tempfile.TemporaryDirectory() as cachedir:
42 with open(
43 Path(cachedir) / "sqla_mypy_config.cfg", "w"
44 ) as config_file:
45 config_file.write(
46 f"""
47 [mypy]\n
48 plugins = sqlalchemy.ext.mypy.plugin\n
49 show_error_codes = True\n
50 {mypy_path}
51 disable_error_code = no-untyped-call
52
53 [mypy-sqlalchemy.*]
54 ignore_errors = True
55
56 """
57 )
58 with open(
59 Path(cachedir) / "plain_mypy_config.cfg", "w"
60 ) as config_file:
61 config_file.write(
62 f"""
63 [mypy]\n
64 show_error_codes = True\n
65 {mypy_path}
66 disable_error_code = var-annotated,no-untyped-call
67 [mypy-sqlalchemy.*]
68 ignore_errors = True
69
70 """
71 )
72 yield cachedir
73
74 @config.fixture()
75 def mypy_runner(self, cachedir):
76 from mypy import api
77
78 def run(path, use_plugin=False, use_cachedir=None):
79 if use_cachedir is None:
80 use_cachedir = cachedir
81 args = [
82 "--strict",
83 "--raise-exceptions",
84 "--cache-dir",
85 use_cachedir,
86 "--config-file",
87 os.path.join(
88 use_cachedir,
89 (
90 "sqla_mypy_config.cfg"
91 if use_plugin
92 else "plain_mypy_config.cfg"
93 ),
94 ),
95 ]
96
97 # mypy as of 0.990 is more aggressively blocking messaging
98 # for paths that are in sys.path, and as pytest puts currdir,
99 # test/ etc in sys.path, just copy the source file to the
100 # tempdir we are working in so that we don't have to try to
101 # manipulate sys.path and/or guess what mypy is doing
102 filename = os.path.basename(path)
103 test_program = os.path.join(use_cachedir, filename)
104 if path != test_program:
105 shutil.copyfile(path, test_program)
106 args.append(test_program)
107
108 # I set this locally but for the suite here needs to be
109 # disabled
110 os.environ.pop("MYPY_FORCE_COLOR", None)
111
112 stdout, stderr, exitcode = api.run(args)
113 return stdout, stderr, exitcode
114
115 return run
116
117 @config.fixture
118 def mypy_typecheck_file(self, mypy_runner):
119 def run(path, use_plugin=False):
120 expected_messages = self._collect_messages(path)
121 stdout, stderr, exitcode = mypy_runner(path, use_plugin=use_plugin)
122 self._check_output(
123 path, expected_messages, stdout, stderr, exitcode
124 )
125
126 return run
127
128 @staticmethod
129 def file_combinations(dirname):
130 if os.path.isabs(dirname):
131 path = dirname
132 else:
133 caller_path = inspect.stack()[1].filename
134 path = os.path.join(os.path.dirname(caller_path), dirname)
135 files = list(Path(path).glob("**/*.py"))
136
137 for extra_dir in config.options.mypy_extra_test_paths:
138 if extra_dir and os.path.isdir(extra_dir):
139 files.extend((Path(extra_dir) / dirname).glob("**/*.py"))
140 return files
141
142 def _collect_messages(self, path):
143 from sqlalchemy.ext.mypy.util import mypy_14
144
145 expected_messages = []
146 expected_re = re.compile(r"\s*# EXPECTED(_MYPY)?(_RE)?(_TYPE)?: (.+)")
147 py_ver_re = re.compile(r"^#\s*PYTHON_VERSION\s?>=\s?(\d+\.\d+)")
148 with open(path) as file_:
149 current_assert_messages = []
150 for num, line in enumerate(file_, 1):
151 m = py_ver_re.match(line)
152 if m:
153 major, _, minor = m.group(1).partition(".")
154 if sys.version_info < (int(major), int(minor)):
155 config.skip_test(
156 "Requires python >= %s" % (m.group(1))
157 )
158 continue
159
160 m = expected_re.match(line)
161 if m:
162 is_mypy = bool(m.group(1))
163 is_re = bool(m.group(2))
164 is_type = bool(m.group(3))
165
166 expected_msg = re.sub(r"# noqa[:]? ?.*", "", m.group(4))
167 if is_type:
168 if not is_re:
169 # the goal here is that we can cut-and-paste
170 # from vscode -> pylance into the
171 # EXPECTED_TYPE: line, then the test suite will
172 # validate that line against what mypy produces
173 expected_msg = re.sub(
174 r"([\[\]])",
175 lambda m: rf"\{m.group(0)}",
176 expected_msg,
177 )
178
179 # note making sure preceding text matches
180 # with a dot, so that an expect for "Select"
181 # does not match "TypedSelect"
182 expected_msg = re.sub(
183 r"([\w_]+)",
184 lambda m: rf"(?:.*\.)?{m.group(1)}\*?",
185 expected_msg,
186 )
187
188 expected_msg = re.sub(
189 "List", "builtins.list", expected_msg
190 )
191
192 expected_msg = re.sub(
193 r"\b(int|str|float|bool)\b",
194 lambda m: rf"builtins.{m.group(0)}\*?",
195 expected_msg,
196 )
197 # expected_msg = re.sub(
198 # r"(Sequence|Tuple|List|Union)",
199 # lambda m: fr"typing.{m.group(0)}\*?",
200 # expected_msg,
201 # )
202
203 is_mypy = is_re = True
204 expected_msg = f'Revealed type is "{expected_msg}"'
205
206 if mypy_14 and util.py39:
207 # use_lowercase_names, py39 and above
208 # https://github.com/python/mypy/blob/304997bfb85200fb521ac727ee0ce3e6085e5278/mypy/options.py#L363 # noqa: E501
209
210 # skip first character which could be capitalized
211 # "List item x not found" type of message
212 expected_msg = expected_msg[0] + re.sub(
213 (
214 r"\b(List|Tuple|Dict|Set)\b"
215 if is_type
216 else r"\b(List|Tuple|Dict|Set|Type)\b"
217 ),
218 lambda m: m.group(1).lower(),
219 expected_msg[1:],
220 )
221
222 if mypy_14 and util.py310:
223 # use_or_syntax, py310 and above
224 # https://github.com/python/mypy/blob/304997bfb85200fb521ac727ee0ce3e6085e5278/mypy/options.py#L368 # noqa: E501
225 expected_msg = re.sub(
226 r"Optional\[(.*?)\]",
227 lambda m: f"{m.group(1)} | None",
228 expected_msg,
229 )
230 current_assert_messages.append(
231 (is_mypy, is_re, expected_msg.strip())
232 )
233 elif current_assert_messages:
234 expected_messages.extend(
235 (num, is_mypy, is_re, expected_msg)
236 for (
237 is_mypy,
238 is_re,
239 expected_msg,
240 ) in current_assert_messages
241 )
242 current_assert_messages[:] = []
243
244 return expected_messages
245
246 def _check_output(self, path, expected_messages, stdout, stderr, exitcode):
247 not_located = []
248 filename = os.path.basename(path)
249 if expected_messages:
250 # mypy 0.990 changed how return codes work, so don't assume a
251 # 1 or a 0 return code here, could be either depending on if
252 # errors were generated or not
253
254 output = []
255
256 raw_lines = stdout.split("\n")
257 while raw_lines:
258 e = raw_lines.pop(0)
259 if re.match(r".+\.py:\d+: error: .*", e):
260 output.append(("error", e))
261 elif re.match(
262 r".+\.py:\d+: note: +(?:Possible overload|def ).*", e
263 ):
264 while raw_lines:
265 ol = raw_lines.pop(0)
266 if not re.match(r".+\.py:\d+: note: +def \[.*", ol):
267 break
268 elif re.match(
269 r".+\.py:\d+: note: .*(?:perhaps|suggestion)", e, re.I
270 ):
271 pass
272 elif re.match(r".+\.py:\d+: note: .*", e):
273 output.append(("note", e))
274
275 for num, is_mypy, is_re, msg in expected_messages:
276 msg = msg.replace("'", '"')
277 prefix = "[SQLAlchemy Mypy plugin] " if not is_mypy else ""
278 for idx, (typ, errmsg) in enumerate(output):
279 if is_re:
280 if re.match(
281 rf".*{filename}\:{num}\: {typ}\: {prefix}{msg}",
282 errmsg,
283 ):
284 break
285 elif (
286 f"{filename}:{num}: {typ}: {prefix}{msg}"
287 in errmsg.replace("'", '"')
288 ):
289 break
290 else:
291 not_located.append(msg)
292 continue
293 del output[idx]
294
295 if not_located:
296 missing = "\n".join(not_located)
297 print("Couldn't locate expected messages:", missing, sep="\n")
298 if output:
299 extra = "\n".join(msg for _, msg in output)
300 print("Remaining messages:", extra, sep="\n")
301 assert False, "expected messages not found, see stdout"
302
303 if output:
304 print(f"{len(output)} messages from mypy were not consumed:")
305 print("\n".join(msg for _, msg in output))
306 assert False, "errors and/or notes remain, see stdout"
307
308 else:
309 if exitcode != 0:
310 print(stdout, stderr, sep="\n")
311
312 eq_(exitcode, 0, msg=stdout)
313