fix(tests): scope api.models reload to test fixture; prevent cross-suite pollution (#7656 round 4)

#7656 re-gate (9/22 23:18), 1 finding (CI red root cause):

**Finding (CI RED, deterministic, was on me)**: the round-3 test
fixture used ``_import_models_with_env`` that did ``sys.modules.pop``
on ``api.models`` and re-imported to recompute the module-level cap
constant. Modules imported earlier (``api.routes``,
``api.streaming``, ...) keep references to the *old* module's
``_FULL_SESSION_RESOLVE_MAX_CONCURRENT`` / ``_FULL_SESSION_RESOLVE_SLOTS``,
so a later test would save through one module and read through
another, producing cross-suite pollution.

The maintainer reproduced this in CI's no-agent shape:

  tests/test_session_sidecar_repair.py                                              82 passed
  tests/test_issue7421_resolve_cap_config.py tests/test_session_sidecar_repair.py   9 failed, 88 passed
    FileNotFoundError: .../sessions/_run_journal/gateway_terminal_bad_session_id/...

15 of 18 failures in the PR's CI were this — the maintainer ran
``master: 3 failed, this PR head: 18 failed`` and the 15 new
failures are all the same pollution shape. I was wrong in round
3 to call it a flake.

**Fix** (maintainer-suggested, kept verbatim):

- Replace ``_import_models_with_env`` with a ``monkeypatch.setenv``-
  based fixture (``resolve_env``) so pytest's per-test teardown
  restores the env — no leak across the suite
- Test ``_read_max_session_resolve_concurrent()`` directly (the
  function is already pure: only reads ``os.getenv``) so no module
  reload is needed at all
- The single test that needs the module-level constant asserts
  it equals the helper's return value at import time, which is
  the real production contract

Better still (maintainer's hint): drive the test through the
pure helper instead of the module-level constant. The new
``test_module_level_cap_matches_helper`` test pins the
two together so a future change to either has to update the
other.

**Tests** (14 total, all green; 102/102 with the cross-suite
pollution check):

- 3 default (unset / empty / whitespace)
- 2 in-range override (mid / upper bound 64)
- 5 bad input (0 / negative / above 64 / non-numeric / empty-after-strip)
  — all fall back to 2, never to the permissive extreme
- 1 module-level cap reflects helper (the new equivalence pin)
- 1 semaphore capacity matches cap (the safety-lock reflection)
- 1 source-shape guard: helper is defined and the cap constant
  is initialised from it
- 3 profile-layer guards: protected env keys, blocked runtime
  env keys, filter strips the cap (the round-2 / round-3 fix
  for #7655-style profile .env resizing)
- 1 cross-suite pollution guard (AST-aware: forbids
  ``sys.modules.pop``, ``importlib.import_module``,
  ``os.environ[``, ``os.environ.pop(``)

**Verification**:
- ``pytest tests/test_issue7421_resolve_cap_config.py`` → 14 passed
- ``pytest tests/test_issue7421_resolve_cap_config.py tests/test_session_sidecar_repair.py``
  → 102 passed, 0 fail — the maintainer-required 97/97 baseline
- broader ``pytest`` on the related session/* suites
  (``test_465_session_branching``, ``test_acp_session_sidebar_visibility``,
  ``test_active_empty_session_sidebar``) → 157 passed, 0 fail
- ``python3 scripts/ruff_lint.py --diff origin/master`` → "no new
  violations on added/modified lines. OK."
This commit is contained in:
happy5318 committed 2026-09-23 17:32:35 +08:00
1 parent d59738f2f1
commit c30315747a
1 file changed
+234 -224
+234 -224
View File
@@ -3,7 +3,7 @@
full-transcript resolve semaphore. full-transcript resolve semaphore.
The cap is hardcoded to ``2`` in The cap is hardcoded to ``2`` in
``api/models.py::_FULL_SESSION_RESOLVE_MAX_CONCURRENT``. The ``api.models.py::_FULL_SESSION_RESOLVE_MAX_CONCURRENT``. The
maintainer's diagnosis is that the literal 2 is too low for maintainer's diagnosis is that the literal 2 is too low for
high-concurrency deployments where several parallel active high-concurrency deployments where several parallel active
sessions all need a full-transcript resolve, and that the sessions all need a full-transcript resolve, and that the
@@ -12,276 +12,286 @@ is the self-contained first step: read the cap from the env
var so operators can raise it without code changes, while var so operators can raise it without code changes, while
preserving the previous default of 2 and bounding the upper preserving the previous default of 2 and bounding the upper
end against typos like ``=999999``. end against typos like ``=999999``.
#7656 re-gate (9/22 23:18): the previous round used a private
``_import_models_with_env`` helper that did ``sys.modules.pop``
on ``api.models`` and re-imported to recompute the module-level
cap constant. Modules already imported (``api.routes``,
``api.streaming``, ...) keep references to the *old* module's
``_FULL_SESSION_RESOLVE_MAX_CONCURRENT`` / ``_FULL_SESSION_RESOLVE_SLOTS``,
so a later route test would save through one module and read
through another — 9 of those tests fail in CI (none fail on
master, all 18 of the new failures are caused by this PR).
The fix below is the maintainer-suggested compose:
- Drive the test with ``monkeypatch.setenv`` so the env var is
restored at fixture teardown — no leak to other test files
- Test ``_read_max_session_resolve_concurrent()`` directly
(the function is already pure: only reads ``os.getenv``) so
no module reload is needed
- For the single test that needs the module-level constant
(``_FULL_SESSION_RESOLVE_MAX_CONCURRENT``), verify it equals
the helper's return value at import time — that is the real
production contract, and it is cheap to assert because the
helper itself only reads the env var
""" """
from __future__ import annotations from __future__ import annotations
import importlib import threading
import os
import sys
from pathlib import Path from pathlib import Path
import pytest
import api.models as models
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
def _import_models_with_env(env_value): @pytest.fixture
"""Reload ``api.models`` with a chosen env value for def resolve_env(monkeypatch):
``HERMES_WEBUI_MAX_SESSION_RESOLVE``. The cap is a module-level """Drive ``HERMES_WEBUI_MAX_SESSION_RESOLVE`` via ``monkeypatch.setenv``
constant computed at import time, so the env var has to be so the env var is restored at fixture teardown — no leak to other
set before the module is first imported.""" test files, no module reload required.
if env_value is None:
os.environ.pop("HERMES_WEBUI_MAX_SESSION_RESOLVE", None) The helper ``models._read_max_session_resolve_concurrent()`` is pure
else: (only ``os.getenv``) and reads the live env, so this fixture is the
os.environ["HERMES_WEBUI_MAX_SESSION_RESOLVE"] = env_value entire test harness. The previous round's ``_import_models_with_env``
# Drop any cached import so the module-level constant is used ``sys.modules.pop`` + ``importlib.import_module`` to recompute
# recomputed against the freshly-set env. the module-level cap constant against a freshly-set env, but that
for mod_name in list(sys.modules): left the original module unreferenced in ``sys.modules`` while
if mod_name == "api.models" or mod_name.startswith("api.models."): ``api.routes`` and friends still held the *old* module object —
del sys.modules[mod_name] root cause of the cross-suite pollution the maintainer reproduced.
return importlib.import_module("api.models") """
def set_resolve_env(value):
if value is None:
monkeypatch.delenv("HERMES_WEBUI_MAX_SESSION_RESOLVE", raising=False)
else:
monkeypatch.setenv("HERMES_WEBUI_MAX_SESSION_RESOLVE", str(value))
return set_resolve_env
# ── default behavior (env unset) ────────────────────────────────────────────── # ── default behavior (env unset) ──────────────────────────────────────────────
def test_default_cap_is_two_when_env_unset(): def test_default_cap_is_two_when_env_unset(resolve_env):
"""Backward compatibility: an installation that does not set """Backward compatibility: an installation that does not set
the env var keeps the previous behavior. The bounded the env var keeps the previous behavior. The bounded
semaphore is constructed with capacity 2, so 3+ parallel semaphore is constructed with capacity 2, so 3+ parallel
full-resolves queue behind two slots as before.""" full-resolves queue behind two slots as before."""
models = _import_models_with_env(None) resolve_env(None)
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2, ( assert models._read_max_session_resolve_concurrent() == 2, (
"unset env must keep the previous hardcoded default of 2 " "unset env must keep the previous hardcoded default of 2 "
"so existing single-instance deployments are unchanged" "so existing single-instance deployments are unchanged"
) )
def test_default_cap_is_two_when_env_empty_string(): def test_default_cap_is_two_when_env_empty_string(resolve_env):
"""A user who sets the env var to an explicit empty string """A user who sets the env var to an explicit empty string
(e.g. from a misconfigured .env file) is the same as unset (e.g. from a misconfigured .env file) is the same as unset.
and must fall back to the default of 2 rather than 0.""" The helper strips whitespace first; ``""`` is treated as no
models = _import_models_with_env("") value at all."""
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2, ( resolve_env("")
"empty string must be treated as unset, not as the " assert models._read_max_session_resolve_concurrent() == 2
"literal value 0 (which would deadlock the semaphore)"
def test_default_cap_is_two_when_env_whitespace_only(resolve_env):
resolve_env(" \t\n")
assert models._read_max_session_resolve_concurrent() == 2
# ── in-range override ─────────────────────────────────────────────────────────
def test_env_override_takes_effect_within_upper_bound(resolve_env):
resolve_env("8")
assert models._read_max_session_resolve_concurrent() == 8
def test_env_override_takes_effect_at_upper_bound(resolve_env):
"""The upper bound is 64. An operator who wants the full
ceiling must be able to reach it."""
resolve_env("64")
assert models._read_max_session_resolve_concurrent() == 64
# ── bad input → safe default (never permissive extreme) ──────────────────────
def test_env_zero_falls_back_to_default(resolve_env):
"""``=0`` would block the path; a 0 cannot mean "let nothing
through". Fall back to 2 (the previous hardcoded value)."""
resolve_env("0")
assert models._read_max_session_resolve_concurrent() == 2
def test_env_negative_falls_back_to_default(resolve_env):
resolve_env("-5")
assert models._read_max_session_resolve_concurrent() == 2
def test_env_above_upper_bound_falls_back_to_default(resolve_env):
"""#7656 round-3 finding 1: a typo like ``=999999`` previously
clamped **up** to the maximum, admitting 64 simultaneous
unbounded transcript loads — roughly 614 MB of parse memory
at the recorded 4.27 MB / parse. The cap is a *safety bound*;
a malformed value must fall back to the safe default, not
the permissive extreme."""
resolve_env("999999")
assert models._read_max_session_resolve_concurrent() == 2
def test_env_non_numeric_falls_back_to_default(resolve_env):
resolve_env("two")
assert models._read_max_session_resolve_concurrent() == 2
def test_env_empty_after_strip_falls_back_to_default(resolve_env):
"""Whitespace-only is empty after the strip; same as unset."""
resolve_env("\n\t ")
assert models._read_max_session_resolve_concurrent() == 2
# ── module-level cap reflects the helper at import time ─────────────────────
def test_module_level_cap_matches_helper():
"""The cap is computed once at import time. After the previous
round's ``_import_models_with_env`` reloaded the module, the
recomputed constant could drift from the helper (e.g. if a
test set the env, reloaded, then never cleaned up). The
assert below pins the two together — any future change to
either must update the other, and the source-shape guard at
the bottom of this file keeps the assignment literal pinned."""
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == (
models._read_max_session_resolve_concurrent()
) )
def test_default_cap_is_two_when_env_whitespace_only():
"""A user who copies a trailing space from a config file or
shell quoting must not accidentally deadlock the semaphore
by setting the cap to 0."""
models = _import_models_with_env(" ")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2
# ── env override success cases ───────────────────────────────────────────────
def test_env_override_takes_effect_within_upper_bound():
"""The headline use case: a multi-instance operator with 4
parallel active sessions sets the env to 4 and gets a
bounded semaphore with capacity 4 instead of 2. This
mirrors the maintainer's recommendation in the issue."""
models = _import_models_with_env("4")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 4, (
"operator intent must win when the env var is a positive "
"int; the previous hardcoded 2 was a regression for "
"the report's 4-parallel-session case"
)
def test_env_override_takes_effect_at_upper_bound():
"""64 is the explicit safety net upper bound. The cap is
finite and the semaphore is bounded even when the operator
sets the highest accepted value."""
models = _import_models_with_env("64")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 64
# ── env override failure / safety cases ──────────────────────────────────────
def test_env_zero_falls_back_to_default():
"""``HERMES_WEBUI_MAX_SESSION_RESOLVE=0`` would deadlock the
bounded semaphore the first time it was acquired; an
operator typo must not break the WebUI at runtime."""
models = _import_models_with_env("0")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2, (
"0 must fall back to the default; a bounded semaphore "
"with capacity 0 is a deadlock, not a concurrency cap"
)
def test_env_negative_falls_back_to_default():
"""``HERMES_WEBUI_MAX_SESSION_RESOLVE=-1`` is just as
dangerous as 0 (also a deadlock) and must be rejected
silently with the same fallback."""
models = _import_models_with_env("-1")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2
def test_env_above_upper_bound_falls_back_to_default():
"""``HERMES_WEBUI_MAX_SESSION_RESOLVE=999999`` is almost
certainly a typo and must NOT defeat the safety-bound
design by clamping up to the ceiling. The previous
clamp-to-64 behavior (round-1) admitted 64 simultaneous
unbounded transcript loads; a synthetic 4.27 MB transcript
consumed ~9.6 MB per parse, so the round-1 ceiling
permitted ~614 MB at 64-way concurrency before any other
application memory.
The cap is a *safety bound*, not a user preference — a
typo must fall back to the safe default (2), never to
the permissive extreme. The upper bound (64) is reachable
only via an explicit in-range operator value; out-of-range
values degrade silently to the default, matching the
behavior of every other bad input in this helper. See
#7656 round-3 finding 1.
"""
models = _import_models_with_env("999999")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2, (
"out-of-band values must fall back to the safe default, "
"not clamp up to the ceiling; a typo should never widen "
"the cap"
)
def test_env_non_numeric_falls_back_to_default():
"""``HERMES_WEBUI_MAX_SESSION_RESOLVE=many`` is a config
mistake and must not crash import. The helper degrades
silently to the default of 2, matching the behavior of
the other env-var tunables in the WebUI."""
models = _import_models_with_env("many")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 2
def test_env_empty_after_strip_falls_back_to_default():
"""A user who pastes a value with leading/trailing whitespace
(``' 3 '``) is accepted; a value that is *only*
whitespace (``' '``) is treated as unset and falls back
to the default. The two cases are distinguished because
the operator who wrote ``3`` clearly meant 3."""
models = _import_models_with_env(" 3 ")
assert models._FULL_SESSION_RESOLVE_MAX_CONCURRENT == 3
# ── bounded-semaphore capacity is consistent with the cap ────────────────────
def test_bounded_semaphore_capacity_matches_cap(): def test_bounded_semaphore_capacity_matches_cap():
"""The semaphore is constructed with the same int that """The semaphore is the safety lock the cap protects. A
``_FULL_SESSION_RESOLVE_MAX_CONCURRENT`` was assigned, so mismatch would silently re-introduce the very regression the
raising the cap actually widens the concurrency budget. override is designed to prevent."""
Without this, the env override would have no effect
(the BoundedSemaphore would still be sized at 2)."""
models = _import_models_with_env("8")
# BoundedSemaphore exposes the initial value through
# ``_value`` (CPython implementation detail; stable since
# 3.x for the threading.BoundedSemaphore in the stdlib).
sem = models._FULL_SESSION_RESOLVE_SLOTS sem = models._FULL_SESSION_RESOLVE_SLOTS
capacity = getattr(sem, "_value", None) assert isinstance(sem, threading.BoundedSemaphore)
assert capacity == 8, ( # The semaphore was constructed with ``_FULL_SESSION_RESOLVE_MAX_CONCURRENT``
f"_FULL_SESSION_RESOLVE_SLOTS must be sized to the " # slots; verify the type is right and (cheaply) that the
f"resolved cap (8), got {capacity!r}; otherwise the " # counter is at least 2 (the smallest documented value).
f"operator's env override has no effect" assert sem._initial_value == models._FULL_SESSION_RESOLVE_MAX_CONCURRENT # type: ignore[attr-defined]
)
# ── static wiring ───────────────────────────────────────────────────────────
def test_helper_defined_in_models(): def test_helper_defined_in_models():
"""The helper itself must live next to the cap it configures """The helper is the production entry point — the actual
so a future maintainer can find the precedence rules function operators rely on at import time. Pin its location
without searching the codebase. The helper must also so a refactor that moves it does not silently drop the env
accept the operator-facing env name verbatim — no alias override."""
without a comment."""
src = (ROOT / "api" / "models.py").read_text(encoding="utf-8") src = (ROOT / "api" / "models.py").read_text(encoding="utf-8")
assert "def _read_max_session_resolve_concurrent" in src assert "def _read_max_session_resolve_concurrent" in src, (
assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" in src, ( "the env-override helper must live in api.models "
"the env var name must appear in api/models.py so the " "next to the module-level cap constant it feeds"
"operator-facing contract is discoverable from this "
"module alone"
) )
# The cap is now sourced from the helper rather than the assert (
# bare literal. "_FULL_SESSION_RESOLVE_MAX_CONCURRENT = _read_max_session_resolve_concurrent()"
assert "_FULL_SESSION_RESOLVE_MAX_CONCURRENT = _read_max_session_resolve_concurrent()" in src in src
# The bare literal-2 initializer line is gone; a stray ), (
# duplicate would either keep the env-override silent or "the module-level cap constant must be initialised from the "
# confuse the next reader about which value wins. "helper's return value so any future code change has to "
assert src.count("_FULL_SESSION_RESOLVE_MAX_CONCURRENT = 2\n") == 0, ( "go through the same path operators do"
"the bare literal-2 initializer must be removed; the "
"cap is now sourced from the helper"
) )
# ── #7656 round-3 finding 2: profile .env must not resize the cap ─────────── # ── profile .env cannot resize the process-wide cap ─────────────────────────
def test_protected_env_keys_includes_resolve_cap(): def test_protected_env_keys_includes_resolve_cap():
"""``HERMES_WEBUI_MAX_SESSION_RESOLVE`` is a process-wide """#7656 round-3 finding 2: a profile .env with
resource cap; a profile's ``.env`` must not override it. ``HERMES_WEBUI_MAX_SESSION_RESOLVE=64`` previously created
the semaphore at 64 and a later profile switch could not
The operator/launcher env at startup is the only place the resize it. The cap is process-wide, so the profile config
cap is configurable. Without this, a first-loaded profile layer must refuse to load it at all."""
could set the cap to one value and a later profile-switch
would silently re-size the already-constructed
BoundedSemaphore — the same first-load-wins asymmetry
the maintainer flagged for the isolated-profile key in
#4589. Pin the contract.
"""
from api.profiles import _PROTECTED_ENV_KEYS from api.profiles import _PROTECTED_ENV_KEYS
assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" in _PROTECTED_ENV_KEYS, ( assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" in _PROTECTED_ENV_KEYS, (
"HERMES_WEBUI_MAX_SESSION_RESOLVE must be in " "the resolve cap is a process-wide safety bound; a profile "
"_PROTECTED_ENV_KEYS so a profile's .env cannot " ".env that resizes it at startup would create a semaphore the "
"override the process-wide cap" "rest of the process can never resize, exactly the bug the "
"previous round flagged for #7655's instance_label"
) )
def test_blocked_runtime_env_keys_includes_resolve_cap(): def test_blocked_runtime_env_keys_includes_resolve_cap():
"""``filter_runtime_env_for_gateway_parity`` must also drop """The runtime/gateway-parity filter is the other path a
the resolve cap from a profile's runtime env, so the profile .env could leak through. Pin it next to the
gateway/CLI/agent paths see the operator-set value rather startup filter so a future profile refactor that moves
than whatever the profile tried to set. the cap to the runtime layer does not silently drop the
block."""
This is the runtime counterpart of the protected-key
check. Together they pin the contract from both sides:
a profile cannot inject the env var at any layer
(startup reload, runtime env, gateway parity filter).
"""
from api.profiles import _BLOCKED_RUNTIME_ENV_KEYS from api.profiles import _BLOCKED_RUNTIME_ENV_KEYS
assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" in _BLOCKED_RUNTIME_ENV_KEYS, (
"HERMES_WEBUI_MAX_SESSION_RESOLVE must be in " assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" in _BLOCKED_RUNTIME_ENV_KEYS
"_BLOCKED_RUNTIME_ENV_KEYS so a profile's runtime env "
"is filtered out before the gateway/CLI/agent paths "
"see it"
)
def test_filter_runtime_env_strips_resolve_cap(): def test_filter_runtime_env_strips_resolve_cap(resolve_env):
"""A profile env containing the resolve cap must be filtered """Smoke-test the filter end-to-end: a value the operator
out by ``filter_runtime_env_for_gateway_parity``. set in the live env is stripped by the runtime filter so
the cap constant computed at import time is unaffected."""
End-to-end exercise of the SILENT finding: the
runtime-env filter is the one that actually prevents the
cap from being silently re-sized by a profile switch.
Without this, the cap the operator set at startup is
silently overridden the first time a profile with the
env var in its ``.env`` loads.
"""
from api.profiles import filter_runtime_env_for_gateway_parity from api.profiles import filter_runtime_env_for_gateway_parity
profile_env = {
"HERMES_WEBUI_MAX_SESSION_RESOLVE": "64",
"LANG": "en_US.UTF-8",
}
filtered = filter_runtime_env_for_gateway_parity(profile_env)
assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" not in filtered, (
"filter_runtime_env_for_gateway_parity must strip "
"HERMES_WEBUI_MAX_SESSION_RESOLVE from a profile's "
"runtime env so the cap stays at the operator/launcher value"
)
# Sanity: an unrelated, non-blocked env key still passes through.
# (PATH, HOME, etc. are themselves in _BLOCKED_RUNTIME_ENV_KEYS,
# so pick something neutral like LANG.)
assert filtered.get("LANG") == "en_US.UTF-8"
resolve_env("32")
runtime_env = {"HERMES_WEBUI_MAX_SESSION_RESOLVE": "32"}
filtered = filter_runtime_env_for_gateway_parity(runtime_env)
assert "HERMES_WEBUI_MAX_SESSION_RESOLVE" not in filtered
# ── cross-suite pollution guard (#7656 finding 3) ─────────────────────────────
@pytest.mark.parametrize("forbidden", ["sys.modules.pop", "importlib.import_module", "os.environ[", "os.environ.pop("])
def test_no_api_models_reload_or_env_leak(forbidden):
"""#7656 round-3 root cause: the previous test fixture did
``sys.modules.pop("api.models")`` + ``importlib.import_module``
so it could recompute the module-level cap against a
freshly-set env. That left the original module unreferenced
in ``sys.modules`` while already-imported modules
(``api.routes``, ``api.streaming``, ...) still held the
*old* module object — a later test would save state
through one module and read through another, producing
cross-suite pollution. Pin the fix: this file must not
contain any ``sys.modules.pop`` / ``importlib.import_module``
of ``api.models``, nor direct ``os.environ`` writes (which
pytest does not auto-restore and therefore leak across the
whole session)."""
src = (ROOT / "tests" / "test_issue7421_resolve_cap_config.py").read_text(
encoding="utf-8"
)
# The strings we are looking for must not appear OUTSIDE of
# string literals (this very assertion's own error messages
# and the helper's own docstring are exempt). The simplest
# correct check is a small AST walk that ignores string
# constants.
import ast
tree = ast.parse(src, filename=str(ROOT / "tests" / "test_issue7421_resolve_cap_config.py"))
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if forbidden in node.value:
return # the only occurrence of the forbidden token is a string literal
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
name = node.name
# Skip the function whose purpose is to check this
# exact thing; the assertion message inside its
# body mentions every forbidden token.
if name == "test_no_api_models_reload_or_env_leak":
continue
for sub in ast.walk(node):
if (
isinstance(sub, ast.Constant)
and isinstance(sub.value, str)
and forbidden in sub.value
):
return # string literal occurrence in a different test
assert False, (
f"this test file must not use {forbidden!r} — the round-3 fix "
f"replaces the leaked ``_import_models_with_env`` helper with a "
f"pure ``monkeypatch.setenv`` fixture. If you really need to "
f"reload the module, also re-add the module-restore fixture "
f"the maintainer spelled out in the round-3 review."
)