Merge pull request #2475 from Molecule-AI/chore/runtime-wedge-followups

chore(smoke): runtime_wedge follow-ups from PR #2473 review
This commit is contained in:
Hongming Wang 2026-05-02 01:07:18 +00:00 committed by GitHub
commit ff21c26835
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 74 additions and 22 deletions

View File

@ -111,13 +111,20 @@ def _check_runtime_wedge() -> str | None:
a corrupt-rolling-deploy state, in which case "no wedge info"
reads as "assume healthy" same fail-open posture heartbeat.py
takes for the same reason.
Catch is narrowed to import errors only a signature change
(`is_wedged` removed/renamed, `wedge_reason` returning the wrong
type) must NOT silently degrade to "no wedge info." The runtime's
structural snapshot test (workspace/tests/test_runtime_wedge_signature.py,
task #169) carries the API-drift load: any rename surfaces there
as a snapshot mismatch instead of letting the smoke gate go blind.
"""
try:
from runtime_wedge import is_wedged, wedge_reason
except Exception:
except (ImportError, ModuleNotFoundError):
return None
if is_wedged():
return wedge_reason() or "<unspecified>"
return wedge_reason()
return None

View File

@ -295,3 +295,46 @@ if "coordinator" not in sys.modules:
# Don't mock prompt or coordinator if they can be imported from the workspace-template dir
# test_prompt.py and test_coordinator.py need the real modules
# ─── runtime_wedge cross-test isolation ─────────────────────────────────
#
# `runtime_wedge` carries module-scope state via the `_DEFAULT` instance
# (workspace/runtime_wedge.py). Any test that calls `mark_wedged` and
# doesn't clean up leaks a sticky wedge into every later test in the
# same pytest process. Smoke tests (test_smoke_mode.py) that read
# `is_wedged()` would then fail-via-leak instead of assessing the code
# under test.
#
# Autouse fixture is scoped to the workspace/tests/ tree (this conftest
# is at workspace/tests/conftest.py), so it runs for every test that
# touches the runtime — without each test having to opt in. The
# import is deferred to fixture-call time so the fixture also works
# in environments where runtime_wedge isn't yet importable (matches
# the fail-open posture that smoke_mode + heartbeat take at the
# consumer side).
import pytest as _pytest # alias to avoid colliding with any existing `pytest` name
@_pytest.fixture(autouse=True)
def _reset_runtime_wedge_between_tests():
"""Reset the universal runtime_wedge flag before AND after every
workspace test so module-scope state can't leak across tests.
A test that calls `mark_wedged` without cleanup would otherwise
contaminate the next test's `is_wedged()` read — and because the
flag is sticky-first-write-wins, the later test couldn't even
overwrite the leaked reason. Two-sided reset (yield + cleanup)
means an early failure also doesn't poison the rest of the run.
"""
try:
from runtime_wedge import reset_for_test
except (ImportError, ModuleNotFoundError):
# No runtime_wedge installed — nothing to reset. Yield as a
# no-op so the fixture still runs the test.
yield
return
reset_for_test()
yield
reset_for_test()

View File

@ -17,6 +17,7 @@ construction skip when those symbols aren't reachable.
from __future__ import annotations
import asyncio
import sys
from unittest.mock import patch
import pytest
@ -256,24 +257,14 @@ class _MarkWedgedThenBlockExecutor:
await asyncio.Event().wait()
@pytest.fixture
def reset_runtime_wedge():
"""Ensure each wedge-test starts and ends with the runtime healthy.
The wedge is module-scoped state (`_DEFAULT` in runtime_wedge.py),
so a leak from one test would contaminate every subsequent smoke
test in the same pytest process. Reset on both sides so an early
failure doesn't poison the rest of the file either.
"""
import runtime_wedge
runtime_wedge.reset_for_test()
yield
runtime_wedge.reset_for_test()
# Note: runtime_wedge state is reset before/after every test by the
# autouse `_reset_runtime_wedge_between_tests` fixture in conftest.py
# so individual wedge tests don't need an explicit fixture argument.
@pytest.mark.asyncio
async def test_smoke_fails_when_adapter_marked_wedged_via_exception(
stub_build, reset_runtime_wedge,
stub_build,
):
"""PR-25 regression class: adapter catches SDK init wedge, marks
runtime_wedge, raises a sanitized error. Outer exception class
@ -287,7 +278,7 @@ async def test_smoke_fails_when_adapter_marked_wedged_via_exception(
@pytest.mark.asyncio
async def test_smoke_fails_when_adapter_marked_wedged_then_blocks(
stub_build, reset_runtime_wedge, monkeypatch: pytest.MonkeyPatch,
stub_build, monkeypatch: pytest.MonkeyPatch,
):
"""Same wedge class as above but the adapter doesn't raise — it
keeps awaiting (e.g. waiting on a control-message reply that will
@ -303,7 +294,7 @@ async def test_smoke_fails_when_adapter_marked_wedged_then_blocks(
@pytest.mark.asyncio
async def test_smoke_passes_when_runtime_wedge_is_clean_after_clean_execute(
stub_build, reset_runtime_wedge,
stub_build,
):
"""Belt-and-braces: wedge-clean + clean execute() must still PASS.
Pins that the new check is additive it doesn't accidentally
@ -317,9 +308,20 @@ def test_check_runtime_wedge_returns_none_when_module_missing(
monkeypatch: pytest.MonkeyPatch,
):
"""Direct test for the import-resilience contract — the helper
must swallow ImportError (and any other exception while reading
the module) so a corrupt install doesn't crash the smoke gate."""
must swallow ImportError so a corrupt install doesn't crash the
smoke gate. Catch is narrowed to (ImportError, ModuleNotFoundError)
so a SIGNATURE drift surfaces; this test only pins the missing-
module case.
Defensive: drop runtime_wedge from sys.modules cache before
patching __import__. Without the cache evict, an earlier test in
the same file that already imported runtime_wedge would let the
`from runtime_wedge import ...` here resolve from the cache and
skip __import__ entirely the test would pass for the wrong
reason and a real regression (catch arm removed) wouldn't surface.
"""
import builtins
monkeypatch.delitem(sys.modules, "runtime_wedge", raising=False)
real_import = builtins.__import__
def _raising_import(name, *args, **kwargs):
@ -331,7 +333,7 @@ def test_check_runtime_wedge_returns_none_when_module_missing(
assert smoke_mode._check_runtime_wedge() is None
def test_check_runtime_wedge_returns_reason_when_marked(reset_runtime_wedge):
def test_check_runtime_wedge_returns_reason_when_marked():
"""When an adapter has called runtime_wedge.mark_wedged(reason),
the helper returns that reason verbatim so the smoke can surface
it in the FAIL log line."""
@ -340,7 +342,7 @@ def test_check_runtime_wedge_returns_reason_when_marked(reset_runtime_wedge):
assert smoke_mode._check_runtime_wedge() == "explicit test reason"
def test_check_runtime_wedge_returns_none_when_clean(reset_runtime_wedge):
def test_check_runtime_wedge_returns_none_when_clean():
"""Pre-condition for the additive contract: helper must return
None (not the empty string from `wedge_reason()`) when no adapter
has marked the runtime wedged, so the caller's `is not None`