| """Utilities for processing .test files containing test case descriptions.""" |
| |
| from __future__ import annotations |
| |
| import os |
| import os.path |
| import posixpath |
| import re |
| import shutil |
| import sys |
| import tempfile |
| from abc import abstractmethod |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Any, Final, Iterator, NamedTuple, NoReturn, Pattern, Union |
| from typing_extensions import TypeAlias as _TypeAlias |
| |
| import pytest |
| |
| from mypy import defaults |
| from mypy.test.config import PREFIX, test_data_prefix, test_temp_dir |
| |
| root_dir = os.path.normpath(PREFIX) |
| |
| # Debuggers that we support for debugging mypyc run tests |
| # implementation of using each of these debuggers is in test_run.py |
| # TODO: support more debuggers |
| SUPPORTED_DEBUGGERS: Final = ["gdb", "lldb"] |
| |
| |
| # File modify/create operation: copy module contents from source_path. |
| class UpdateFile(NamedTuple): |
| module: str |
| content: str |
| target_path: str |
| |
| |
| # File delete operation: delete module file. |
| class DeleteFile(NamedTuple): |
| module: str |
| path: str |
| |
| |
| FileOperation: _TypeAlias = Union[UpdateFile, DeleteFile] |
| |
| |
| def _file_arg_to_module(filename: str) -> str: |
| filename, _ = os.path.splitext(filename) |
| parts = filename.split("/") # not os.sep since it comes from test data |
| if parts[-1] == "__init__": |
| parts.pop() |
| return ".".join(parts) |
| |
| |
| def parse_test_case(case: DataDrivenTestCase) -> None: |
| """Parse and prepare a single case from suite with test case descriptions. |
| |
| This method is part of the setup phase, just before the test case is run. |
| """ |
| test_items = parse_test_data(case.data, case.name) |
| base_path = case.suite.base_path |
| if case.suite.native_sep: |
| join = os.path.join |
| else: |
| join = posixpath.join |
| |
| out_section_missing = case.suite.required_out_section |
| |
| files: list[tuple[str, str]] = [] # path and contents |
| output_files: list[tuple[str, str | Pattern[str]]] = [] # output path and contents |
| output: list[str] = [] # Regular output errors |
| output2: dict[int, list[str]] = {} # Output errors for incremental, runs 2+ |
| deleted_paths: dict[int, set[str]] = {} # from run number of paths |
| stale_modules: dict[int, set[str]] = {} # from run number to module names |
| rechecked_modules: dict[int, set[str]] = {} # from run number module names |
| triggered: list[str] = [] # Active triggers (one line per incremental step) |
| targets: dict[int, list[str]] = {} # Fine-grained targets (per fine-grained update) |
| test_modules: list[str] = [] # Modules which are deemed "test" (vs "fixture") |
| |
| def _case_fail(msg: str) -> NoReturn: |
| pytest.fail(f"{case.file}:{case.line}: {msg}", pytrace=False) |
| |
| # Process the parsed items. Each item has a header of form [id args], |
| # optionally followed by lines of text. |
| item = first_item = test_items[0] |
| test_modules.append("__main__") |
| for item in test_items[1:]: |
| |
| def _item_fail(msg: str) -> NoReturn: |
| item_abs_line = case.line + item.line - 2 |
| pytest.fail(f"{case.file}:{item_abs_line}: {msg}", pytrace=False) |
| |
| if item.id in {"file", "fixture", "outfile", "outfile-re"}: |
| # Record an extra file needed for the test case. |
| assert item.arg is not None |
| contents = expand_variables("\n".join(item.data)) |
| path = join(base_path, item.arg) |
| if item.id != "fixture": |
| test_modules.append(_file_arg_to_module(item.arg)) |
| if item.id in {"file", "fixture"}: |
| files.append((path, contents)) |
| elif item.id == "outfile-re": |
| output_files.append((path, re.compile(contents.rstrip(), re.S))) |
| elif item.id == "outfile": |
| output_files.append((path, contents)) |
| elif item.id == "builtins": |
| # Use an alternative stub file for the builtins module. |
| assert item.arg is not None |
| mpath = join(os.path.dirname(case.file), item.arg) |
| with open(mpath, encoding="utf8") as f: |
| files.append((join(base_path, "builtins.pyi"), f.read())) |
| elif item.id == "typing": |
| # Use an alternative stub file for the typing module. |
| assert item.arg is not None |
| src_path = join(os.path.dirname(case.file), item.arg) |
| with open(src_path, encoding="utf8") as f: |
| files.append((join(base_path, "typing.pyi"), f.read())) |
| elif item.id == "_typeshed": |
| # Use an alternative stub file for the _typeshed module. |
| assert item.arg is not None |
| src_path = join(os.path.dirname(case.file), item.arg) |
| with open(src_path, encoding="utf8") as f: |
| files.append((join(base_path, "_typeshed.pyi"), f.read())) |
| elif re.match(r"stale[0-9]*$", item.id): |
| passnum = 1 if item.id == "stale" else int(item.id[len("stale") :]) |
| assert passnum > 0 |
| modules = set() if item.arg is None else {t.strip() for t in item.arg.split(",")} |
| stale_modules[passnum] = modules |
| elif re.match(r"rechecked[0-9]*$", item.id): |
| passnum = 1 if item.id == "rechecked" else int(item.id[len("rechecked") :]) |
| assert passnum > 0 |
| modules = set() if item.arg is None else {t.strip() for t in item.arg.split(",")} |
| rechecked_modules[passnum] = modules |
| elif re.match(r"targets[0-9]*$", item.id): |
| passnum = 1 if item.id == "targets" else int(item.id[len("targets") :]) |
| assert passnum > 0 |
| reprocessed = [] if item.arg is None else [t.strip() for t in item.arg.split(",")] |
| targets[passnum] = reprocessed |
| elif item.id == "delete": |
| # File/directory to delete during a multi-step test case |
| assert item.arg is not None |
| m = re.match(r"(.*)\.([0-9]+)$", item.arg) |
| if m is None: |
| _item_fail(f"Invalid delete section {item.arg!r}") |
| num = int(m.group(2)) |
| if num < 2: |
| _item_fail(f"Can't delete during step {num}") |
| full = join(base_path, m.group(1)) |
| deleted_paths.setdefault(num, set()).add(full) |
| elif re.match(r"out[0-9]*$", item.id): |
| if item.arg is None: |
| args = [] |
| else: |
| args = item.arg.split(",") |
| |
| version_check = True |
| for arg in args: |
| if arg.startswith("version"): |
| compare_op = arg[7:9] |
| if compare_op not in {">=", "=="}: |
| _item_fail("Only >= and == version checks are currently supported") |
| version_str = arg[9:] |
| try: |
| version = tuple(int(x) for x in version_str.split(".")) |
| except ValueError: |
| _item_fail(f"{version_str!r} is not a valid python version") |
| if compare_op == ">=": |
| if version <= defaults.PYTHON3_VERSION: |
| _item_fail( |
| f"{arg} always true since minimum runtime version is {defaults.PYTHON3_VERSION}" |
| ) |
| version_check = sys.version_info >= version |
| elif compare_op == "==": |
| if version < defaults.PYTHON3_VERSION: |
| _item_fail( |
| f"{arg} always false since minimum runtime version is {defaults.PYTHON3_VERSION}" |
| ) |
| if not 1 < len(version) < 4: |
| _item_fail( |
| f'Only minor or patch version checks are currently supported with "==": {version_str!r}' |
| ) |
| version_check = sys.version_info[: len(version)] == version |
| if version_check: |
| tmp_output = [expand_variables(line) for line in item.data] |
| if os.path.sep == "\\" and case.normalize_output: |
| tmp_output = [fix_win_path(line) for line in tmp_output] |
| if item.id == "out" or item.id == "out1": |
| output = tmp_output |
| else: |
| passnum = int(item.id[len("out") :]) |
| assert passnum > 1 |
| output2[passnum] = tmp_output |
| out_section_missing = False |
| elif item.id == "triggered" and item.arg is None: |
| triggered = item.data |
| else: |
| section_str = item.id + (f" {item.arg}" if item.arg else "") |
| _item_fail(f"Invalid section header [{section_str}] in case {case.name!r}") |
| |
| if out_section_missing: |
| _case_fail(f"Required output section not found in case {case.name!r}") |
| |
| for passnum in stale_modules.keys(): |
| if passnum not in rechecked_modules: |
| # If the set of rechecked modules isn't specified, make it the same as the set |
| # of modules with a stale public interface. |
| rechecked_modules[passnum] = stale_modules[passnum] |
| if ( |
| passnum in stale_modules |
| and passnum in rechecked_modules |
| and not stale_modules[passnum].issubset(rechecked_modules[passnum]) |
| ): |
| _case_fail(f"Stale modules after pass {passnum} must be a subset of rechecked modules") |
| |
| output_inline_start = len(output) |
| input = first_item.data |
| expand_errors(input, output, "main") |
| for file_path, contents in files: |
| expand_errors(contents.split("\n"), output, file_path) |
| |
| seen_files = set() |
| for file, _ in files: |
| if file in seen_files: |
| _case_fail(f"Duplicated filename {file}. Did you include it multiple times?") |
| |
| seen_files.add(file) |
| |
| case.input = input |
| case.output = output |
| case.output_inline_start = output_inline_start |
| case.output2 = output2 |
| case.last_line = case.line + item.line + len(item.data) - 2 |
| case.files = files |
| case.output_files = output_files |
| case.expected_stale_modules = stale_modules |
| case.expected_rechecked_modules = rechecked_modules |
| case.deleted_paths = deleted_paths |
| case.triggered = triggered or [] |
| case.expected_fine_grained_targets = targets |
| case.test_modules = test_modules |
| |
| |
| class DataDrivenTestCase(pytest.Item): |
| """Holds parsed data-driven test cases, and handles directory setup and teardown.""" |
| |
| # Override parent member type |
| parent: DataSuiteCollector |
| |
| input: list[str] |
| output: list[str] # Output for the first pass |
| output_inline_start: int |
| output2: dict[int, list[str]] # Output for runs 2+, indexed by run number |
| |
| # full path of test suite |
| file = "" |
| line = 0 |
| |
| # (file path, file content) tuples |
| files: list[tuple[str, str]] |
| # Modules which is to be considered "test" rather than "fixture" |
| test_modules: list[str] |
| expected_stale_modules: dict[int, set[str]] |
| expected_rechecked_modules: dict[int, set[str]] |
| expected_fine_grained_targets: dict[int, list[str]] |
| |
| # Whether or not we should normalize the output to standardize things like |
| # forward vs backward slashes in file paths for Windows vs Linux. |
| normalize_output: bool |
| |
| # Extra attributes used by some tests. |
| last_line: int |
| output_files: list[tuple[str, str | Pattern[str]]] # Path and contents for output files |
| deleted_paths: dict[int, set[str]] # Mapping run number -> paths |
| triggered: list[str] # Active triggers (one line per incremental step) |
| |
| def __init__( |
| self, |
| parent: DataSuiteCollector, |
| suite: DataSuite, |
| *, |
| file: str, |
| name: str, |
| writescache: bool, |
| only_when: str, |
| normalize_output: bool, |
| platform: str | None, |
| skip: bool, |
| xfail: bool, |
| data: str, |
| line: int, |
| ) -> None: |
| super().__init__(name, parent) |
| self.suite = suite |
| self.file = file |
| self.writescache = writescache |
| self.only_when = only_when |
| self.normalize_output = normalize_output |
| if (platform == "windows" and sys.platform != "win32") or ( |
| platform == "posix" and sys.platform == "win32" |
| ): |
| skip = True |
| self.skip = skip |
| self.xfail = xfail |
| self.data = data |
| self.line = line |
| self.old_cwd: str | None = None |
| self.tmpdir: tempfile.TemporaryDirectory[str] | None = None |
| |
| def runtest(self) -> None: |
| if self.skip: |
| pytest.skip() |
| # TODO: add a better error message for when someone uses skip and xfail at the same time |
| elif self.xfail: |
| self.add_marker(pytest.mark.xfail) |
| parent = self.getparent(DataSuiteCollector) |
| assert parent is not None, "Should not happen" |
| suite = parent.obj() |
| suite.setup() |
| try: |
| suite.run_case(self) |
| except Exception: |
| # As a debugging aid, support copying the contents of the tmp directory somewhere |
| save_dir: str | None = self.config.getoption("--save-failures-to", None) |
| if save_dir: |
| assert self.tmpdir is not None |
| target_dir = os.path.join(save_dir, os.path.basename(self.tmpdir.name)) |
| print(f"Copying data from test {self.name} to {target_dir}") |
| if not os.path.isabs(target_dir): |
| assert self.old_cwd |
| target_dir = os.path.join(self.old_cwd, target_dir) |
| shutil.copytree(self.tmpdir.name, target_dir) |
| raise |
| |
| def setup(self) -> None: |
| parse_test_case(case=self) |
| self.old_cwd = os.getcwd() |
| self.tmpdir = tempfile.TemporaryDirectory(prefix="mypy-test-") |
| os.chdir(self.tmpdir.name) |
| os.mkdir(test_temp_dir) |
| |
| # Precalculate steps for find_steps() |
| steps: dict[int, list[FileOperation]] = {} |
| |
| for path, content in self.files: |
| m = re.match(r".*\.([0-9]+)$", path) |
| if m: |
| # Skip writing subsequent incremental steps - rather |
| # store them as operations. |
| num = int(m.group(1)) |
| assert num >= 2 |
| target_path = re.sub(r"\.[0-9]+$", "", path) |
| module = module_from_path(target_path) |
| operation = UpdateFile(module, content, target_path) |
| steps.setdefault(num, []).append(operation) |
| else: |
| # Write the first incremental steps |
| dir = os.path.dirname(path) |
| os.makedirs(dir, exist_ok=True) |
| with open(path, "w", encoding="utf8") as f: |
| f.write(content) |
| |
| for num, paths in self.deleted_paths.items(): |
| assert num >= 2 |
| for path in paths: |
| module = module_from_path(path) |
| steps.setdefault(num, []).append(DeleteFile(module, path)) |
| max_step = max(steps) if steps else 2 |
| self.steps = [steps.get(num, []) for num in range(2, max_step + 1)] |
| |
| def teardown(self) -> None: |
| if self.old_cwd is not None: |
| os.chdir(self.old_cwd) |
| if self.tmpdir is not None: |
| try: |
| self.tmpdir.cleanup() |
| except OSError: |
| pass |
| self.old_cwd = None |
| self.tmpdir = None |
| |
| def reportinfo(self) -> tuple[str, int, str]: |
| return self.file, self.line, self.name |
| |
| def repr_failure( |
| self, excinfo: pytest.ExceptionInfo[BaseException], style: Any | None = None |
| ) -> str: |
| excrepr: object |
| if isinstance(excinfo.value, SystemExit): |
| # We assume that before doing exit() (which raises SystemExit) we've printed |
| # enough context about what happened so that a stack trace is not useful. |
| # In particular, uncaught exceptions during semantic analysis or type checking |
| # call exit() and they already print out a stack trace. |
| excrepr = excinfo.exconly() |
| elif isinstance(excinfo.value, pytest.fail.Exception) and not excinfo.value.pytrace: |
| excrepr = excinfo.exconly() |
| else: |
| excinfo.traceback = self.parent._traceback_filter(excinfo) |
| excrepr = excinfo.getrepr(style="short") |
| |
| return f"data: {self.file}:{self.line}:\n{excrepr}" |
| |
| def find_steps(self) -> list[list[FileOperation]]: |
| """Return a list of descriptions of file operations for each incremental step. |
| |
| The first list item corresponds to the first incremental step, the second for the |
| second step, etc. Each operation can either be a file modification/creation (UpdateFile) |
| or deletion (DeleteFile). |
| |
| Defaults to having two steps if there aern't any operations. |
| """ |
| return self.steps |
| |
| |
| def module_from_path(path: str) -> str: |
| path = re.sub(r"\.pyi?$", "", path) |
| # We can have a mix of Unix-style and Windows-style separators. |
| parts = re.split(r"[/\\]", path) |
| del parts[0] |
| module = ".".join(parts) |
| module = re.sub(r"\.__init__$", "", module) |
| return module |
| |
| |
| @dataclass |
| class TestItem: |
| """Parsed test caseitem. |
| |
| An item is of the form |
| [id arg] |
| .. data .. |
| """ |
| |
| id: str |
| arg: str | None |
| # Processed, collapsed text data |
| data: list[str] |
| # Start line: 1-based, inclusive, relative to testcase |
| line: int |
| # End line: 1-based, exclusive, relative to testcase; not same as `line + len(test_item.data)` due to collapsing |
| end_line: int |
| |
| @property |
| def trimmed_newlines(self) -> int: # compensates for strip_list |
| return self.end_line - self.line - len(self.data) |
| |
| |
| def parse_test_data(raw_data: str, name: str) -> list[TestItem]: |
| """Parse a list of lines that represent a sequence of test items.""" |
| |
| lines = ["", "[case " + name + "]"] + raw_data.split("\n") |
| ret: list[TestItem] = [] |
| data: list[str] = [] |
| |
| id: str | None = None |
| arg: str | None = None |
| |
| i = 0 |
| i0 = 0 |
| while i < len(lines): |
| s = lines[i].strip() |
| |
| if lines[i].startswith("[") and s.endswith("]"): |
| if id: |
| data = collapse_line_continuation(data) |
| data = strip_list(data) |
| ret.append(TestItem(id, arg, data, i0 + 1, i)) |
| |
| i0 = i |
| id = s[1:-1] |
| arg = None |
| if " " in id: |
| arg = id[id.index(" ") + 1 :] |
| id = id[: id.index(" ")] |
| data = [] |
| elif lines[i].startswith("\\["): |
| data.append(lines[i][1:]) |
| elif not lines[i].startswith("--"): |
| data.append(lines[i]) |
| elif lines[i].startswith("----"): |
| data.append(lines[i][2:]) |
| i += 1 |
| |
| # Process the last item. |
| if id: |
| data = collapse_line_continuation(data) |
| data = strip_list(data) |
| ret.append(TestItem(id, arg, data, i0 + 1, i - 1)) |
| |
| return ret |
| |
| |
| def strip_list(l: list[str]) -> list[str]: |
| """Return a stripped copy of l. |
| |
| Strip whitespace at the end of all lines, and strip all empty |
| lines from the end of the array. |
| """ |
| |
| r: list[str] = [] |
| for s in l: |
| # Strip spaces at end of line |
| r.append(re.sub(r"\s+$", "", s)) |
| |
| while r and r[-1] == "": |
| r.pop() |
| |
| return r |
| |
| |
| def collapse_line_continuation(l: list[str]) -> list[str]: |
| r: list[str] = [] |
| cont = False |
| for s in l: |
| ss = re.sub(r"\\$", "", s) |
| if cont: |
| r[-1] += re.sub("^ +", "", ss) |
| else: |
| r.append(ss) |
| cont = s.endswith("\\") |
| return r |
| |
| |
| def expand_variables(s: str) -> str: |
| return s.replace("<ROOT>", root_dir) |
| |
| |
| def expand_errors(input: list[str], output: list[str], fnam: str) -> None: |
| """Transform comments such as '# E: message' or |
| '# E:3: message' in input. |
| |
| The result is lines like 'fnam:line: error: message'. |
| """ |
| |
| for i in range(len(input)): |
| # The first in the split things isn't a comment |
| for possible_err_comment in input[i].split(" # ")[1:]: |
| m = re.search( |
| r"^([ENW]):((?P<col>\d+):)? (?P<message>.*)$", possible_err_comment.strip() |
| ) |
| if m: |
| if m.group(1) == "E": |
| severity = "error" |
| elif m.group(1) == "N": |
| severity = "note" |
| elif m.group(1) == "W": |
| severity = "warning" |
| col = m.group("col") |
| message = m.group("message") |
| message = message.replace("\\#", "#") # adds back escaped # character |
| if col is None: |
| output.append(f"{fnam}:{i + 1}: {severity}: {message}") |
| else: |
| output.append(f"{fnam}:{i + 1}:{col}: {severity}: {message}") |
| |
| |
| def fix_win_path(line: str) -> str: |
| r"""Changes Windows paths to Linux paths in error messages. |
| |
| E.g. foo\bar.py -> foo/bar.py. |
| """ |
| line = line.replace(root_dir, root_dir.replace("\\", "/")) |
| m = re.match(r"^([\S/]+):(\d+:)?(\s+.*)", line) |
| if not m: |
| return line |
| else: |
| filename, lineno, message = m.groups() |
| return "{}:{}{}".format(filename.replace("\\", "/"), lineno or "", message) |
| |
| |
| def fix_cobertura_filename(line: str) -> str: |
| r"""Changes filename paths to Linux paths in Cobertura output files. |
| |
| E.g. filename="pkg\subpkg\a.py" -> filename="pkg/subpkg/a.py". |
| """ |
| m = re.search(r'<class .* filename="(?P<filename>.*?)"', line) |
| if not m: |
| return line |
| return "{}{}{}".format( |
| line[: m.start(1)], m.group("filename").replace("\\", "/"), line[m.end(1) :] |
| ) |
| |
| |
| ## |
| # |
| # pytest setup |
| # |
| ## |
| |
| |
| # This function name is special to pytest. See |
| # https://docs.pytest.org/en/latest/reference.html#initialization-hooks |
| def pytest_addoption(parser: Any) -> None: |
| group = parser.getgroup("mypy") |
| group.addoption( |
| "--update-data", |
| action="store_true", |
| default=False, |
| help="Update test data to reflect actual output (supported only for certain tests)", |
| ) |
| group.addoption( |
| "--save-failures-to", |
| default=None, |
| help="Copy the temp directories from failing tests to a target directory", |
| ) |
| group.addoption( |
| "--mypy-verbose", action="count", help="Set the verbose flag when creating mypy Options" |
| ) |
| group.addoption( |
| "--mypyc-showc", |
| action="store_true", |
| default=False, |
| help="Display C code on mypyc test failures", |
| ) |
| group.addoption( |
| "--mypyc-debug", |
| default=None, |
| dest="debugger", |
| choices=SUPPORTED_DEBUGGERS, |
| help="Run the first mypyc run test with the specified debugger", |
| ) |
| |
| |
| def pytest_configure(config: pytest.Config) -> None: |
| if config.getoption("--update-data") and config.getoption("--numprocesses", default=1) > 1: |
| raise pytest.UsageError( |
| "--update-data incompatible with parallelized tests; re-run with -n 1" |
| ) |
| |
| |
| # This function name is special to pytest. See |
| # https://doc.pytest.org/en/latest/how-to/writing_plugins.html#collection-hooks |
| def pytest_pycollect_makeitem(collector: Any, name: str, obj: object) -> Any | None: |
| """Called by pytest on each object in modules configured in conftest.py files. |
| |
| collector is pytest.Collector, returns Optional[pytest.Class] |
| """ |
| if isinstance(obj, type): |
| # Only classes derived from DataSuite contain test cases, not the DataSuite class itself |
| if issubclass(obj, DataSuite) and obj is not DataSuite: |
| # Non-None result means this obj is a test case. |
| # The collect method of the returned DataSuiteCollector instance will be called later, |
| # with self.obj being obj. |
| return DataSuiteCollector.from_parent( # type: ignore[no-untyped-call] |
| parent=collector, name=name |
| ) |
| return None |
| |
| |
| _case_name_pattern = re.compile( |
| r"(?P<name>[a-zA-Z_0-9]+)" |
| r"(?P<writescache>-writescache)?" |
| r"(?P<only_when>-only_when_cache|-only_when_nocache)?" |
| r"(?P<skip_path_normalization>-skip_path_normalization)?" |
| r"(-(?P<platform>posix|windows))?" |
| r"(?P<skip>-skip)?" |
| r"(?P<xfail>-xfail)?" |
| ) |
| |
| |
| def split_test_cases( |
| parent: DataFileCollector, suite: DataSuite, file: str |
| ) -> Iterator[DataDrivenTestCase]: |
| """Iterate over raw test cases in file, at collection time, ignoring sub items. |
| |
| The collection phase is slow, so any heavy processing should be deferred to after |
| uninteresting tests are filtered (when using -k PATTERN switch). |
| """ |
| with open(file, encoding="utf-8") as f: |
| data = f.read() |
| cases = re.split(r"^\[case ([^]+)]+)\][ \t]*$\n", data, flags=re.DOTALL | re.MULTILINE) |
| cases_iter = iter(cases) |
| line_no = next(cases_iter).count("\n") + 1 |
| test_names = set() |
| for case_id in cases_iter: |
| data = next(cases_iter) |
| |
| m = _case_name_pattern.fullmatch(case_id) |
| if not m: |
| raise RuntimeError(f"Invalid testcase id {case_id!r}") |
| name = m.group("name") |
| if name in test_names: |
| raise RuntimeError( |
| 'Found a duplicate test name "{}" in {} on line {}'.format( |
| name, parent.name, line_no |
| ) |
| ) |
| yield DataDrivenTestCase.from_parent( |
| parent=parent, |
| suite=suite, |
| file=file, |
| name=add_test_name_suffix(name, suite.test_name_suffix), |
| writescache=bool(m.group("writescache")), |
| only_when=m.group("only_when"), |
| platform=m.group("platform"), |
| skip=bool(m.group("skip")), |
| xfail=bool(m.group("xfail")), |
| normalize_output=not m.group("skip_path_normalization"), |
| data=data, |
| line=line_no, |
| ) |
| line_no += data.count("\n") + 1 |
| |
| # Record existing tests to prevent duplicates: |
| test_names.update({name}) |
| |
| |
| class DataSuiteCollector(pytest.Class): |
| def collect(self) -> Iterator[DataFileCollector]: |
| """Called by pytest on each of the object returned from pytest_pycollect_makeitem""" |
| |
| # obj is the object for which pytest_pycollect_makeitem returned self. |
| suite: DataSuite = self.obj |
| |
| assert os.path.isdir( |
| suite.data_prefix |
| ), f"Test data prefix ({suite.data_prefix}) not set correctly" |
| |
| for data_file in suite.files: |
| yield DataFileCollector.from_parent(parent=self, name=data_file) |
| |
| |
| class DataFileFix(NamedTuple): |
| lineno: int # 1-offset, inclusive |
| end_lineno: int # 1-offset, exclusive |
| lines: list[str] |
| |
| |
| class DataFileCollector(pytest.Collector): |
| """Represents a single `.test` data driven test file. |
| |
| More context: https://github.com/python/mypy/issues/11662 |
| """ |
| |
| parent: DataSuiteCollector |
| |
| _fixes: list[DataFileFix] |
| |
| @classmethod # We have to fight with pytest here: |
| def from_parent( |
| cls, parent: DataSuiteCollector, *, name: str # type: ignore[override] |
| ) -> DataFileCollector: |
| collector = super().from_parent(parent, name=name) |
| assert isinstance(collector, DataFileCollector) |
| return collector |
| |
| def collect(self) -> Iterator[DataDrivenTestCase]: |
| yield from split_test_cases( |
| parent=self, |
| suite=self.parent.obj, |
| file=os.path.join(self.parent.obj.data_prefix, self.name), |
| ) |
| |
| def setup(self) -> None: |
| super().setup() |
| self._fixes = [] |
| |
| def teardown(self) -> None: |
| super().teardown() |
| self._apply_fixes() |
| |
| def enqueue_fix(self, fix: DataFileFix) -> None: |
| self._fixes.append(fix) |
| |
| def _apply_fixes(self) -> None: |
| if not self._fixes: |
| return |
| data_path = Path(self.parent.obj.data_prefix) / self.name |
| lines = data_path.read_text().split("\n") |
| # start from end to prevent line offsets from shifting as we update |
| for fix in sorted(self._fixes, reverse=True): |
| lines[fix.lineno - 1 : fix.end_lineno - 1] = fix.lines |
| data_path.write_text("\n".join(lines)) |
| |
| |
| def add_test_name_suffix(name: str, suffix: str) -> str: |
| # Find magic suffix of form "-foobar" (used for things like "-skip"). |
| m = re.search(r"-[-A-Za-z0-9]+$", name) |
| if m: |
| # Insert suite-specific test name suffix before the magic suffix |
| # which must be the last thing in the test case name since we |
| # are using endswith() checks. |
| magic_suffix = m.group(0) |
| return name[: -len(magic_suffix)] + suffix + magic_suffix |
| else: |
| return name + suffix |
| |
| |
| def is_incremental(testcase: DataDrivenTestCase) -> bool: |
| return "incremental" in testcase.name.lower() or "incremental" in testcase.file |
| |
| |
| def has_stable_flags(testcase: DataDrivenTestCase) -> bool: |
| if any(re.match(r"# flags[2-9]:", line) for line in testcase.input): |
| return False |
| for filename, contents in testcase.files: |
| if os.path.basename(filename).startswith("mypy.ini."): |
| return False |
| return True |
| |
| |
| class DataSuite: |
| # option fields - class variables |
| files: list[str] |
| |
| base_path = test_temp_dir |
| |
| # Allow external users of the test code to override the data prefix |
| data_prefix = test_data_prefix |
| |
| required_out_section = False |
| |
| native_sep = False |
| |
| # Name suffix automatically added to each test case in the suite (can be |
| # used to distinguish test cases in suites that share data files) |
| test_name_suffix = "" |
| |
| def setup(self) -> None: |
| """Setup fixtures (ad-hoc)""" |
| |
| @abstractmethod |
| def run_case(self, testcase: DataDrivenTestCase) -> None: |
| raise NotImplementedError |
