Code review requests

Author: nicoddemus

src/_pytest/junitxml.py

@@ -11,8 +11,6 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from datetime import datetime
-from datetime import timezone
 import functools
 import os
 import platform
@@ -636,15 +634,15 @@ def pytest_internalerror(self, excrepr: ExceptionRepr) -> None:
         reporter._add_simple("error", "internal error", str(excrepr))
 
     def pytest_sessionstart(self) -> None:
-        self.suite_start_instant = timing.Instant()
+        self.suite_start = 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_time_delta = self.suite_start_instant.elapsed_s()
+            duration = self.suite_start.duration()
 
             numtests = (
                 self.stats["passed"]
@@ -662,12 +660,8 @@ def pytest_sessionfinish(self) -> None:
                 failures=str(self.stats["failure"]),
                 skipped=str(self.stats["skipped"]),
                 tests=str(numtests),
-                time=f"{suite_time_delta:.3f}",
-                timestamp=datetime.fromtimestamp(
-                    self.suite_start_instant.interval()[0], timezone.utc
-                )
-                .astimezone()
-                .isoformat(),
+                time=f"{duration.elapsed_s:.3f}",
+                timestamp=duration.start_utc().astimezone().isoformat(),
                 hostname=platform.node(),
             )
             global_properties = self._get_global_properties_node()

src/_pytest/pytester.py

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

src/_pytest/runner.py

@@ -347,12 +347,11 @@ def from_call(
             if reraise is not None and isinstance(excinfo.value, reraise):
                 raise
             result = None
-        duration = instant.elapsed_s()
-        start, stop = instant.interval()
+        duration = instant.duration()
         return cls(
-            start=start,
-            stop=stop,
-            duration=duration,
+            start=duration.start,
+            stop=duration.stop,
+            duration=duration.elapsed_s,
             when=when,
             result=result,
             excinfo=excinfo,

src/_pytest/terminal.py

@@ -789,7 +789,7 @@ def report_collect(self, final: bool = False) -> None:
         if not final:
             # Only write the "collecting" report every `REPORT_COLLECTING_RESOLUTION`.
             if (
-                self._collect_report_last_instant.elapsed_s()
+                self._collect_report_last_instant.duration().elapsed_s
                 < REPORT_COLLECTING_RESOLUTION
             ):
                 return
@@ -821,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._session_start_instant = timing.Instant()
+        self._session_start = timing.Instant()
         if not self.showheader:
             return
         self.write_sep("=", "test session starts", bold=True)
@@ -1200,7 +1200,7 @@ def summary_stats(self) -> None:
         if self.verbosity < -1:
             return
 
-        session_duration = self._session_start_instant.elapsed_s()
+        session_duration = self._session_start.duration()
         (parts, main_color) = self.build_summary_stats_line()
         line_parts = []
 
@@ -1215,7 +1215,7 @@ def summary_stats(self) -> None:
         msg = ", ".join(line_parts)
 
         main_markup = {main_color: True}
-        duration = f" in {format_session_duration(session_duration)}"
+        duration = f" in {format_session_duration(session_duration.elapsed_s)}"
         duration_with_markup = self._tw.markup(duration, **main_markup)
         if display_sep:
             fullwidth += len(duration_with_markup) - len(duration)

src/_pytest/timing.py

@@ -10,6 +10,7 @@
 
 import dataclasses
 from datetime import datetime
+from datetime import timezone
 from time import perf_counter
 from time import sleep
 from time import time
@@ -20,36 +21,49 @@
     from pytest import MonkeyPatch
 
 
+@dataclasses.dataclass(frozen=True)
 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).
+    Measures timing between different points in the code.
 
     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}, elapsed_s={self.elapsed_s()})"
+    # Creation time of this instant, using time.time(), to measure actual time.
+    # Use a `lambda` to initialize the default to correctly get the mocked time via `MockTiming`.
+    _start_time: float = dataclasses.field(default_factory=lambda: time())
+
+    # Initial "tick" of the performance counter to measure precise elapsed time.
+    # Use a `lambda` to initialize the default to correctly get the mocked time via `MockTiming`.
+    _start_perf: float = dataclasses.field(default_factory=lambda: perf_counter())
+
+    def duration(self) -> Duration:
+        """Measure the duration since `Instant` was created."""
+        return Duration(
+            start=self._start_time,
+            elapsed_s=perf_counter() - self._start_perf,
+            stop=time(),
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class Duration:
+    """A slice of time as measured by `Instant.duration()`."""
+
+    # Start time of the duration, as seconds since epoch.
+    start: float
+    # Stop time of the duration, as seconds since epoch.
+    stop: float
+    # Elapsed time of the duration, in seconds, measured using a performance counter for precise timing.
+    elapsed_s: float
+
+    def start_utc(self) -> datetime:
+        """Start time of the duration, as a datetime in UTC."""
+        return datetime.fromtimestamp(self.start, timezone.utc)
+
+    def stop_utc(self) -> datetime:  # pragma: no cover
+        """Stop time of the duration, as a datetime in UTC."""
+        return datetime.fromtimestamp(self.stop, timezone.utc)
 
 
 @dataclasses.dataclass

testing/_py/test_local.py

@@ -12,7 +12,6 @@
 from py import error
 from py.path import local
 
-import _pytest.timing
 import pytest
 
 
@@ -747,7 +746,8 @@ def test_setmtime(self):
             name = tempfile.mktemp()
             open(name, "w").close()
         try:
-            mtime = int(_pytest.timing.time()) - 100
+            # Do not use _pytest.timing here, as we do not want time mocking to affect this test.
+            mtime = int(time.time()) - 100
             path = local(name)
             assert path.mtime() != mtime
             path.setmtime(mtime)
@@ -1405,15 +1405,16 @@ def test_atime(self, tmpdir):
         import time
 
         path = tmpdir.ensure("samplefile")
-        now = _pytest.timing.Instant()
+        # Do not use _pytest.timing here, as we do not want time mocking to affect this test.
+        now = time.time()
         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 = now.elapsed_s()
+        duration = time.time() - now
         assert (atime2 - atime1) <= duration
 
     def test_commondir(self, path1):

testing/test_pytester.py

@@ -453,10 +453,10 @@ def test_pytester_run_with_timeout(pytester: Pytester) -> None:
 
     instant = _pytest.timing.Instant()
     result = pytester.runpytest_subprocess(testfile, timeout=timeout)
-    duration = instant.elapsed_s()
+    duration = instant.duration()
 
     assert result.ret == ExitCode.OK
-    assert duration < timeout
+    assert duration.elapsed_s < timeout
 
 
 def test_pytester_run_timeout_expires(pytester: Pytester) -> None: