Allow custom metadata for replay file

CHANGELOG.rst

@@ -5,6 +5,9 @@ UNRELEASED
 
 * Add support for Python 3.13.
 * Dropped support for EOL Python 3.8.
+* Allow cutomization of metadata in replay file (`#78`_).
+
+.. _`#78`: https://github.com/ESSS/pytest-replay/issues/78
 
 
 1.5.3

README.rst

@@ -69,6 +69,38 @@ execute the tests in the same order with::
 
 Hopefully this will make it easier to reproduce the problem and fix it.
 
+Additional metadata
+-------------------
+
+In cases where it is necessary to add new metadata to the replay file to make the test reproducible, `pytest-replay`
+provides a fixture called ``replay_metadata`` that allows new information to be added using the ``metadata``
+attribute.
+
+Example:
+
+.. code-block:: python
+
+    import pytest
+    import numpy as np
+    import random
+
+    @pytest.fixture
+    def rng(replay_metadata):
+        seed = replay_metadata.metadata.setdefault("seed", random.randint(0, 100))
+        return np.random.default_rng(seed=seed)
+
+    def test_random(rng):
+        data = rng.standard_normal((100, 100))
+        assert data.shape == (100, 100)
+
+
+When using it with pytest-replay it generates a replay file similar to
+
+.. code-block:: json
+
+    {"nodeid": "test_bar.py::test_random", "start": 0.000}
+    {"nodeid": "test_bar.py::test_random", "start": 0.000, "finish": 1.5, "outcome": "passed", "metadata": {"seed": 12}}
+
 
 FAQ
 ~~~

src/pytest_replay/__init__.py

@@ -1,7 +1,12 @@
+import collections
+import dataclasses
 import json
 import os
 import time
+from dataclasses import asdict
 from glob import glob
+from typing import Any
+from typing import Optional
 
 import pytest
 
@@ -39,6 +44,24 @@ def pytest_addoption(parser):
     )
 
 
+@dataclasses.dataclass
+class ReplayTestMetadata:
+    nodeid: str
+    start: float = 0.0
+    finish: Optional[float] = None
+    outcome: Optional[str] = None
+    metadata: dict[str, Any] = dataclasses.field(default_factory=dict)
+
+    def to_clean_dict(self) -> dict[str, Any]:
+        return {k: v for k, v in asdict(self).items() if v}
+
+
+class _ReplayTestMetadataDefaultDict(collections.defaultdict):
+    def __missing__(self, key):
+        self[key] = ReplayTestMetadata(nodeid=key)
+        return self[key]
+
+
 class ReplayPlugin:
     def __init__(self, config):
         self.dir = config.getoption("replay_record_dir")
@@ -53,10 +76,13 @@ def __init__(self, config):
         skip_cleanup = config.getoption("skip_cleanup", False)
         if not skip_cleanup:
             self.cleanup_scripts()
-        self.node_start_time = {}
-        self.node_outcome = {}
+        self.nodes = _ReplayTestMetadataDefaultDict()
         self.session_start_time = config.replay_start_time
 
+    @pytest.fixture(scope="function")
+    def replay_metadata(self, request):
+        return self.nodes[request.node.nodeid]
+
     def cleanup_scripts(self):
         if self.xdist_worker_name:
             # only cleanup scripts on the master node
@@ -79,31 +105,28 @@ def pytest_runtest_logstart(self, nodeid):
             # only workers report running tests when running in xdist
             return
         if self.dir:
-            self.node_start_time[nodeid] = time.perf_counter() - self.session_start_time
-            json_content = json.dumps(
-                {"nodeid": nodeid, "start": self.node_start_time[nodeid]}
-            )
+            self.nodes[nodeid].start = time.perf_counter() - self.session_start_time
+            json_content = json.dumps(self.nodes[nodeid].to_clean_dict())
             self.append_test_to_script(nodeid, json_content)
 
     @pytest.hookimpl(hookwrapper=True)
     def pytest_runtest_makereport(self, item):
         report = yield
         result = report.get_result()
         if self.dir:
-            current = self.node_outcome.setdefault(item.nodeid, result.outcome)
+            self.nodes[item.nodeid].outcome = (
+                self.nodes[item.nodeid].outcome or result.outcome
+            )
+            current = self.nodes[item.nodeid].outcome
             if not result.passed and current != "failed":
                 # do not overwrite a failed outcome with a skipped one
-                self.node_outcome[item.nodeid] = result.outcome
+                self.nodes[item.nodeid].outcome = result.outcome
 
             if result.when == "teardown":
-                json_content = json.dumps(
-                    {
-                        "nodeid": item.nodeid,
-                        "start": self.node_start_time[item.nodeid],
-                        "finish": time.perf_counter() - self.session_start_time,
-                        "outcome": self.node_outcome.pop(item.nodeid),
-                    }
+                self.nodes[item.nodeid].finish = (
+                    time.perf_counter() - self.session_start_time
                 )
+                json_content = json.dumps(self.nodes[item.nodeid].to_clean_dict())
                 self.append_test_to_script(item.nodeid, json_content)
 
     def pytest_collection_modifyitems(self, items, config):
@@ -119,7 +142,10 @@ def pytest_collection_modifyitems(self, items, config):
                 stripped = line.strip()
                 # Ignore blank linkes and comments. (#70)
                 if stripped and not stripped.startswith(("#", "//")):
-                    nodeid = json.loads(stripped)["nodeid"]
+                    node_metadata = json.loads(stripped)
+                    nodeid = node_metadata["nodeid"]
+                    if "finish" in node_metadata:
+                        self.nodes[nodeid] = ReplayTestMetadata(**node_metadata)
                     nodeids[nodeid] = None
 
         items_dict = {item.nodeid: item for item in items}

tests/test_replay.py

@@ -274,6 +274,68 @@ def test_filter_out_tests_not_in_file(testdir):
     )
 
 
+def test_replay_extra_metafunc(pytester, tmp_path):
+    pytester.makepyfile(
+        """
+        import pytest
+        import random
+
+        @pytest.fixture
+        def extra_metafunc(replay_metadata):
+            assert replay_metadata.metadata == {}
+            rand_int = random.randint(0, 100)
+            replay_metadata.metadata["seed"] = rand_int
+            return rand_int
+
+        @pytest.mark.parametrize('i', range(10))
+        def test_abc(extra_metafunc, i):
+            assert i % 2 == 0
+        """
+    )
+    dir = tmp_path / "replay"
+    result = pytester.runpytest(f"--replay-record-dir={dir}", "-n 2")
+    assert result.ret != 0
+
+    contents = [
+        json.loads(s)
+        for replay_file in (".pytest-replay-gw0.txt", ".pytest-replay-gw1.txt")
+        for s in (dir / replay_file).read_text().splitlines()
+    ]
+    contents.sort(key=lambda x: x["nodeid"])
+    assert contents[1]["metadata"]["seed"] < 100
+    assert (
+        len(
+            {
+                val["metadata"]["seed"]
+                for val in contents
+                if val.get("metadata", {}).get("seed")
+            }
+        )
+        > 1
+    )
+
+
+def test_replay_extra_metadata_load(pytester, tmp_path):
+    pytester.makepyfile(
+        """
+        import pytest
+
+        def test_load(replay_metadata):
+            assert replay_metadata.metadata == {"seed": 1234}
+        """
+    )
+    pytester.maketxtfile(
+        """{\"nodeid\": \"test_replay_extra_metadata_load.py::test_load\", \"start\": 1.0}
+    {\"nodeid\": \"test_replay_extra_metadata_load.py::test_load\", \"start\": 1.0, \"finish\": 2.0, \"outcome\": \"passed\", \"metadata\": {\"seed\": 1234}}
+    """
+    )
+    result = pytester.runpytest(
+        f"--replay={pytester.path / 'test_replay_extra_metadata_load.txt'}"
+    )
+    assert result.ret == 0
+    result.assert_outcomes(passed=1)
+
+
 def test_replay_file_outcome_is_correct(testdir):
     """Tests that the outcomes in the replay file are correct."""
     testdir.makepyfile(