Add run_in_pyodide_coverage (#183)

Add `@run_in_pyodide_coverage` which is a variant of `@run_in_pyodide` that includes coverage reports.

Author: hoodmane

CHANGELOG.md

@@ -1,3 +1,8 @@
+## [Unreleased]
+
+- Add `run_in_pyodide_coverage()` function for coverage testing.
+  [#183](https://github.com/pyodide/pytest-pyodide/pull/183)
+
 ## [0.59.0] - 2026-02-27
 
 ### Changed

pyproject.toml

@@ -32,6 +32,7 @@ test = [
   "build",
   "requests",
   "selenium<4.21.0",
+  "coverage",
 ]
 
 # pytest will look up `pytest11` entrypoints to find plugins

pytest_pyodide/_decorator_in_pyodide.py

@@ -20,6 +20,7 @@
 from base64 import b64decode, b64encode
 from inspect import isclass
 from io import BytesIO
+from pathlib import Path
 from typing import Any
 
 import pyodide_js
@@ -144,4 +145,21 @@ def get_locals(frame):
         return (1, encode(e), repr(e))
 
 
+def start_coverage(coverage_args64):
+    coverage_args = decode(coverage_args64)
+
+    import coverage
+
+    coverage = coverage.Coverage(**coverage_args)
+    coverage.start()
+    return coverage
+
+
+def end_coverage(coverage):
+    coverage.stop()
+    coverage.save()
+    coverage_bytes = Path(".coverage").read_bytes()
+    return b64encode(coverage_bytes).decode()
+
+
 __all__ = ["PyodideHandle", "encode"]

pytest_pyodide/decorator.py

@@ -6,6 +6,9 @@
 from collections.abc import Callable, Collection
 from copy import deepcopy
 from io import BytesIO
+from pathlib import Path
+from secrets import token_hex
+from textwrap import dedent, indent
 from typing import Any, Protocol
 
 from .copy_files_to_pyodide import copy_files_to_emscripten_fs
@@ -371,7 +374,7 @@ def __new__(cls, function: Callable[..., Any] | None = None, /, **kwargs):
             # @run_in_pyodide
             # def f():
             #   pass
-            return run_in_pyodide(**kwargs)(function)
+            return cls(**kwargs)(function)
         # Just do normal __new__ behavior
         return object.__new__(cls)
 
@@ -449,7 +452,8 @@ def _run(self, selenium: SeleniumType, args: tuple[Any, ...]):
             selenium.load_package(self._pkgs)
 
         r = selenium.run_async(code)
-        [status, result, repr] = r
+        [status, result, repr, *extra] = r
+        self._process_extra(*extra)
 
         result = _decode(selenium, result, status, repr)
         if status:
@@ -462,18 +466,25 @@ def _code_template(self, args: tuple[Any, ...]) -> str:
         if the function is async await the result. Last, if there was an
         exception, pickle it and send it back.
         """
+        # Indent by 12 to match the indentation level of the body of __tmp()
+        prelude = indent(self._get_code_prelude(), " " * 12)
+        epilogue = indent(self._get_code_epilogue(), " " * 12)
+
         return f"""
         async def __tmp():
             __tracebackhide__ = True
 
             from pytest_pyodide.decorator import run_in_pyodide_main
-            return run_in_pyodide_main(
+            \n{prelude}
+            result = await run_in_pyodide_main(
                 {_encode(self._mod)!r},
                 {_encode(args)!r},
                 {self._module_filename!r},
                 {self._func_name!r},
                 {self._async_func!r},
             )
+            \n{epilogue}
+            return result
 
         try:
             result = await __tmp()
@@ -482,6 +493,81 @@ async def __tmp():
         result
         """
 
+    # Hooks for run_in_pyodide_coverage
+
+    def _get_code_prelude(self):
+        return ""
+
+    def _get_code_epilogue(self):
+        return ""
+
+    def _process_extra(self):
+        """Hook to handle any extra data computed.
+
+        _get_code_epilogue can add extra data to "result". That extra data will
+        be passed to this function. See the overload in run_in_pyodide_coverage
+        for how this is used.
+        """
+        pass
+
+
+# * coverage_count ensures that multiple files from one test run don't overwrite
+#   each other.
+# * coverage_hex prevents us from overwriting files generated by other test
+#   runs.
+_COVERAGE_COUNT = 1
+_COVERAGE_HEX = token_hex(20)
+
+
+def _get_coverage_path() -> Path:
+    global _COVERAGE_COUNT
+    count = _COVERAGE_COUNT
+    _COVERAGE_COUNT += 1
+    return Path(f".coverage.emscripten.{count}.{_COVERAGE_HEX}")
+
+
+class run_in_pyodide_coverage(run_in_pyodide):
+    def __init__(
+        self,
+        packages: Collection[str] = (),
+        pytest_assert_rewrites: bool = True,
+        *,
+        _force_assert_rewrites: bool = False,
+        coverage_args: dict[str, Any] | None = None,
+    ):
+        super().__init__(
+            packages,
+            pytest_assert_rewrites,
+            _force_assert_rewrites=_force_assert_rewrites,
+        )
+        self._pkgs.append("coverage")
+        if coverage_args is None:
+            coverage_args = {}
+        self._coverage_args = coverage_args
+
+    def _get_code_prelude(self):
+        """Start coverage with the coverage_args passed from the host"""
+        return dedent(
+            f"""
+            from pytest_pyodide.decorator import start_coverage
+            coverage = start_coverage({_encode(self._coverage_args)!r})
+            """
+        )
+
+    def _get_code_epilogue(self):
+        """Stop coverage and append the data to the result"""
+        return dedent(
+            """
+            from pytest_pyodide.decorator import end_coverage
+            coverage_outdata = end_coverage(coverage)
+            result = (*result, coverage_outdata)
+            """
+        )
+
+    def _process_extra(self, coverage_out_binary):  # type:ignore[override]
+        """Write coverage data to the file system"""
+        _get_coverage_path().write_bytes(b64decode(coverage_out_binary))
+
 
 def copy_files_to_pyodide(file_list, install_wheels=True, recurse_directories=True):
     """A decorator that copies files across to pyodide"""

tests/coverage-test/dummy_pkg/__init__.py

@@ -0,0 +1,30 @@
+"""Dummy math utilities for exercising run_in_pyodide_coverage."""
+
+
+def add(a, b):
+    return a + b
+
+
+def multiply(a, b):
+    return a * b
+
+
+def factorial(n):
+    if n < 0:
+        raise ValueError("factorial is not defined for negative numbers")
+    result = 1
+    for i in range(2, n + 1):
+        result *= i
+    return result
+
+
+def never_called():
+    # This function is intentionally never called by the tests so that the
+    # coverage report will flag it as uncovered.
+    return "this line should be reported as not covered"
+
+
+def unreachable():  # pragma: no cover
+    # This function should never be called by the tests and is excluded
+    # from coverage reporting via the pragma above.
+    raise RuntimeError("this function should never run")

tests/coverage-test/pyproject.toml

@@ -0,0 +1,19 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "dummy-pkg"
+version = "0.1.0"
+description = "Dummy package for exercising run_in_pyodide_coverage"
+requires-python = ">=3.10"
+
+[tool.hatch.build.targets.wheel]
+packages = ["dummy_pkg"]
+
+
+[tool.coverage.paths]
+source = [
+  "{COVERAGE_TEST_PATH}",
+  "/lib/python*/site-packages",
+]

tests/test_decorator.py

@@ -167,22 +167,19 @@ def test_selenium4(selenium_standalone):
     ]
 
 
-def test_local_fail_load_package(selenium_standalone):
+def test_local_fail_load_package(selenium_standalone, monkeypatch):
     selenium = selenium_standalone
 
-    def _load_package_error(*args, **kwargs):
+    def load_package_error(*args, **kwargs):
         raise OSError("STOP!")
 
-    _load_package_original = selenium.load_package
-    selenium.load_package = _load_package_error
+    monkeypatch.setattr(selenium, "load_package", load_package_error)
 
     exc = None
     try:
         example_func(selenium)
     except OSError:
         exc = pytest.ExceptionInfo.from_current()
-    finally:
-        selenium.load_package = _load_package_original
 
     assert exc
     try:

tests/test_run_in_pyodide_coverage.py

@@ -0,0 +1,124 @@
+"""End-to-end test for ``run_in_pyodide_coverage``.
+
+Builds the dummy-pkg wheel from ``tests/coverage-test``, installs it into the
+selenium pyodide environment, then calls ``add``, ``multiply`` and
+``factorial`` each inside their own ``run_in_pyodide_coverage`` block. Finally,
+the emitted ``.coverage.emscripten.*`` files are combined and a coverage
+report is produced, which we assert contains the expected covered /
+not-covered lines (and that the ``# pragma: no cover`` line is excluded).
+"""
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+from pprint import pprint
+
+from pytest_pyodide.decorator import run_in_pyodide_coverage
+from pytest_pyodide.server import spawn_web_server
+
+COVERAGE_TEST_DIR = Path(__file__).parent / "coverage-test"
+DUMMY_PKG_FILE = COVERAGE_TEST_DIR / "dummy_pkg/__init__.py"
+COVERAGE_TEST_PYPROJECT_TOML = COVERAGE_TEST_DIR / "pyproject.toml"
+
+
+def _build_wheel(outdir: Path) -> Path:
+    """Build the dummy-pkg wheel into ``outdir`` and return its path."""
+    subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "build",
+            "--wheel",
+            "--outdir",
+            str(outdir),
+            str(COVERAGE_TEST_DIR),
+        ],
+        check=True,
+    )
+    wheels = list(outdir.glob("dummy_pkg-*.whl"))
+    assert len(wheels) == 1, f"expected one wheel, got {wheels}"
+    return wheels[0]
+
+
+def test_run_in_pyodide_coverage(selenium, tmp_path, monkeypatch):
+    # tmp_path = Path("/home/hood/Documents/programming/pytest-pyodide/tests/tmp_path")
+    # Build the dummy-pkg wheel and install it into pyodide.
+    wheel = _build_wheel(tmp_path)
+    with spawn_web_server(tmp_path) as server:
+        hostname, port, _ = server
+        selenium.load_package(f"http://{hostname}:{port}/{wheel.name}")
+
+    # Run everything from tmp_path so the `.coverage.emscripten.*` files
+    # created by run_in_pyodide_coverage land in an isolated directory.
+    monkeypatch.chdir(tmp_path)
+
+    @run_in_pyodide_coverage(coverage_args={"source_pkgs": ["dummy_pkg"]})
+    def exercise_add(selenium):
+        from dummy_pkg import add
+
+        assert add(2, 3) == 5
+
+    @run_in_pyodide_coverage(coverage_args={"source_pkgs": ["dummy_pkg"]})
+    def exercise_multiply(selenium):
+        from dummy_pkg import multiply
+
+        assert multiply(4, 5) == 20
+
+    @run_in_pyodide_coverage(coverage_args={"source_pkgs": ["dummy_pkg"]})
+    def exercise_factorial(selenium):
+        from dummy_pkg import factorial
+
+        assert factorial(5) == 120
+
+    exercise_add(selenium)
+    exercise_multiply(selenium)
+    exercise_factorial(selenium)
+
+    # Make sure coverage data files were actually written.
+    coverage_files = list(tmp_path.glob(".coverage.emscripten.*"))
+    assert len(coverage_files) == 3
+    pyproject_toml = COVERAGE_TEST_PYPROJECT_TOML.read_text()
+    pyproject_toml = pyproject_toml.format(
+        COVERAGE_TEST_PATH=str(COVERAGE_TEST_DIR.absolute())
+    )
+    (tmp_path / "pyproject.toml").write_text(pyproject_toml)
+
+    # Combine the per-run files into a single .coverage database.
+    subprocess.run(
+        [sys.executable, "-m", "coverage", "combine"],
+        check=True,
+        cwd=tmp_path,
+    )
+
+    # Produce a report restricted to dummy_pkg.
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "coverage",
+            "json",
+        ],
+        cwd=tmp_path,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    assert result.returncode == 0
+    assert result.stdout == "Wrote JSON report to coverage.json\n"
+
+    coverage_json = json.loads((tmp_path / "coverage.json").read_text())
+    file = coverage_json["files"][str(DUMMY_PKG_FILE.absolute())]
+    functions = file["functions"]
+    pprint(functions)
+
+    assert functions["add"]["missing_lines"] == []
+    assert functions["add"]["excluded_lines"] == []
+    assert functions["multiply"]["missing_lines"] == []
+    assert functions["multiply"]["excluded_lines"] == []
+    assert functions["factorial"]["missing_lines"] == [14]
+    assert functions["factorial"]["excluded_lines"] == []
+    assert functions["never_called"]["missing_lines"] == [24]
+    assert functions["never_called"]["excluded_lines"] == []
+    assert functions["unreachable"]["missing_lines"] == []
+    assert functions["unreachable"]["excluded_lines"] == [30]