Add option to launch each test in separate process

Either specify directly on command line, or add
tool.robotpy.pyfrc.multiprocess to pyproject.toml.

If it works out, expect to make this default in 2026

Author: virtuald

pyfrc/mains/cli_test.py

@@ -7,6 +7,7 @@
 
 import wpilib
 
+import tomli
 import pytest
 
 from ..util import yesno
@@ -37,6 +38,12 @@ def __init__(self, parser=None):
                 action="store_true",
                 help="Use pyfrc's builtin tests if no tests are specified",
             )
+            parser.add_argument(
+                "--isolated",
+                default=None,
+                action="store_true",
+                help="Run each test in a separate robot process. Set `tool.robotpy.pyfrc.isolated` to true in your pyproject.toml to enable by default",
+            )
             parser.add_argument(
                 "--coverage-mode",
                 default=False,
@@ -55,16 +62,41 @@ def run(
         project_path: pathlib.Path,
         robot_class: typing.Type[wpilib.RobotBase],
         builtin: bool,
+        isolated: typing.Optional[bool],
         coverage_mode: bool,
+        verbose: bool,
         pytest_args: typing.List[str],
     ):
+        if isolated is None:
+            pyproject_path = project_path / "pyproject.toml"
+            if pyproject_path.exists():
+                with open(pyproject_path, "rb") as fp:
+                    d = tomli.load(fp)
+
+                try:
+                    v = d["tool"]["robotpy"]["pyfrc"]["isolated"]
+                except KeyError:
+                    pass
+                else:
+                    if not isinstance(v, bool):
+                        raise ValueError(
+                            f"tool.robotpy.pyfrc.isolated must be a boolean value (got {v})"
+                        )
+
+                    isolated = v
+
+        if isolated is None:
+            isolated = False
+
         try:
             return self._run_test(
                 main_file,
                 project_path,
                 robot_class,
                 builtin,
+                isolated,
                 coverage_mode,
+                verbose,
                 pytest_args,
             )
         except _TryAgain:
@@ -73,7 +105,9 @@ def run(
                 project_path,
                 robot_class,
                 builtin,
+                isolated,
                 coverage_mode,
+                verbose,
                 pytest_args,
             )
 
@@ -83,7 +117,9 @@ def _run_test(
         project_path: pathlib.Path,
         robot_class: typing.Type[wpilib.RobotBase],
         builtin: bool,
+        isolated: bool,
         coverage_mode: bool,
+        verbose: bool,
         pytest_args: typing.List[str],
     ):
         # find test directory, change current directory so pytest can find the tests
@@ -92,14 +128,15 @@ def _run_test(
         curdir = pathlib.Path.cwd().absolute()
 
         self.try_dirs = [
-            (project_path / "tests").absolute(),
-            (project_path / ".." / "tests").absolute(),
+            ((project_path / "tests").absolute(), False),
+            ((project_path / ".." / "tests").absolute(), True),
         ]
 
-        for d in self.try_dirs:
+        for d, chdir in self.try_dirs:
             if d.exists():
-                test_directory = d
-                os.chdir(test_directory)
+                builtin = False
+                if chdir:
+                    os.chdir(d)
                 break
         else:
             if not builtin:
@@ -112,10 +149,22 @@ def _run_test(
             pytest_args.insert(0, abspath(inspect.getfile(basic)))
 
         try:
-            retv = pytest.main(
-                pytest_args,
-                plugins=[pytest_plugin.PyFrcPlugin(robot_class, main_file)],
-            )
+            if isolated:
+                from ..test_support import pytest_dist_plugin
+
+                retv = pytest.main(
+                    pytest_args,
+                    plugins=[
+                        pytest_dist_plugin.DistPlugin(
+                            robot_class, main_file, builtin, verbose
+                        )
+                    ],
+                )
+            else:
+                retv = pytest.main(
+                    pytest_args,
+                    plugins=[pytest_plugin.PyFrcPlugin(robot_class, main_file, False)],
+                )
         finally:
             os.chdir(curdir)
 
@@ -132,7 +181,7 @@ def _no_tests(
     ):
         print()
         print("Looked for tests at:")
-        for d in self.try_dirs:
+        for d, _ in self.try_dirs:
             print("-", d)
         print()
         print(

pyfrc/test_support/pytest_dist_plugin.py

@@ -0,0 +1,112 @@
+import multiprocessing
+import pathlib
+import sys
+
+from typing import Type
+
+import pytest
+
+import robotpy.logconfig
+import wpilib
+
+
+from .pytest_plugin import PyFrcPlugin
+
+
+def _run_test(item_nodeid, config_args, robot_class, robot_file, verbose):
+    """This function runs in a subprocess"""
+    robotpy.logconfig.configure_logging(verbose)
+    ec = pytest.main(
+        [item_nodeid, "--no-header", "--no-summary", *config_args],
+        plugins=[PyFrcPlugin(robot_class, robot_file, True)],
+    )
+    sys.exit(ec)
+
+
+def _run_test_in_new_process(
+    test_function, config, robot_class, robot_file, builtin_tests, verbose
+):
+    """Run a test function in a new process."""
+
+    config_args = config.invocation_params.args
+    if builtin_tests:
+        item_nodeid = f"{config_args[0]}::{test_function.name}"
+        config_args = config_args[1:]
+    else:
+        item_nodeid = test_function.nodeid
+
+    process = multiprocessing.Process(
+        target=_run_test,
+        args=(item_nodeid, config_args, robot_class, robot_file, verbose),
+    )
+    process.start()
+    process.join()
+
+    if process.exitcode != 0:
+        pytest.fail(f"Test failed in subprocess: {item_nodeid}", pytrace=False)
+
+
+def _make_runtest(item, config, robot_class, robot_file, builtin_tests, verbose):
+    def isolated_runtest():
+        _run_test_in_new_process(
+            item, config, robot_class, robot_file, builtin_tests, verbose
+        )
+
+    return isolated_runtest
+
+
+class DistPlugin:
+
+    def __init__(
+        self,
+        robot_class: Type[wpilib.RobotBase],
+        robot_file: pathlib.Path,
+        builtin_tests: bool,
+        verbose: bool,
+    ) -> None:
+        self._robot_class = robot_class
+        self._robot_file = robot_file
+        self._builtin_tests = builtin_tests
+        self._verbose = verbose
+
+    @pytest.hookimpl(tryfirst=True)
+    def pytest_collection_modifyitems(
+        self,
+        session: pytest.Session,
+        config: pytest.Config,
+        items: list[pytest.Function],
+    ):
+        """Modify collected test items to run each in a new process."""
+
+        multiprocessing.set_start_method("spawn")
+
+        for item in items:
+            # Overwrite the runtest protocol for each item
+            item.runtest = _make_runtest(
+                item,
+                config,
+                self._robot_class,
+                self._robot_file,
+                self._builtin_tests,
+                self._verbose,
+            )
+
+    #
+    # These fixtures match the ones in PyFrcPlugin but these have no effect
+    #
+
+    @pytest.fixture(scope="function", autouse=True)
+    def robot(self):
+        pass
+
+    @pytest.fixture(scope="function")
+    def control(self, reraise, robot):
+        pass
+
+    @pytest.fixture()
+    def robot_file(self):
+        pass
+
+    @pytest.fixture()
+    def robot_path(self):
+        pass

pyfrc/test_support/pytest_plugin.py

@@ -31,7 +31,14 @@ class PyFrcPlugin:
     be passed to your test function.
     """
 
-    def __init__(self, robot_class: Type[wpilib.RobotBase], robot_file: pathlib.Path):
+    def __init__(
+        self,
+        robot_class: Type[wpilib.RobotBase],
+        robot_file: pathlib.Path,
+        isolated: bool,
+    ):
+        self.isolated = isolated
+
         # attach physics
         physics, robot_class = PhysicsInterface._create_and_attach(
             robot_class,
@@ -103,6 +110,10 @@ def robot(self):
         # Tests only get a proxy to ensure cleanup is more reliable
         yield weakref.proxy(robot)
 
+        # If running in separate processes, no need to do cleanup
+        if self.isolated:
+            return
+
         # reset engine to ensure it gets cleaned up too
         # -> might be holding wpilib objects, or the robot
         if self._physics:

setup.cfg

@@ -29,6 +29,7 @@ install_requires =
 
     wpilib>=2025.1.1,<2026
     robotpy-cli~=2024.0
+    tomli
 setup_requires =
     setuptools_scm > 6
 python_requires = >=3.9