feat: record traces of playwright.request API contexts

pytest-playwright-asyncio/pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 dynamic = ["version"]
 dependencies = [
     "playwright>=1.18",
-    "pytest>=6.2.4,<10.0.0",
+    "pytest>=8.2.0,<10.0.0",
     "python-slugify>=6.0.0,<10.0.0",
     "pytest-asyncio>=0.24.0",
 ]

pytest-playwright-asyncio/pytest_playwright_asyncio/pytest_playwright.py

@@ -20,6 +20,7 @@
 import os
 import sys
 import warnings
+import zipfile
 from pathlib import Path
 from typing import (
     Any,
@@ -34,6 +35,7 @@
     Optional,
     Protocol,
     Sequence,
+    Set,
     Tuple,
     Union,
     Pattern,
@@ -52,6 +54,7 @@
 except ImportError:
     _soft_scope = None
 from playwright.async_api import (
+    APIRequestContext,
     Browser,
     BrowserContext,
     BrowserType,
@@ -183,6 +186,8 @@ def pytest_configure(config: Any) -> None:
 
 # Set on a test item by _artifacts_recorder, picked up for the teardown report.
 _ARTIFACTS_RECORDER_ATTR = "_playwright_artifacts_recorder"
+# Node id of the test being set up, run or torn down, see _APIRequestContexts.
+_CURRENT_TEST_NODEID_KEY = pytest.StashKey[str]()
 
 
 # Making test result information available in fixtures
@@ -321,6 +326,7 @@ def _get_skiplist(item: Any, values: List[str], value_name: str) -> List[str]:
 
 
 def pytest_runtest_setup(item: Any) -> None:
+    item.config.stash[_CURRENT_TEST_NODEID_KEY] = item.nodeid
     if not hasattr(item, "callspec"):
         return
     browser_name = item.callspec.params.get("browser_name")
@@ -438,11 +444,19 @@ async def _artifacts_recorder(
     playwright: Playwright,
     pytestconfig: Any,
     _pw_artifacts_folder: tempfile.TemporaryDirectory,
+    _pw_api_request_contexts: Optional["_APIRequestContexts"],
 ) -> AsyncGenerator["ArtifactsRecorder", None]:
     artifacts_recorder = ArtifactsRecorder(
-        pytestconfig, request, output_path, playwright, _pw_artifacts_folder
+        pytestconfig,
+        request,
+        output_path,
+        playwright,
+        _pw_artifacts_folder,
+        _pw_api_request_contexts,
     )
+    await artifacts_recorder.start_api_request_tracing()
     yield artifacts_recorder
+    await artifacts_recorder.stop_api_request_tracing()
     await artifacts_recorder.collect_videos()
     if _test_failed_before_teardown(request.node):
         artifacts_recorder.did_finish_test(failed=True)
@@ -452,9 +466,31 @@ async def _artifacts_recorder(
     setattr(request.node, _ARTIFACTS_RECORDER_ATTR, artifacts_recorder)
 
 
+@pytest.fixture(autouse=True)
+def _pw_trace_api_requests(request: pytest.FixtureRequest) -> None:
+    # Tests that only use playwright.request (no page or context) don't request
+    # the artifacts recorder otherwise, and it records API request traces too.
+    if (
+        "playwright" in request.fixturenames
+        and request.config.getoption("--tracing") != "off"
+    ):
+        request.getfixturevalue("_artifacts_recorder")
+
+
+@pytest.fixture(scope="session")
+def _pw_api_request_contexts(pytestconfig: Any) -> Optional["_APIRequestContexts"]:
+    if pytestconfig.getoption("--tracing") == "off":
+        return None
+    return _APIRequestContexts(pytestconfig)
+
+
 @pytest_asyncio.fixture(scope="session")
-async def playwright() -> AsyncGenerator[Playwright, None]:
+async def playwright(
+    _pw_api_request_contexts: Optional["_APIRequestContexts"],
+) -> AsyncGenerator[Playwright, None]:
     pw = await async_playwright().start()
+    if _pw_api_request_contexts:
+        _pw_api_request_contexts.instrument(pw)
     yield pw
     await pw.stop()
 
@@ -811,8 +847,10 @@ def __init__(
         output_path: str,
         playwright: Playwright,
         pw_artifacts_folder: tempfile.TemporaryDirectory,
+        api_request_contexts: Optional["_APIRequestContexts"] = None,
     ) -> None:
         self._request = request
+        self._api_request_contexts = api_request_contexts
         self._pytestconfig = pytestconfig
         self._playwright = playwright
         self._output_path = output_path
@@ -821,6 +859,8 @@ def __init__(
         self._all_pages: List[Page] = []
         self._screenshots: List[str] = []
         self._traces: List[str] = []
+        # API request contexts that existed before this test, see start_api_request_tracing.
+        self._shared_api_request_contexts: Set[APIRequestContext] = set()
         # (temporary path, file name in the output folder)
         self._videos: List[Tuple[str, str]] = []
         self.finished = False
@@ -943,6 +983,120 @@ async def on_will_close_browser_context(self, context: BrowserContext) -> None:
                 except Error:
                     pass
 
+    async def start_api_request_tracing(self) -> None:
+        contexts = self._api_request_contexts
+        if contexts is None:
+            return
+        contexts.recorder = self
+        # Contexts from before this test, e.g. session-scoped ones: their trace is
+        # only kept when the test used them. Contexts the test creates itself keep
+        # their trace, like browser contexts do.
+        self._shared_api_request_contexts = set(contexts.live)
+        for context in contexts.live:
+            if context not in contexts.tracing_chunk:
+                await context.tracing.start_chunk(
+                    title=slugify(self._request.node.nodeid)
+                )
+                contexts.tracing_chunk.add(context)
+
+    async def stop_api_request_tracing(self) -> None:
+        contexts = self._api_request_contexts
+        if contexts is None:
+            return
+        contexts.recorder = None
+        for context in contexts.live:
+            await self.on_will_dispose_api_request_context(context)
+
+    async def on_will_dispose_api_request_context(
+        self, context: APIRequestContext
+    ) -> None:
+        contexts = self._api_request_contexts
+        if contexts is None or context not in contexts.tracing_chunk:
+            return
+        # Forgotten first: should stop_chunk fail (the error is reported like one
+        # from a browser context), the next start_chunk discards what is still
+        # being recorded.
+        contexts.tracing_chunk.discard(context)
+        trace_path = Path(self._pw_artifacts_folder.name) / _create_guid()
+        await context.tracing.stop_chunk(path=trace_path)
+        if context in self._shared_api_request_contexts and not _trace_has_actions(
+            trace_path
+        ):
+            os.remove(trace_path)
+            return
+        self._traces.append(str(trace_path))
+
+
+class _APIRequestContexts:
+    # Tracks contexts created via playwright.request.new_context() so that, like
+    # browser contexts, their traces are recorded per test.
+
+    def __init__(self, config: Any) -> None:
+        self._config = config
+        self.live: List[APIRequestContext] = []
+        # Contexts with a trace chunk in progress.
+        self.tracing_chunk: Set[APIRequestContext] = set()
+        # Recorder of the running test, if any.
+        self.recorder: Optional[ArtifactsRecorder] = None
+
+    def instrument(self, playwright: Playwright) -> None:
+        # playwright.request is one APIRequest instance for the lifetime of the
+        # Playwright object, so wrapping its method here covers all contexts.
+        original_new_context = playwright.request.new_context
+
+        async def _new_context(*args: Any, **kwargs: Any) -> APIRequestContext:
+            context = await original_new_context(*args, **kwargs)
+            if getattr(context, "tracing", None) is None:
+                # APIRequestContext.tracing requires playwright>=1.60.
+                return context
+            original_dispose = context.dispose
+
+            async def _dispose_wrapper(*args: Any, **kwargs: Any) -> None:
+                if context in self.live:
+                    self.live.remove(context)
+                try:
+                    if self.recorder is not None:
+                        await self.recorder.on_will_dispose_api_request_context(context)
+                    self.tracing_chunk.discard(context)
+                finally:
+                    # Dispose even if saving the trace failed; that error still
+                    # propagates.
+                    await original_dispose(*args, **kwargs)
+
+            context.dispose = _dispose_wrapper
+            # Contexts are usually created by fixtures, before the artifacts
+            # recorder of the test exists, so the title comes from the hook.
+            nodeid = self._config.stash.get(_CURRENT_TEST_NODEID_KEY, None)
+            await context.tracing.start(
+                title=slugify(nodeid) if nodeid else None,
+                screenshots=True,
+                snapshots=True,
+                sources=True,
+            )
+            self.live.append(context)
+            self.tracing_chunk.add(context)
+            return context
+
+        playwright.request.new_context = _new_context
+
+
+def _trace_has_actions(trace_path: Path) -> bool:
+    # Looks for an action ("before" event) in the trace. The format (one JSON
+    # event per line in *.trace) is Playwright's own and may change, in which
+    # case, or on any error, the trace is kept.
+    try:
+        with zipfile.ZipFile(trace_path) as trace:
+            for name in trace.namelist():
+                if not name.endswith(".trace"):
+                    continue
+                with trace.open(name) as events:
+                    for event in events:
+                        if b'"type":"before"' in event:
+                            return True
+        return False
+    except (OSError, zipfile.BadZipFile):
+        return True
+
 
 def _create_guid() -> str:
     return hashlib.sha256(os.urandom(16)).hexdigest()

pytest-playwright/pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 dynamic = ["version"]
 dependencies = [
     "playwright>=1.18",
-    "pytest>=6.2.4,<10.0.0",
+    "pytest>=7.0.0,<10.0.0",
     "python-slugify>=6.0.0,<10.0.0",
 ]