From 57a52b1a9933a7093498d988911f90a325615714 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Wed, 22 Jul 2026 05:40:15 -0500 Subject: [PATCH] fix(parity): hermetic live-remote master reads under pytest (Closes #610) Remediate PR #788 review F1/F2: suite-wide hermetic mode prevents git ls-remote from running in tests so feature worktrees no longer flip legacy runtime-context assertions to live_stale, and unit tests stay offline. Module flag survives patch.dict(clear=True); env override still wins. Co-Authored-By: Grok 4.5 --- master_parity_gate.py | 42 ++++++++++++++++++ tests/conftest.py | 29 ++++++++++++ tests/test_master_parity_gate.py | 75 +++++++++++++++++++++++++++++++- 3 files changed, 144 insertions(+), 2 deletions(-) diff --git a/master_parity_gate.py b/master_parity_gate.py index e611387..2e38c66 100644 --- a/master_parity_gate.py +++ b/master_parity_gate.py @@ -32,11 +32,32 @@ import time _REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {} _REMOTE_HEAD_TTL = 60.0 +# When True, ``read_remote_master_head`` never performs ``git ls-remote`` unless +# ``GITEA_TEST_LIVE_REMOTE_HEAD`` is set. Conftest enables this suite-wide so +# feature worktrees (whose HEAD differs from live master) cannot flip legacy +# runtime-context assertions to live_stale, and so unit tests never depend on +# a live network (PR #788 F1/F2 / issue #610). Module-level (not env-only) so +# ``patch.dict(os.environ, …, clear=True)`` cannot re-enable the probe. +_HERMETIC_TEST_MODE: bool = False + def _clear_remote_head_cache() -> None: """Reset the live-remote head cache (test isolation / forced refresh).""" _REMOTE_HEAD_CACHE.clear() + +def set_hermetic_test_mode(enabled: bool) -> None: + """Enable or disable suite-wide hermetic live-remote reads (tests only).""" + global _HERMETIC_TEST_MODE + _HERMETIC_TEST_MODE = bool(enabled) + _clear_remote_head_cache() + + +def hermetic_test_mode() -> bool: + """Return whether hermetic live-remote reads are active.""" + return bool(_HERMETIC_TEST_MODE) + + # Environment escape hatches (ops + tests): # GITEA_MCP_DISABLE_PARITY_GATE -> disable enforcement entirely (fail open). # GITEA_TEST_CURRENT_HEAD -> force the "current" HEAD read, for tests. @@ -44,6 +65,9 @@ ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE" ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD" # GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests. ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD" +# GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE -> opt a single test into a real ls-remote +# even when hermetic mode is on (rare; prefer ENV_TEST_LIVE_REMOTE_HEAD). +ENV_TEST_ALLOW_LIVE_REMOTE_PROBE = "GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE" def read_git_head(root: str) -> str | None: @@ -92,10 +116,28 @@ def read_remote_master_head( gate does not run a network probe on every mutation/read (``ttl=0`` forces a live probe). Both hits and ``None`` misses are cached to bound offline latency; the env override bypasses the cache and the subprocess entirely. + + Under suite hermetic mode (``set_hermetic_test_mode(True)``, set by + conftest) a missing override returns ``None`` without network I/O so + feature-worktree test runs cannot observe live_stale against real master + (PR #788 F1) and unit tests stay offline (F2). Opt out with an explicit + ``GITEA_TEST_LIVE_REMOTE_HEAD`` pin or ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``. """ forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD) if forced is not None: return forced.strip() or None + if _HERMETIC_TEST_MODE and not ( + os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or "" + ).strip(): + # Hermetic default: live head unknown. live_stale stays False; + # mutation_safe is False when live is unknown (documented #610 note). + return None + # Defense in depth: even without the module flag, never probe while pytest + # is running unless the test opted into a real probe or set an override. + if (os.environ.get("PYTEST_CURRENT_TEST") or "").strip() and not ( + os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or "" + ).strip(): + return None if not root: return None key = (root, remote, branch) diff --git a/tests/conftest.py b/tests/conftest.py index 31408a3..4276e32 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -167,6 +167,35 @@ def _reset_mutation_authority(monkeypatch): import pytest +@pytest.fixture(autouse=True) +def _hermetic_live_remote_master_head(): + """#610 / PR #788 F1/F2: keep live-remote parity reads offline in tests. + + ``read_remote_master_head`` would otherwise ``git ls-remote`` whenever + ``GITEA_TEST_LIVE_REMOTE_HEAD`` is unset. Feature worktrees under + ``branches/`` always differ from live master, so legacy suites that assert + runtime-context ``safe_next_action`` flip to live_stale. Module-level + hermetic mode survives ``patch.dict(os.environ, …, clear=True)``. + Tests that exercise the real probe path call + ``master_parity_gate.set_hermetic_test_mode(False)`` and/or set + ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``. + """ + try: + import master_parity_gate as _mpg + + _mpg.set_hermetic_test_mode(True) + except Exception: + _mpg = None + try: + yield + finally: + if _mpg is not None: + try: + _mpg.set_hermetic_test_mode(False) + except Exception: + pass + + @pytest.fixture(autouse=True) def _deterministic_workspace_remotes(): try: diff --git a/tests/test_master_parity_gate.py b/tests/test_master_parity_gate.py index ffefd37..a1a5cbc 100644 --- a/tests/test_master_parity_gate.py +++ b/tests/test_master_parity_gate.py @@ -212,13 +212,27 @@ class TestRemoteHeadCache(unittest.TestCase): """ def setUp(self): + # These cases intentionally exercise the subprocess/cache path, so they + # opt out of suite-wide hermetic mode (PR #788 F1). + self._saved_hermetic = mp.hermetic_test_mode() + mp.set_hermetic_test_mode(False) mp._clear_remote_head_cache() - env = {k: v for k, v in os.environ.items() - if k != mp.ENV_TEST_LIVE_REMOTE_HEAD} + env = { + k: v for k, v in os.environ.items() + if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD, + mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE, + "PYTEST_CURRENT_TEST") + } + # Allow the probe path under hermetic defenses while still mocking + # subprocess so no real network call runs. + env[mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE] = "1" self._env = patch.dict(os.environ, env, clear=True) self._env.start() self.addCleanup(self._env.stop) self.addCleanup(mp._clear_remote_head_cache) + self.addCleanup( + lambda: mp.set_hermetic_test_mode(self._saved_hermetic) + ) def _fake_run(self, sha): class _R: @@ -256,6 +270,63 @@ class TestRemoteHeadCache(unittest.TestCase): self.assertEqual(calls["n"], 0) +class TestHermeticLiveRemoteReads(unittest.TestCase): + """#610 / PR #788 F1/F2: suite hermetic mode never hits the network.""" + + def setUp(self): + self._saved = mp.hermetic_test_mode() + mp.set_hermetic_test_mode(True) + mp._clear_remote_head_cache() + self.addCleanup(lambda: mp.set_hermetic_test_mode(self._saved)) + self.addCleanup(mp._clear_remote_head_cache) + + def test_hermetic_mode_returns_none_without_subprocess(self): + run_calls = {"n": 0} + + def boom(*args, **kwargs): + run_calls["n"] += 1 + raise AssertionError("ls-remote must not run under hermetic mode") + + env = { + k: v for k, v in os.environ.items() + if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD, + mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) + } + with patch.dict(os.environ, env, clear=True): + with patch.object(mp.subprocess, "run", boom): + self.assertIsNone( + mp.read_remote_master_head("/repo", remote="prgs") + ) + self.assertEqual(run_calls["n"], 0) + + def test_hermetic_mode_survives_clear_true_env(self): + """Module flag, not env pin: clear=True cannot re-enable the probe.""" + run_calls = {"n": 0} + + def boom(*args, **kwargs): + run_calls["n"] += 1 + raise AssertionError("ls-remote must not run after clear=True") + + with patch.dict(os.environ, {}, clear=True): + with patch.object(mp.subprocess, "run", boom): + self.assertIsNone(mp.read_remote_master_head("/repo")) + self.assertEqual(run_calls["n"], 0) + + def test_explicit_override_still_wins_under_hermetic(self): + run_calls = {"n": 0} + + def boom(*args, **kwargs): + run_calls["n"] += 1 + raise AssertionError("override must bypass subprocess") + + with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}): + with patch.object(mp.subprocess, "run", boom): + self.assertEqual( + mp.read_remote_master_head("/repo"), SHA_B + ) + self.assertEqual(run_calls["n"], 0) + + class TestServerWiring(unittest.TestCase): """Integration with the gate choke point in the server namespace."""