Test tools force multimer's source with MULTIMER_SOURCE

lib/examples/lv_test_timer.py

@@ -421,7 +421,7 @@ def run_kit():
 
 def _wants_kit():
     # Scan the whole command line: under a runner (e.g.
-    # tools/multimer_backend_preload.py) the token is not at a fixed index, and
+    # tools/multimer_source_preload.py) the token is not at a fixed index, and
     # CircuitPython cannot rewrite sys.argv to move it.
     return any(arg in ("kit", "harness") for arg in sys.argv[1:])
 

tests/test_multimer_source.py

@@ -0,0 +1,98 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+"""The test tools force multimer's wake source through MULTIMER_SOURCE."""
+
+import importlib.util
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+import _env
+
+_REPO = Path(__file__).resolve().parent.parent
+_TOOLS = _REPO / "tools"
+_PRELOAD = _TOOLS / "multimer_source_preload.py"
+
+
+def _load(name, filename):
+    if str(_TOOLS) not in sys.path:
+        sys.path.insert(0, str(_TOOLS))
+    spec = importlib.util.spec_from_file_location(name, _TOOLS / filename)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _child_env():
+    env = os.environ.copy()
+    env.pop("MULTIMER_SOURCE", None)
+    env.pop("MULTIMER_BACKEND", None)
+    if _env._HARDWARE_ROOT:
+        lib = os.path.join(_env._HARDWARE_ROOT, "lib")
+        env["PYTHONPATH"] = os.pathsep.join(filter(None, [lib, env.get("PYTHONPATH")]))
+    return env
+
+
+class TestPreload(unittest.TestCase):
+    def _run(self, source):
+        with tempfile.TemporaryDirectory() as tmp:
+            script = os.path.join(tmp, "probe.py")
+            with open(script, "w") as fh:
+                fh.write(
+                    "import multimer\nprint('SOURCE_IN_SCRIPT=' + str(multimer.info()['source']))\n"
+                )
+            return subprocess.run(
+                [sys.executable, str(_PRELOAD), source, script],
+                cwd=str(_REPO / "lib"),
+                env=_child_env(),
+                capture_output=True,
+                text=True,
+                timeout=60,
+                check=False,
+            )
+
+    def test_forced_source_is_the_one_in_use(self):
+        for source in ("pending", "none"):
+            with self.subTest(source=source):
+                proc = self._run(source)
+                self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+                self.assertIn(f"MULTIMER_SOURCE_FORCED={source}", proc.stdout)
+                self.assertIn(f"SOURCE_IN_SCRIPT={source}", proc.stdout)
+
+    def test_source_this_host_lacks_is_unavailable(self):
+        proc = self._run("wasm")
+        self.assertEqual(proc.returncode, 3, proc.stdout + proc.stderr)
+        self.assertIn("MULTIMER_SOURCE_UNAVAILABLE='wasm'", proc.stdout)
+        self.assertNotIn("SOURCE_IN_SCRIPT", proc.stdout)
+
+
+class TestKitAndWrapper(unittest.TestCase):
+    def test_kit_forwards_multimer_source(self):
+        kit = _load("pydevices_example_test_kit", "example_test_kit.py")
+        self.assertEqual(kit.multimer_source_args({}), [])
+        self.assertEqual(
+            kit.multimer_source_args({"MULTIMER_SOURCE": "signal"}),
+            ["--multimer-source", "signal"],
+        )
+
+    def test_kit_refuses_retired_multimer_backend(self):
+        kit = _load("pydevices_example_test_kit", "example_test_kit.py")
+        with self.assertRaises(SystemExit) as ctx:
+            kit.multimer_source_args({"MULTIMER_BACKEND": "sdl2"})
+        self.assertIn("MULTIMER_SOURCE", str(ctx.exception))
+
+    def test_wrapper_takes_multimer_source(self):
+        wrapper = _load("pydevices_example_test_wrapper", "example_test_wrapper.py")
+        base = ["wrapper", "demo", "--script", "x.py", "--kind", "loop"]
+        args = wrapper._parse_args([*base, "--multimer-source", "pending"])
+        self.assertEqual(args["multimer_source"], "pending")
+        with self.assertRaises(ValueError):
+            wrapper._parse_args([*base, "--multimer-backend", "sdl2"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/README.md

@@ -112,7 +112,7 @@ see [Windows PE under WSL](#windows-pe-under-wsl).
 `micropython.exe` and `python.exe` are Windows PE binaries launched from WSL.
 They cannot read Linux-exported environment variables. The kit therefore
 forwards only values that must cross that boundary via wrapper argv +
-`displaydev.env_set` (notably `--timer-async` / `--multimer-backend`).
+`displaydev.env_set` (notably `--timer-async` / `--multimer-source`).
 
 **Do not forward `SDL_VIDEODRIVER` / `SDL_AUDIODRIVER` to PE.** Unix cells stay
 headless from the shell `SDL_*=dummy` export; PE keeps a real Windows video
@@ -189,15 +189,17 @@ Headless needs Playwright (`.venv/bin/pip install -r requirements-dev.txt` and
 | [`run_desktop_lv_tests.py`](run_desktop_lv_tests.py) | LVGL desktop matrix (sync/async, strict clicks) |
 | [`lv_timer_test_kit.py`](lv_timer_test_kit.py) | Full LVGL timer matrix (sync/async, all interpreters) |
 | [`run_test_timers.py`](run_test_timers.py) | Run the sibling core multimer timer probe across desktop interpreters |
-| [`multimer_backend_preload.py`](multimer_backend_preload.py) | Force one multimer backend, then run a script |
-
-**Comparing multimer providers:** `lv_timer_test_kit.py --backend NAME` (or
-`example_test_kit.py` with `MULTIMER_BACKEND` set, which forwards
-`--multimer-backend` to the wrapper). Both set `MULTIMER_BACKEND` inside the
-child before importing `multimer.auto`, so they also work for the Windows
-`.exe` interpreters, which cannot read WSL-exported env vars. Interpreters lacking that
-provider report `unavailable` and do not fail the run. See the
-[multimer automatic-selection documentation](https://github.com/PyDevices/pydevices/blob/main/docs/multimer.md#automatic-selection).
+| [`multimer_source_preload.py`](multimer_source_preload.py) | Force one multimer wake source, then run a script |
+
+**Comparing multimer wake sources:** `lv_timer_test_kit.py --source NAME` (or
+`example_test_kit.py` with `MULTIMER_SOURCE` set, which forwards
+`--multimer-source` to the wrapper). Both set `MULTIMER_SOURCE` inside the
+child and make multimer choose its source before the script runs, so they also
+work for the Windows `.exe` interpreters, which cannot read WSL-exported env
+vars. An interpreter that can't start that source reports `unavailable` and
+doesn't fail the run. `MULTIMER_BACKEND` is retired: the kit refuses to run
+with it set. The sources, and which host uses which, are in pydevices'
+[multimer hosts table](https://github.com/PyDevices/pydevices/blob/main/docs/multimer.md#hosts).
 
 TestPyPI package smoke tests are owned by the repositories that publish the
 packages: core checks live in

tools/example_test_kit.py

@@ -436,6 +436,21 @@ def run_unit_tests() -> int:
     return proc.returncode
 
 
+def multimer_source_args(env) -> list[str]:
+    """Wrapper argv that forces ``MULTIMER_SOURCE`` from ``env``, if set.
+
+    ``MULTIMER_BACKEND`` was retired with pydevices 0.6: nothing reads it, so a
+    run that set it would test the default source while claiming another.
+    """
+    if env.get("MULTIMER_BACKEND"):
+        raise SystemExit(
+            "MULTIMER_BACKEND is retired; set MULTIMER_SOURCE "
+            "(signal, pending, asyncio, machine, wasm, native, none)"
+        )
+    source = env.get("MULTIMER_SOURCE")
+    return ["--multimer-source", str(source)] if source else []
+
+
 def run_subprocess_case(
     interpreter_id: str,
     exe: str,
@@ -479,9 +494,7 @@ def run_subprocess_case(
     timer_async = env.get("PYDEVICES_TIMER_ASYNC")
     if timer_async is not None:
         cmd.extend(["--timer-async", str(timer_async)])
-    multimer_backend = env.get("MULTIMER_BACKEND")
-    if multimer_backend:
-        cmd.extend(["--multimer-backend", str(multimer_backend)])
+    cmd.extend(multimer_source_args(env))
     # Do not forward SDL_* to Windows PE. WSL-exported env is invisible to
     # .exe children (so unix stays headless via the shell export), and PE
     # should keep a real Windows video driver — dummy there hides the brief

tools/example_test_wrapper.py

@@ -357,7 +357,7 @@ def _start_multimer_quit_schedule(duration_s, quit_mode, kind, injected):
     try:
         import quit_inject
 
-        from multimer import auto as timer
+        import multimer as timer
     except ImportError:
         return False
     try:
@@ -549,7 +549,7 @@ def _parse_args(argv):
         "duration": 5.0,
         "timeout": 30.0,
         "timer_async": None,
-        "multimer_backend": None,
+        "multimer_source": None,
         "env": [],
     }
     i = 2
@@ -576,8 +576,8 @@ def _parse_args(argv):
         elif arg == "--timer-async" and i + 1 < len(argv):
             out["timer_async"] = argv[i + 1]
             i += 2
-        elif arg == "--multimer-backend" and i + 1 < len(argv):
-            out["multimer_backend"] = argv[i + 1]
+        elif arg == "--multimer-source" and i + 1 < len(argv):
+            out["multimer_source"] = argv[i + 1]
             i += 2
         elif arg == "--env" and i + 1 < len(argv):
             out["env"].append(argv[i + 1])
@@ -589,6 +589,25 @@ def _parse_args(argv):
     return out
 
 
+def _force_multimer_source(name):
+    """Make multimer select wake source ``name`` now, or raise.
+
+    Same check as ``multimer_source_preload.force_source``.
+    """
+    _env_set("MULTIMER_SOURCE", name)
+    import multimer
+    from multimer import _dispatch
+
+    _dispatch._ensure_source()
+    info = multimer.info()
+    active = info.get("source")
+    if active != name:
+        raise RuntimeError(
+            "multimer chose {!r}: {}".format(active, info.get("source_error", "already selected"))
+        )
+    return active
+
+
 def _subprocess_hard_exit(code, *, headless=False):
     """Exit past SDL teardown, which can block normal interpreter shutdown.
 
@@ -712,24 +731,21 @@ def main(argv=None):
     except Exception:
         pass
 
-    # MULTIMER_BACKEND is the sole auto-provider override. Set it inside the
-    # child because Windows PE launched from WSL cannot see the parent's
-    # exported environment. A provider this host cannot supply is a skip, not
-    # a failure — sweeps ask every interpreter for every provider.
-    if args.get("multimer_backend") is not None:
+    # MULTIMER_SOURCE is multimer's one override. Set it inside the child
+    # because Windows PE launched from WSL cannot see the parent's exported
+    # environment, and select the source now: a forced source that cannot start
+    # falls back to "none" silently. A source this host cannot supply is a
+    # skip, not a failure: sweeps ask every interpreter for every source.
+    if args.get("multimer_source") is not None:
         try:
-            _env_set("MULTIMER_BACKEND", args["multimer_backend"])
-            from multimer import auto as timer
-
-            if timer.name != args["multimer_backend"]:
-                raise RuntimeError("multimer.auto was already selected as {!r}".format(timer.name))
+            _force_multimer_source(args["multimer_source"])
         except (ImportError, RuntimeError, ValueError) as exc:
             _print_result(
                 {
                     "example": args["example"],
                     "status": "skip",
-                    "error": "multimer backend {!r} unavailable: {}".format(
-                        args["multimer_backend"], exc
+                    "error": "multimer source {!r} unavailable: {}".format(
+                        args["multimer_source"], exc
                     ),
                     "backend": "headless" if headless else "?",
                 }

tools/lv_timer_test_kit.py

@@ -13,12 +13,12 @@
 From repo root:
     python tools/lv_timer_test_kit.py
     python tools/lv_timer_test_kit.py --only cpython-venv
-    python tools/lv_timer_test_kit.py --backend sdl2
+    python tools/lv_timer_test_kit.py --source pending
 
-``--backend`` forces one multimer backend through
-``tools/multimer_backend_preload.py`` (in-process, so it also works for the
+``--source`` forces one multimer wake source through
+``tools/multimer_source_preload.py`` (in-process, so it also works for the
 Windows ``.exe`` interpreters, which cannot read WSL-exported env vars). Interpreters
-without that backend report ``unavailable`` and do not fail the run.
+without that source report ``unavailable`` and do not fail the run.
 
 Interpreters resolve via ``tools/example_interpreters.toml`` (same as example_test_kit).
 Missing executables show as ``missing`` in the table.
@@ -165,21 +165,21 @@ def run_case(
     timeout: int = DEFAULT_TIMEOUT,
     *,
     cwd: Path | None = None,
-    backend: str | None = None,
+    source: str | None = None,
 ) -> dict:
     # Every case goes through the preload so its settings are applied in-process:
     # Windows MicroPython / CPython launched from WSL never see exported
     # variables, and a mode that silently no-ops would report a sync run in the
     # async column. The process env is still set for code that reads os.environ.
     timer_async = {"async": "1", "sync": "0"}.get(mode)
-    preload = os.path.relpath(TOOLS / "multimer_backend_preload.py", SRC)
+    preload = os.path.relpath(TOOLS / "multimer_source_preload.py", SRC)
     cmd = [*cmd_base, preload, "--source-workspace"]
     if timer_async is not None:
         cmd += ["--env", f"PYDEVICES_TIMER_ASYNC={timer_async}"]
-    cmd += [backend or "-", HARNESS_ARG, "kit"]
+    cmd += [source or "-", HARNESS_ARG, "kit"]
     env = os.environ.copy()
-    if backend:
-        env["MULTIMER_BACKEND"] = backend
+    if source:
+        env["MULTIMER_SOURCE"] = source
     if timer_async is not None:
         env["PYDEVICES_TIMER_ASYNC"] = timer_async
     run_cwd = str(cwd or SRC)
@@ -207,16 +207,16 @@ def run_case(
 
     result = parse_result(stdout)
     summary = summarize(result, returncode, timed_out)
-    # This host has no such backend; a sweep asks every interpreter for every
-    # backend, so that is a skip rather than a failure. Match on the sentinel,
+    # This host has no such source; a sweep asks every interpreter for every
+    # source, so that is a skip rather than a failure. Match on the sentinel,
     # not the preload exit code: CircuitPython does not propagate sys.exit(3).
-    unavailable = "MULTIMER_BACKEND_UNAVAILABLE" in stdout
+    unavailable = "MULTIMER_SOURCE_UNAVAILABLE" in stdout
     if unavailable:
         summary = "unavailable"
     return {
         "interpreter": interpreter,
         "mode": mode,
-        "backend": backend,
+        "source": source,
         "unavailable": unavailable,
         "summary": summary,
         "returncode": returncode,
@@ -259,7 +259,7 @@ def run_kit(
     strict_clicks: bool = False,
     results_path: Path = DEFAULT_RESULTS,
     emit_json: bool = False,
-    backend: str | None = None,
+    source: str | None = None,
 ) -> int:
     modes_tuple = tuple(modes)
     interpreters = _resolve_interpreters(only)
@@ -272,9 +272,9 @@ def run_kit(
                 print(f"Skipping {name} {mode} (not found: {hint})", file=sys.stderr)
                 rows.append(_missing_row(name, mode, exe_hint=hint))
                 continue
-            label = f"{name} {mode}" + (f" [{backend}]" if backend else "")
+            label = f"{name} {mode}" + (f" [{source}]" if source else "")
             print(f"Running {label}...", file=sys.stderr)
-            row = run_case(name, cmd_base, mode, timeout, backend=backend)
+            row = run_case(name, cmd_base, mode, timeout, source=source)
             rows.append(row)
             if emit_json:
                 print(json.dumps(row, indent=2))
@@ -313,11 +313,12 @@ def main(argv: list[str] | None = None) -> int:
         ),
     )
     parser.add_argument(
-        "--backend",
+        "--source",
         metavar="NAME",
+        choices=("signal", "pending", "asyncio", "machine", "wasm", "native", "none"),
         help=(
-            "Force one multimer backend (machine, librt, win32, sdl2, threading, "
-            "polling, async) instead of the platform default"
+            "Force one multimer wake source (signal, pending, asyncio, machine, "
+            "wasm, native, none) instead of the host's default"
         ),
     )
     parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT)
@@ -335,7 +336,7 @@ def main(argv: list[str] | None = None) -> int:
         timeout=args.timeout,
         strict_clicks=args.strict_clicks,
         emit_json=args.json,
-        backend=args.backend,
+        source=args.source,
     )