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 <[email protected]>
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user