MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / _collect_messages

Method _collect_messages

lib/sqlalchemy/testing/fixtures/mypy.py:127–176  ·  view source on GitHub ↗
(self, path)

Source from the content-addressed store, hash-verified

125 return files
126
127 def _collect_messages(self, path):
128 expected_messages = []
129 expected_re = re.compile(r"\s*# EXPECTED(_MYPY)?(_RE)?(_TYPE)?: (.+)")
130 py_ver_re = re.compile(r"^#\s*PYTHON_VERSION\s?>=\s?(\d+\.\d+)")
131 with open(path) as file_:
132 current_assert_messages = []
133 for num, line in enumerate(file_, 1):
134 m = py_ver_re.match(line)
135 if m:
136 major, _, minor = m.group(1).partition(".")
137 if sys.version_info < (int(major), int(minor)):
138 config.skip_test(
139 "Requires python >= %s" % (m.group(1))
140 )
141 continue
142
143 m = expected_re.match(line)
144 if m:
145 is_mypy = bool(m.group(1))
146 is_re = bool(m.group(2))
147 is_type = bool(m.group(3))
148 expected_msg = re.sub(r"# noqa[:]? ?.*", "", m.group(4))
149
150 if is_type:
151
152 is_mypy = is_re = True
153 expected_msg = f'Revealed type is "{expected_msg}"'
154
155 # use_or_syntax
156 # https://github.com/python/mypy/blob/304997bfb85200fb521ac727ee0ce3e6085e5278/mypy/options.py#L368 # noqa: E501
157 expected_msg = re.sub(
158 r"Optional\[(.*?)\]",
159 lambda m: f"{m.group(1)} | None",
160 expected_msg,
161 )
162 current_assert_messages.append(
163 (is_mypy, is_re, expected_msg.strip())
164 )
165 elif current_assert_messages:
166 expected_messages.extend(
167 (num, is_mypy, is_re, expected_msg)
168 for (
169 is_mypy,
170 is_re,
171 expected_msg,
172 ) in current_assert_messages
173 )
174 current_assert_messages[:] = []
175
176 return expected_messages
177
178 def _check_output(
179 self, path, expected_messages, stdout: str, stderr, exitcode

Callers 1

runMethod · 0.95

Calls 8

openFunction · 0.85
skip_testMethod · 0.80
subMethod · 0.80
stripMethod · 0.80
compileMethod · 0.45
matchMethod · 0.45
appendMethod · 0.45
extendMethod · 0.45

Tested by

no test coverage detected