Add Instant class to conveniently track elapsed and start/end times

Throughout the codebase we often need to track elapsed times, using a performance clock, and also start/end times that given seconds since epoch (via `time.time()`).

`Instant` encapsulates both functionalities to simplify the code and ensure we are using the correct functions from `_pytest.timing`, which we can mock in tests for reliability.

Author: nicoddemus

src/_pytest/junitxml.py

@@ -636,17 +636,15 @@ def pytest_internalerror(self, excrepr: ExceptionRepr) -> None:
         reporter._add_simple("error", "internal error", str(excrepr))
 
     def pytest_sessionstart(self) -> None:
-        self.suite_start_time = timing.time()
-        self.suite_start_perf = timing.perf_counter()
+        self.suite_start_instant = timing.Instant()
 
     def pytest_sessionfinish(self) -> None:
         dirname = os.path.dirname(os.path.abspath(self.logfile))
         # exist_ok avoids filesystem race conditions between checking path existence and requesting creation
         os.makedirs(dirname, exist_ok=True)
 
         with open(self.logfile, "w", encoding="utf-8") as logfile:
-            suite_stop_perf = timing.perf_counter()
-            suite_time_delta = suite_stop_perf - self.suite_start_perf
+            suite_time_delta = self.suite_start_instant.elapsed_s()
 
             numtests = (
                 self.stats["passed"]
@@ -665,7 +663,9 @@ def pytest_sessionfinish(self) -> None:
                 skipped=str(self.stats["skipped"]),
                 tests=str(numtests),
                 time=f"{suite_time_delta:.3f}",
-                timestamp=datetime.fromtimestamp(self.suite_start_time, timezone.utc)
+                timestamp=datetime.fromtimestamp(
+                    self.suite_start_instant.interval()[0], timezone.utc
+                )
                 .astimezone()
                 .isoformat(),
                 hostname=platform.node(),

src/_pytest/pytester.py

@@ -1150,7 +1150,7 @@ def runpytest_inprocess(
 
         if syspathinsert:
             self.syspathinsert()
-        now = timing.perf_counter()
+        now = timing.Instant()
         capture = _get_multicapture("sys")
         capture.start_capturing()
         try:
@@ -1179,9 +1179,7 @@ class reprec:  # type: ignore
             sys.stderr.write(err)
 
         assert reprec.ret is not None
-        res = RunResult(
-            reprec.ret, out.splitlines(), err.splitlines(), timing.perf_counter() - now
-        )
+        res = RunResult(reprec.ret, out.splitlines(), err.splitlines(), now.elapsed_s())
         res.reprec = reprec  # type: ignore
         return res
 
@@ -1408,7 +1406,7 @@ def run(
         print("     in:", Path.cwd())
 
         with p1.open("w", encoding="utf8") as f1, p2.open("w", encoding="utf8") as f2:
-            now = timing.perf_counter()
+            now = timing.Instant()
             popen = self.popen(
                 cmdargs,
                 stdin=stdin,
@@ -1445,7 +1443,7 @@ def handle_timeout() -> None:
 
         with contextlib.suppress(ValueError):
             ret = ExitCode(ret)
-        return RunResult(ret, out, err, timing.perf_counter() - now)
+        return RunResult(ret, out, err, now.elapsed_s())
 
     def _dump_lines(self, lines, fp):
         try:

src/_pytest/runner.py

@@ -339,19 +339,16 @@ def from_call(
             function, instead of being wrapped in the CallInfo.
         """
         excinfo = None
-        start = timing.time()
-        precise_start = timing.perf_counter()
+        instant = timing.Instant()
         try:
             result: TResult | None = func()
         except BaseException:
             excinfo = ExceptionInfo.from_current()
             if reraise is not None and isinstance(excinfo.value, reraise):
                 raise
             result = None
-        # use the perf counter
-        precise_stop = timing.perf_counter()
-        duration = precise_stop - precise_start
-        stop = timing.time()
+        duration = instant.elapsed_s()
+        start, stop = instant.interval()
         return cls(
             start=start,
             stop=stop,

src/_pytest/terminal.py

@@ -391,7 +391,7 @@ def __init__(self, config: Config, file: TextIO | None = None) -> None:
         self._progress_nodeids_reported: set[str] = set()
         self._timing_nodeids_reported: set[str] = set()
         self._show_progress_info = self._determine_show_progress_info()
-        self._collect_report_last_write: float | None = None
+        self._collect_report_last_instant = timing.Instant()
         self._already_displayed_warnings: int | None = None
         self._keyboardinterrupt_memo: ExceptionRepr | None = None
 
@@ -769,7 +769,6 @@ def pytest_collection(self) -> None:
         if self.isatty:
             if self.config.option.verbose >= 0:
                 self.write("collecting ... ", flush=True, bold=True)
-                self._collect_report_last_write = timing.perf_counter()
         elif self.config.option.verbose >= 1:
             self.write("collecting ... ", flush=True, bold=True)
 
@@ -788,14 +787,13 @@ def report_collect(self, final: bool = False) -> None:
             return
 
         if not final:
-            # Only write "collecting" report every 0.5s.
-            t = timing.perf_counter()
+            # Only write the "collecting" report every `REPORT_COLLECTING_RESOLUTION`.
             if (
-                self._collect_report_last_write is not None
-                and self._collect_report_last_write > t - REPORT_COLLECTING_RESOLUTION
+                self._collect_report_last_instant.elapsed_s()
+                < REPORT_COLLECTING_RESOLUTION
             ):
                 return
-            self._collect_report_last_write = t
+            self._collect_report_last_instant = timing.Instant()
 
         errors = len(self.stats.get("error", []))
         skipped = len(self.stats.get("skipped", []))
@@ -823,7 +821,7 @@ def report_collect(self, final: bool = False) -> None:
     @hookimpl(trylast=True)
     def pytest_sessionstart(self, session: Session) -> None:
         self._session = session
-        self._sessionstarttime = timing.perf_counter()
+        self._session_start_instant = timing.Instant()
         if not self.showheader:
             return
         self.write_sep("=", "test session starts", bold=True)
@@ -1202,7 +1200,7 @@ def summary_stats(self) -> None:
         if self.verbosity < -1:
             return
 
-        session_duration = timing.perf_counter() - self._sessionstarttime
+        session_duration = self._session_start_instant.elapsed_s()
         (parts, main_color) = self.build_summary_stats_line()
         line_parts = []
 

src/_pytest/timing.py

@@ -20,6 +20,38 @@
     from pytest import MonkeyPatch
 
 
+class Instant:
+    """
+    Provides a measurement of timing between different points in the code.
+
+    Useful to compute the elapsed time between two points in time,
+    using a performance counter, and the beginning/end in actual times (as seconds since epoch).
+
+    Inspired by Rust's `std::time::Instant`.
+    """
+
+    def __init__(self) -> None:
+        import _pytest.timing  # noqa: PLW0406
+
+        self._start_perf = _pytest.timing.perf_counter()
+        self._start_time = _pytest.timing.time()
+
+    def elapsed_s(self) -> float:
+        """Return the elapsed time (in seconds) since this Instant was created, using a precise clock."""
+        import _pytest.timing  # noqa: PLW0406
+
+        return _pytest.timing.perf_counter() - self._start_perf
+
+    def interval(self) -> tuple[float, float]:
+        """Return the beginning and end times of this instant, in seconds since epoch, as provided by time.time()."""
+        import _pytest.timing  # noqa: PLW0406
+
+        return self._start_time, _pytest.timing.time()
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"Instant(start_time={self._start_time})"
+
+
 @dataclasses.dataclass
 class MockTiming:
     """Mocks _pytest.timing with a known object that can be used to control timing in tests

testing/_py/test_local.py

@@ -747,7 +747,7 @@ def test_setmtime(self):
             name = tempfile.mktemp()
             open(name, "w").close()
         try:
-            mtime = int(_pytest.timing.perf_counter()) - 100
+            mtime = int(_pytest.timing.time()) - 100
             path = local(name)
             assert path.mtime() != mtime
             path.setmtime(mtime)
@@ -1405,15 +1405,15 @@ def test_atime(self, tmpdir):
         import time
 
         path = tmpdir.ensure("samplefile")
-        now = _pytest.timing.perf_counter()
+        now = _pytest.timing.Instant()
         atime1 = path.atime()
         # we could wait here but timer resolution is very
         # system dependent
         path.read_binary()
         time.sleep(ATIME_RESOLUTION)
         atime2 = path.atime()
         time.sleep(ATIME_RESOLUTION)
-        duration = _pytest.timing.perf_counter() - now
+        duration = now.elapsed_s()
         assert (atime2 - atime1) <= duration
 
     def test_commondir(self, path1):

testing/test_pytester.py

@@ -451,10 +451,9 @@ def test_pytester_run_with_timeout(pytester: Pytester) -> None:
 
     timeout = 120
 
-    start = _pytest.timing.perf_counter()
+    instant = _pytest.timing.Instant()
     result = pytester.runpytest_subprocess(testfile, timeout=timeout)
-    end = _pytest.timing.perf_counter()
-    duration = end - start
+    duration = instant.elapsed_s()
 
     assert result.ret == ExitCode.OK
     assert duration < timeout