meta tests: refactor run_pytest

mypy/test/meta/_pytest.py

@@ -0,0 +1,72 @@
+import shlex
+import subprocess
+import sys
+import textwrap
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable
+
+from mypy.test.config import test_data_prefix
+
+
+@dataclass
+class PytestResult:
+    input: str
+    input_updated: str  # any updates made by --update-data
+    stdout: str
+    stderr: str
+
+
+def dedent_docstring(s: str) -> str:
+    return textwrap.dedent(s).lstrip()
+
+
+def run_pytest_data_suite(
+    data_suite: str,
+    *,
+    data_file_prefix: str = "check",
+    pytest_node_prefix: str = "mypy/test/testcheck.py::TypeCheckSuite",
+    extra_args: Iterable[str],
+    max_attempts: int,
+) -> PytestResult:
+    """
+    Runs a suite of data test cases through pytest until either tests pass
+    or until a maximum number of attempts (needed for incremental tests).
+
+    :param data_suite: the actual "suite" i.e. the contents of a .test file
+    """
+    p_test_data = Path(test_data_prefix)
+    p_root = p_test_data.parent.parent
+    p = p_test_data / f"{data_file_prefix}-meta-{uuid.uuid4()}.test"
+    assert not p.exists()
+    data_suite = dedent_docstring(data_suite)
+    try:
+        p.write_text(data_suite)
+
+        test_nodeid = f"{pytest_node_prefix}::{p.name}"
+        extra_args = [sys.executable, "-m", "pytest", "-n", "0", "-s", *extra_args, test_nodeid]
+        cmd = shlex.join(extra_args)
+        for i in range(max_attempts - 1, -1, -1):
+            print(f">> {cmd}")
+            proc = subprocess.run(extra_args, capture_output=True, check=False, cwd=p_root)
+            if proc.returncode == 0:
+                break
+            prefix = "NESTED PYTEST STDOUT"
+            for line in proc.stdout.decode().splitlines():
+                print(f"{prefix}: {line}")
+                prefix = " " * len(prefix)
+            prefix = "NESTED PYTEST STDERR"
+            for line in proc.stderr.decode().splitlines():
+                print(f"{prefix}: {line}")
+                prefix = " " * len(prefix)
+            print(f"Exit code {proc.returncode} ({i} attempts remaining)")
+
+        return PytestResult(
+            input=data_suite,
+            input_updated=p.read_text(),
+            stdout=proc.stdout.decode(),
+            stderr=proc.stderr.decode(),
+        )
+    finally:
+        p.unlink()

mypy/test/meta/test_parse_data.py

@@ -2,37 +2,18 @@
 A "meta test" which tests the parsing of .test files. This is not meant to become exhaustive
 but to ensure we maintain a basic level of ergonomics for mypy contributors.
 """
-import subprocess
-import sys
-import textwrap
-import uuid
-from pathlib import Path
-
-from mypy.test.config import test_data_prefix
 from mypy.test.helpers import Suite
+from mypy.test.meta._pytest import PytestResult, run_pytest_data_suite
 
 
-class ParseTestDataSuite(Suite):
-    def _dedent(self, s: str) -> str:
-        return textwrap.dedent(s).lstrip()
+def _run_pytest(data_suite: str) -> PytestResult:
+    return run_pytest_data_suite(data_suite, extra_args=[], max_attempts=1)
 
-    def _run_pytest(self, data_suite: str) -> str:
-        p_test_data = Path(test_data_prefix)
-        p_root = p_test_data.parent.parent
-        p = p_test_data / f"check-meta-{uuid.uuid4()}.test"
-        assert not p.exists()
-        try:
-            p.write_text(data_suite)
-            test_nodeid = f"mypy/test/testcheck.py::TypeCheckSuite::{p.name}"
-            args = [sys.executable, "-m", "pytest", "-n", "0", "-s", test_nodeid]
-            proc = subprocess.run(args, cwd=p_root, capture_output=True, check=False)
-            return proc.stdout.decode()
-        finally:
-            p.unlink()
 
+class ParseTestDataSuite(Suite):
     def test_parse_invalid_case(self) -> None:
-        # Arrange
-        data = self._dedent(
+        # Act
+        result = _run_pytest(
             """
             [case abc]
             s: str
@@ -41,15 +22,12 @@ def test_parse_invalid_case(self) -> None:
             """
         )
 
-        # Act
-        actual = self._run_pytest(data)
-
         # Assert
-        assert "Invalid testcase id 'foo-XFAIL'" in actual
+        assert "Invalid testcase id 'foo-XFAIL'" in result.stdout
 
     def test_parse_invalid_section(self) -> None:
-        # Arrange
-        data = self._dedent(
+        # Act
+        result = _run_pytest(
             """
             [case abc]
             s: str
@@ -58,19 +36,16 @@ def test_parse_invalid_section(self) -> None:
             """
         )
 
-        # Act
-        actual = self._run_pytest(data)
-
         # Assert
-        expected_lineno = data.splitlines().index("[unknownsection]") + 1
+        expected_lineno = result.input.splitlines().index("[unknownsection]") + 1
         expected = (
             f".test:{expected_lineno}: Invalid section header [unknownsection] in case 'abc'"
         )
-        assert expected in actual
+        assert expected in result.stdout
 
     def test_bad_ge_version_check(self) -> None:
-        # Arrange
-        data = self._dedent(
+        # Act
+        actual = _run_pytest(
             """
             [case abc]
             s: str
@@ -79,15 +54,12 @@ def test_bad_ge_version_check(self) -> None:
             """
         )
 
-        # Act
-        actual = self._run_pytest(data)
-
         # Assert
-        assert "version>=3.8 always true since minimum runtime version is (3, 8)" in actual
+        assert "version>=3.8 always true since minimum runtime version is (3, 8)" in actual.stdout
 
     def test_bad_eq_version_check(self) -> None:
-        # Arrange
-        data = self._dedent(
+        # Act
+        actual = _run_pytest(
             """
             [case abc]
             s: str
@@ -96,8 +68,5 @@ def test_bad_eq_version_check(self) -> None:
             """
         )
 
-        # Act
-        actual = self._run_pytest(data)
-
         # Assert
-        assert "version==3.7 always false since minimum runtime version is (3, 8)" in actual
+        assert "version==3.7 always false since minimum runtime version is (3, 8)" in actual.stdout

mypy/test/meta/test_update_data.py

@@ -3,47 +3,23 @@
 Updating the expected output, especially when it's in the form of inline (comment) assertions,
 can be brittle, which is why we're "meta-testing" here.
 """
-import shlex
-import subprocess
-import sys
-import textwrap
-import uuid
-from pathlib import Path
-
-from mypy.test.config import test_data_prefix
 from mypy.test.helpers import Suite
+from mypy.test.meta._pytest import PytestResult, dedent_docstring, run_pytest_data_suite
 
 
-class UpdateDataSuite(Suite):
-    def _run_pytest_update_data(self, data_suite: str, *, max_attempts: int) -> str:
-        """
-        Runs a suite of data test cases through 'pytest --update-data' until either tests pass
-        or until a maximum number of attempts (needed for incremental tests).
-        """
-        p_test_data = Path(test_data_prefix)
-        p_root = p_test_data.parent.parent
-        p = p_test_data / f"check-meta-{uuid.uuid4()}.test"
-        assert not p.exists()
-        try:
-            p.write_text(textwrap.dedent(data_suite).lstrip())
-
-            test_nodeid = f"mypy/test/testcheck.py::TypeCheckSuite::{p.name}"
-            args = [sys.executable, "-m", "pytest", "-n", "0", "-s", "--update-data", test_nodeid]
-            cmd = shlex.join(args)
-            for i in range(max_attempts - 1, -1, -1):
-                res = subprocess.run(args, cwd=p_root)
-                if res.returncode == 0:
-                    break
-                print(f"`{cmd}` returned {res.returncode}: {i} attempts remaining")
-
-            return p.read_text()
-        finally:
-            p.unlink()
+def _run_pytest_update_data(data_suite: str) -> PytestResult:
+    """
+    Runs a suite of data test cases through 'pytest --update-data' until either tests pass
+    or until a maximum number of attempts (needed for incremental tests).
+    """
+    return run_pytest_data_suite(data_suite, extra_args=["--update-data"], max_attempts=3)
+
 
+class UpdateDataSuite(Suite):
     def test_update_data(self) -> None:
         # Note: We test multiple testcases rather than 'test case per test case'
         #       so we could also exercise rewriting multiple testcases at once.
-        actual = self._run_pytest_update_data(
+        result = _run_pytest_update_data(
             """
             [case testCorrect]
             s: str = 42  # E: Incompatible types in assignment (expression has type "int", variable has type "str")
@@ -97,12 +73,12 @@ def test_update_data(self) -> None:
             [file b.py]
             s2: str = 43  # E: baz
             [builtins fixtures/list.pyi]
-            """,
-            max_attempts=3,
+            """
         )
 
         # Assert
-        expected = """
+        expected = dedent_docstring(
+            """
         [case testCorrect]
         s: str = 42  # E: Incompatible types in assignment (expression has type "int", variable has type "str")
 
@@ -154,4 +130,5 @@ def test_update_data(self) -> None:
         s2: str = 43  # E: Incompatible types in assignment (expression has type "int", variable has type "str")
         [builtins fixtures/list.pyi]
         """
-        assert actual == textwrap.dedent(expected).lstrip()
+        )
+        assert result.input_updated == expected