Better testing support

Author: ilevkivskyi

mypy/build.py

@@ -234,7 +234,7 @@ def wait_for_worker(status_file: str, timeout: float = 5.0) -> tuple[int, str]:
     sys.exit(2)
 
 
-def start_worker(options_data: str, idx: int) -> subprocess.Popen[bytes]:
+def start_worker(options_data: str, idx: int, env: Mapping[str, str]) -> subprocess.Popen[bytes]:
     status_file = f".mypy_worker.{idx}.json"
     if os.path.isfile(status_file):
         os.unlink(status_file)
@@ -245,7 +245,7 @@ def start_worker(options_data: str, idx: int) -> subprocess.Popen[bytes]:
         f"--status-file={status_file}",
         f'--options-data="{options_data}"',
     ]
-    return subprocess.Popen(command)
+    return subprocess.Popen(command, env=env)
 
 
 def get_worker(idx: int, proc: subprocess.Popen[bytes]) -> WorkerClient:
@@ -268,6 +268,7 @@ def build(
     stdout: TextIO | None = None,
     stderr: TextIO | None = None,
     extra_plugins: Sequence[Plugin] | None = None,
+    worker_env: Mapping[str, str] | None = None,
 ) -> BuildResult:
     """Analyze a program.
 
@@ -311,15 +312,10 @@ def default_flush_errors(
     workers = []
     procs = []
     if options.num_workers > 0:
-        if options.use_builtins_fixtures:
-            os.environ["MYPY_TEST_PREFIX"] = os.path.dirname(os.path.dirname(__file__))
-        if alt_lib_path:
-            os.environ["MYPY_ALT_LIB_PATH"] = alt_lib_path
-
         pickled_options = pickle.dumps(options.snapshot())
         options_data = base64.b64encode(pickled_options).decode()
         for i in range(options.num_workers):
-            procs.append(start_worker(options_data, i))
+            procs.append(start_worker(options_data, i, worker_env or os.environ))
         for i, proc in enumerate(procs):
             workers.append(get_worker(i, proc))
 
@@ -350,6 +346,18 @@ def default_flush_errors(
         flush_errors(None, e.messages, serious)
         e.messages = messages
         raise
+    finally:
+        for worker in workers:
+            try:
+                send(worker.conn, {"final": True})
+            except OSError:
+                pass
+        for worker in workers:
+            worker.conn.close()
+            worker.proc.wait()
+            status_file = f".mypy_worker.{worker.idx}.json"
+            if os.path.isfile(status_file):
+                os.unlink(status_file)
 
 
 def build_inner(
@@ -426,19 +434,6 @@ def build_inner(
             dump_line_checking_stats(options.line_checking_stats, graph)
         return BuildResult(manager, graph)
     finally:
-
-        for worker in workers:
-            try:
-                send(worker.conn, {"final": True})
-            except OSError:
-                pass
-        for worker in workers:
-            worker.conn.close()
-            worker.proc.wait()
-            status_file = f".mypy_worker.{worker.idx}.json"
-            if os.path.isfile(status_file):
-                os.unlink(status_file)
-
         t0 = time.time()
         manager.metastore.commit()
         manager.add_stats(cache_commit_time=time.time() - t0)

mypy/build_worker/worker.py

@@ -10,6 +10,7 @@
 import sys
 import time
 
+from mypy import util
 from mypy.build import (
     SCC,
     BuildManager,
@@ -68,47 +69,27 @@ def main(argv: list[str]) -> None:
     finally:
         server.cleanup()
 
+    if options.fast_exit:
+        util.hard_exit(0)
+
 
 def serve(server: IPCServer, options: Options, errors: Errors, fscache: FileSystemCache) -> None:
     data = receive(server)
     sources = [BuildSource(*st) for st in data["sources"]]
+    manager = setup_worker_manager(sources, options, errors, fscache)
+    if manager is None:
+        return
 
-    data_dir = os.path.dirname(os.path.dirname(__file__))
-    alt_lib_path = os.environ.get("MYPY_ALT_LIB_PATH")
-    search_paths = compute_search_paths(sources, options, data_dir, alt_lib_path)
-
-    source_set = BuildSourceSet(sources)
-    plugin, snapshot = load_plugins(options, errors, sys.stdout, [])
-
-    def flush_errors(filename: str | None, new_messages: list[str], is_serious: bool) -> None:
-        pass
-
-    manager = BuildManager(
-        data_dir,
-        search_paths,
-        ignore_prefix=os.getcwd(),
-        source_set=source_set,
-        reports=None,
-        options=options,
-        version_id=__version__,
-        plugin=plugin,
-        plugins_snapshot=snapshot,
-        errors=errors,
-        error_formatter=None,
-        flush_errors=flush_errors,
-        fscache=fscache,
-        stdout=sys.stdout,
-        stderr=sys.stderr,
-    )
-
-    gc.disable()
+    if platform.python_implementation() == "CPython":
+        gc.disable()
     try:
         graph = load_graph(sources, manager)
     except CompileError:
         return
-    gc.freeze()
-    gc.unfreeze()
-    gc.enable()
+    if platform.python_implementation() == "CPython":
+        gc.freeze()
+        gc.unfreeze()
+        gc.enable()
 
     for id in graph:
         manager.import_map[id] = set(graph[id].dependencies + graph[id].suppressed)
@@ -144,5 +125,40 @@ def flush_errors(filename: str | None, new_messages: list[str], is_serious: bool
         manager.add_stats(total_process_stale_time=time.time() - t0, stale_sccs_processed=1)
 
 
+def setup_worker_manager(
+    sources: list[BuildSource], options: Options, errors: Errors, fscache: FileSystemCache
+) -> BuildManager | None:
+    data_dir = os.path.dirname(os.path.dirname(__file__))
+    alt_lib_path = os.environ.get("MYPY_ALT_LIB_PATH")
+    search_paths = compute_search_paths(sources, options, data_dir, alt_lib_path)
+
+    source_set = BuildSourceSet(sources)
+    try:
+        plugin, snapshot = load_plugins(options, errors, sys.stdout, [])
+    except CompileError:
+        return None
+
+    def flush_errors(filename: str | None, new_messages: list[str], is_serious: bool) -> None:
+        pass
+
+    return BuildManager(
+        data_dir,
+        search_paths,
+        ignore_prefix=os.getcwd(),
+        source_set=source_set,
+        reports=None,
+        options=options,
+        version_id=__version__,
+        plugin=plugin,
+        plugins_snapshot=snapshot,
+        errors=errors,
+        error_formatter=None,
+        flush_errors=flush_errors,
+        fscache=fscache,
+        stdout=sys.stdout,
+        stderr=sys.stderr,
+    )
+
+
 def console_entry() -> None:
     main(sys.argv[1:])

mypy/main.py

@@ -167,7 +167,6 @@ def main(
         # Exit without freeing objects -- it's faster.
         #
         # NOTE: We don't flush all open files on exit (or run other destructors)!
-        # TODO: use this for workers and join?
         util.hard_exit(code)
     elif code:
         sys.exit(code)

mypy/test/helpers.py

@@ -482,3 +482,7 @@ def find_test_files(pattern: str, exclude: list[str] | None = None) -> list[str]
         for path in (pathlib.Path(test_data_prefix).rglob(pattern))
         if path.name not in (exclude or [])
     ]
+
+
+def remove_typevar_ids(a: list[str]) -> list[str]:
+    return [re.sub(r"`-?\d+", "", line) for line in a]

mypy/test/testcheck.py

@@ -22,6 +22,7 @@
     normalize_error_messages,
     parse_options,
     perform_file_operations,
+    remove_typevar_ids,
 )
 from mypy.test.update_data import update_testcase_output
 
@@ -133,6 +134,11 @@ def run_case_once(
         options.use_builtins_fixtures = True
         options.show_traceback = True
 
+        if options.num_workers:
+            options.fixed_format_cache = True
+            if testcase.output_files:
+                raise pytest.skip("Reports are not supported in parallel mode yet")
+
         # Enable some options automatically based on test file name.
         if "columns" in testcase.file:
             options.show_column_numbers = True
@@ -160,12 +166,22 @@ def run_case_once(
             )
 
         plugin_dir = os.path.join(test_data_prefix, "plugins")
-        sys.path.insert(0, plugin_dir)
 
+        worker_env = None
+        if options.num_workers > 0:
+            worker_env = os.environ.copy()
+            root_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+            worker_env["PYTHONPATH"] = os.pathsep.join([root_dir, plugin_dir])
+            worker_env["MYPY_TEST_PREFIX"] = root_dir
+            worker_env["MYPY_ALT_LIB_PATH"] = test_temp_dir
+
+        sys.path.insert(0, plugin_dir)
         res = None
         blocker = False
         try:
-            res = build.build(sources=sources, options=options, alt_lib_path=test_temp_dir)
+            res = build.build(
+                sources=sources, options=options, alt_lib_path=test_temp_dir, worker_env=worker_env
+            )
             a = res.errors
         except CompileError as e:
             a = e.messages
@@ -196,6 +212,9 @@ def run_case_once(
         if output != a and testcase.config.getoption("--update-data", False):
             update_testcase_output(testcase, a, incremental_step=incremental_step)
 
+        if options.num_workers > 0:
+            a = remove_typevar_ids(a)
+            output = remove_typevar_ids(output)
         assert_string_arrays_equal(output, a, msg.format(testcase.file, testcase.line))
 
         if res:
@@ -211,7 +230,8 @@ def run_case_once(
                 for module, target in res.manager.processed_targets
                 if module in testcase.test_modules
             ]
-            if expected is not None:
+            # TODO: check targets in parallel mode (e.g. per SCC).
+            if options.num_workers == 0 and expected is not None:
                 assert_target_equivalence(name, expected, actual)
             if incremental_step > 1:
                 suffix = "" if incremental_step == 2 else str(incremental_step - 1)

test-data/unit/check-ctypes.test

@@ -166,11 +166,11 @@ reveal_type(intarr4(*int_values))  # N: Revealed type is "_ctypes.Array[ctypes.c
 reveal_type(intarr4(*c_int_values))  # N: Revealed type is "_ctypes.Array[ctypes.c_int]"
 reveal_type(intarr6(1, ctypes.c_int(2), *int_values))  # N: Revealed type is "_ctypes.Array[ctypes.c_int]"
 reveal_type(intarr6(1, ctypes.c_int(2), *c_int_values))  # N: Revealed type is "_ctypes.Array[ctypes.c_int]"
-[typing fixtures/typing-medium.pyi]
 
 float_values = [1.0, 2.0, 3.0, 4.0]
-intarr4(*float_values) # E: Array constructor argument 1 of type "List[float]" is not convertible to the array element type "Iterable[c_int]"
+intarr4(*float_values) # E: Array constructor argument 1 of type "list[float]" is not convertible to the array element type "Iterable[c_int]"
 [builtins fixtures/floatdict.pyi]
+[typing fixtures/typing-medium.pyi]
 
 [case testCtypesArrayConstructorKwargs]
 import ctypes

test-data/unit/lib-stub/typing_extensions.pyi

@@ -1,3 +1,4 @@
+import collections
 import typing
 from typing import Any, Callable, Mapping, Iterable, Iterator, NoReturn as NoReturn, Dict, Tuple, Type, Union
 from typing import TYPE_CHECKING as TYPE_CHECKING