diff --git a/tests/ci/test_windows_update_crash_oracles.py b/tests/ci/test_windows_update_crash_oracles.py index ba49da0b51..8b8e1c1d9f 100644 --- a/tests/ci/test_windows_update_crash_oracles.py +++ b/tests/ci/test_windows_update_crash_oracles.py @@ -1,9 +1,12 @@ """The Windows crash cells' oracles, judged on this OS: a verdict the cells can only reach on a Windows runner must still say pass exactly when the user's update was what the cell claims.""" +import os import subprocess +import time from pathlib import Path +import psutil import pytest import tests.e2e.core.windows_update.test_crash_cells as crash @@ -57,3 +60,15 @@ def test_kill_point_counts_only_an_update_killed_while_running( else: with pytest.raises(AssertionError, match="cell: the update"): crash._kill_when(_Machine(tmp_path), proc, "cell", point, "t") + + +@pytest.mark.parametrize("age,with_ct,live", [ + (60, False, True), # v1 marker inside update_lock's 20-minute ceiling + (1260, False, False), # v1 past the ceiling: the product reads it DEAD (the pid may be reused) + (1260, True, True), # a matching creation time is live at any age +]) +def test_marker_oracle_reads_a_marker_live_exactly_when_update_lock_does(age, with_ct, live): + pid = os.getpid() + ct = f"ct:{psutil.Process(pid).create_time():.3f}\n" if with_ct else "" + text = f"{pid}\n{time.time() - age}\n{ct}" + assert crash._marker_live(text) == (f"owner {pid}" if live else None) diff --git a/tests/e2e/core/windows_update/test_crash_cells.py b/tests/e2e/core/windows_update/test_crash_cells.py index 70e2927d05..9cc0f95e73 100644 --- a/tests/e2e/core/windows_update/test_crash_cells.py +++ b/tests/e2e/core/windows_update/test_crash_cells.py @@ -67,6 +67,7 @@ from pathlib import Path import psutil import pytest +from hermes_cli.update_lock import UPDATE_MARKER_MAX_AGE_SECONDS from tests.e2e.core._pending_fixes import known_failure from tests.e2e.core.windows._helpers import _decode @@ -404,14 +405,17 @@ def _parse_marker(text: str) -> dict: "delegate": delegate, "delegate_ct": delegate_ct} -def _identity_live(pid: int | None, ct: float | None) -> bool: +def _identity_live(pid: int | None, ct: float | None, age: float) -> bool: + """update_lock's C1 verdict for one identity of a marker ``age`` seconds old.""" if not pid or pid <= 0: return False try: proc = psutil.Process(pid) if not proc.is_running(): return False - return ct is None or abs(proc.create_time() - ct) <= CT_TOLERANCE + if ct is None: # v1 / no creation time: the pid may be reused, so only the age ceiling bounds it + return age <= UPDATE_MARKER_MAX_AGE_SECONDS + return abs(proc.create_time() - ct) <= CT_TOLERANCE except psutil.Error: return False @@ -419,9 +423,10 @@ def _identity_live(pid: int | None, ct: float | None) -> bool: def _marker_live(text: str) -> str | None: """Who keeps the marker LIVE (``"owner "`` / ``"delegate "``), or None.""" m = _parse_marker(text) - if _identity_live(m["pid"], m["ct"]): + age = time.time() - (m["started_at"] or 0.0) + if _identity_live(m["pid"], m["ct"], age): return f"owner {m['pid']}" - if _identity_live(m["delegate"], m["delegate_ct"]): + if _identity_live(m["delegate"], m["delegate_ct"], age): return f"delegate {m['delegate']}" return None