Fix stdout/stderr redirects

src/pytest_codspeed/_wrapper/.gitignore

@@ -1 +1,2 @@
 dist_callgrind_wrapper.*
+build.lock

src/pytest_codspeed/plugin.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import functools
 import gc
 import os
 import pkgutil
@@ -16,11 +17,12 @@
 from ._wrapper import get_lib
 
 if TYPE_CHECKING:
-    from typing import Any, Callable, TypeVar
+    from typing import Any, Callable, Iterator, ParamSpec, TypeVar
 
     from ._wrapper import LibType
 
     T = TypeVar("T")
+    P = ParamSpec("P")
 
 IS_PYTEST_BENCHMARK_INSTALLED = pkgutil.find_loader("pytest_benchmark") is not None
 SUPPORTS_PERF_TRAMPOLINE = sys.version_info >= (3, 12)
@@ -170,88 +172,58 @@ def pytest_collection_modifyitems(
         items[:] = selected
 
 
-def _run_with_instrumentation(
+def wrap_pyfunc_with_instrumentation(
     lib: LibType,
-    nodeId: str,
+    nodeid: str,
     config: pytest.Config,
-    fn: Callable[..., Any],
-    *args,
-    **kwargs,
-):
-    is_gc_enabled = gc.isenabled()
-    if is_gc_enabled:
-        gc.collect()
-        gc.disable()
-
-    result = None
-
-    def __codspeed_root_frame__():
-        nonlocal result
-        result = fn(*args, **kwargs)
-
-    if SUPPORTS_PERF_TRAMPOLINE:
-        # Warmup CPython performance map cache
-        __codspeed_root_frame__()
-    lib.zero_stats()
-    lib.start_instrumentation()
-    __codspeed_root_frame__()
-    lib.stop_instrumentation()
-    uri = get_git_relative_uri(nodeId, config.rootpath)
-    lib.dump_stats_at(uri.encode("ascii"))
-    if is_gc_enabled:
-        gc.enable()
-
-    return result
-
-
-@pytest.hookimpl(tryfirst=True)
-def pytest_runtest_protocol(item: pytest.Item, nextitem: pytest.Item | None):
-    plugin = get_plugin(item.config)
-    if not plugin.is_codspeed_enabled or not should_benchmark_item(item):
-        return (
-            None  # Defer to the default test protocol since no benchmarking is needed
-        )
-
-    if has_benchmark_fixture(item):
-        return None  # Instrumentation is handled by the fixture
-
-    plugin.benchmark_count += 1
-    if not plugin.should_measure:
-        return None  # Benchmark counted but will be run in the default protocol
-
-    # Setup phase
-    reports = []
-    ihook = item.ihook
-    ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
-    setup_call = pytest.CallInfo.from_call(
-        lambda: ihook.pytest_runtest_setup(item=item, nextitem=nextitem), "setup"
-    )
-    setup_report = ihook.pytest_runtest_makereport(item=item, call=setup_call)
-    ihook.pytest_runtest_logreport(report=setup_report)
-    reports.append(setup_report)
-    # Run phase
-    if setup_report.passed and not item.config.getoption("setuponly"):
-        assert plugin.lib is not None
-        runtest_call = pytest.CallInfo.from_call(
-            lambda: _run_with_instrumentation(
-                plugin.lib, item.nodeid, item.config, item.runtest
-            ),
-            "call",
-        )
-        runtest_report = ihook.pytest_runtest_makereport(item=item, call=runtest_call)
-        ihook.pytest_runtest_logreport(report=runtest_report)
-        reports.append(runtest_report)
-
-    # Teardown phase
-    teardown_call = pytest.CallInfo.from_call(
-        lambda: ihook.pytest_runtest_teardown(item=item, nextitem=nextitem), "teardown"
-    )
-    teardown_report = ihook.pytest_runtest_makereport(item=item, call=teardown_call)
-    ihook.pytest_runtest_logreport(report=teardown_report)
-    reports.append(teardown_report)
-    ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)
-
-    return reports  # Deny further protocol hooks execution
+    testfunction: Callable[P, T],
+) -> Callable[P, T]:
+    @functools.wraps(testfunction)
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+        def __codspeed_root_frame__():
+            return testfunction(*args, **kwargs)
+
+        is_gc_enabled = gc.isenabled()
+        if is_gc_enabled:
+            gc.collect()
+            gc.disable()
+        try:
+            if SUPPORTS_PERF_TRAMPOLINE:
+                # Warmup CPython performance map cache
+                __codspeed_root_frame__()
+
+            lib.zero_stats()
+            lib.start_instrumentation()
+            try:
+                return __codspeed_root_frame__()
+            finally:
+                lib.stop_instrumentation()
+                uri = get_git_relative_uri(nodeid, config.rootpath)
+                lib.dump_stats_at(uri.encode("ascii"))
+        finally:
+            if is_gc_enabled:
+                gc.enable()
+
+    return wrapper
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_pyfunc_call(pyfuncitem: pytest.Function) -> Iterator[None]:
+    plugin = get_plugin(pyfuncitem.config)
+    if (
+        plugin.is_codspeed_enabled
+        and should_benchmark_item(pyfuncitem)
+        and not has_benchmark_fixture(pyfuncitem)
+    ):
+        plugin.benchmark_count += 1
+        if plugin.lib is not None and plugin.should_measure:
+            pyfuncitem.obj = wrap_pyfunc_with_instrumentation(
+                plugin.lib,
+                pyfuncitem.nodeid,
+                pyfuncitem.config,
+                pyfuncitem.obj,
+            )
+    yield
 
 
 class BenchmarkFixture:
@@ -266,11 +238,17 @@ def __call__(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
         config = self._request.config
         plugin = get_plugin(config)
         plugin.benchmark_count += 1
-        if plugin.is_codspeed_enabled and plugin.should_measure:
-            assert plugin.lib is not None
-            return _run_with_instrumentation(
-                plugin.lib, self._request.node.nodeid, config, func, *args, **kwargs
-            )
+        if (
+            plugin.is_codspeed_enabled
+            and plugin.lib is not None
+            and plugin.should_measure
+        ):
+            return wrap_pyfunc_with_instrumentation(
+                plugin.lib,
+                self._request.node.nodeid,
+                config,
+                func,
+            )(*args, **kwargs)
         else:
             return func(*args, **kwargs)
 

tests/test_pytest_plugin.py

@@ -306,8 +306,10 @@ def fixtured_child():
 
     with open(perf_filepath) as perf_file:
         lines = perf_file.readlines()
+
         assert any(
-            "py::_run_with_instrumentation.<locals>.__codspeed_root_frame__" in line
+            "py::wrap_pyfunc_with_instrumentation.<locals>.wrapper.<locals>.__codspeed_root_frame__"
+            in line
             for line in lines
         ), "No root frame found in perf map"
         assert any(
@@ -346,3 +348,52 @@ def test_my_stuff(benchmark, i):
             result = pytester.runpytest("--codspeed", "-n", "128")
         assert result.ret == 0, "the run should have succeeded"
         result.stdout.fnmatch_lines(["*256 passed*"])
+
+
+@skip_without_valgrind
+def test_print(pytester: pytest.Pytester, codspeed_env) -> None:
+    """Test print statements are captured by pytest (i.e., not printed to terminal in
+    the middle of the progress bar) and only displayed after test run (on failures)."""
+    pytester.makepyfile(
+        """
+        import pytest, sys
+
+        @pytest.mark.benchmark
+        def test_print():
+            print("print to stdout")
+            print("print to stderr", file=sys.stderr)
+        """
+    )
+    with codspeed_env():
+        result = pytester.runpytest("--codspeed")
+    assert result.ret == 0, "the run should have succeeded"
+    result.stdout.fnmatch_lines(["*1 benchmarked*"])
+    result.stdout.no_fnmatch_line("*print to stdout*")
+    result.stderr.no_fnmatch_line("*print to stderr*")
+
+
+@skip_without_valgrind
+def test_capsys(pytester: pytest.Pytester, codspeed_env) -> None:
+    """Test print statements are captured by capsys (i.e., not printed to terminal in
+    the middle of the progress bar) and can be inspected within test."""
+    pytester.makepyfile(
+        """
+        import pytest, sys
+
+        @pytest.mark.benchmark
+        def test_capsys(capsys):
+            print("print to stdout")
+            print("print to stderr", file=sys.stderr)
+
+            stdout, stderr = capsys.readouterr()
+
+            assert stdout == "print to stdout\\n"
+            assert stderr == "print to stderr\\n"
+        """
+    )
+    with codspeed_env():
+        result = pytester.runpytest("--codspeed")
+    assert result.ret == 0, "the run should have succeeded"
+    result.stdout.fnmatch_lines(["*1 benchmarked*"])
+    result.stdout.no_fnmatch_line("*print to stdout*")
+    result.stderr.no_fnmatch_line("*print to stderr*")