ouyangbo/EvoScientist
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity
Files
a1bfbd92ca1412b6dce33df74fc2d6310d9b440d
EvoScientist/tests/test_configurable_model_middleware.py
T
m4 421a664336 feat(runtime)!: complete legacy removal, local snapshot entries, and TTL cleanup 
- Remove legacy provider profiles, admin-token auth, /model command,
  model picker widget, and config.yaml LLM fields (design doc section 10)
- Wire CLI/channels/cron and async sub-agents through the local snapshot
  entry; run creation rejects model config outside runtime_snapshot_id
- Add periodic run-snapshot TTL cleanup to the config service lifespan
- Isolate tests from the real config dir and activate the registry where
  run/model paths fail closed in bootstrap

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-07-21 18:10:23 +08:00

539 lines
19 KiB
Python
Raw Blame History

"""Tests for ``EvoScientist.middleware.configurable_model`` (design doc 8.3).
Verifies the snapshot-only contract: the middleware resolves the frozen
``ResolvedModelConfig`` for its role from ``configurable["runtime_snapshot_id"]``
and builds the model via ``build_chat_model``; ``model``/``model_provider``
in ``configurable`` are rejected with ``MODEL_CONFIG_OUTSIDE_SNAPSHOT``; a
run without a snapshot ID gets a lazily created local snapshot bound to its
thread (section 8.1), or fails closed when no thread ID exists either.
"""
from __future__ import annotations
from contextlib import contextmanager
from unittest.mock import MagicMock, patch
import pytest
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI
from EvoScientist.middleware.configurable_model import (
ConfigurableModelMiddleware,
check_no_outside_snapshot_model_config,
ensure_snapshot_binding,
read_snapshot_binding,
)
from EvoScientist.model_registry.errors import (
MODEL_CONFIG_OUTSIDE_SNAPSHOT,
MODEL_REGISTRY_NOT_READY,
SNAPSHOT_EXPIRED,
SNAPSHOT_NOT_FOUND,
ModelRegistryError,
)
from EvoScientist.model_registry.runtime import SnapshotRuntime
from EvoScientist.model_registry.store import ModelRuntimeStore
from tests.registry_fixtures import (
ZHIPU_SECRET,
make_active_store,
make_snapshot,
)
@contextmanager
def _patched_config(configurable: dict | object | None):
"""Patch ``langgraph.config.get_config`` to return a controlled value.
Pass ``None`` to simulate "outside a runnable context" (raises
``RuntimeError`` like the real ``get_config()`` does).
Pass a dict for ``configurable`` to expose just that key.
Pass any other object to simulate a malformed config.
"""
import langgraph.config as _lg_cfg
if configurable is None:
# Simulate get_config raising outside a runnable context.
with patch.object(
_lg_cfg,
"get_config",
side_effect=RuntimeError("Called get_config outside of a runnable context"),
):
yield
elif isinstance(configurable, dict):
with patch.object(
_lg_cfg,
"get_config",
return_value={"configurable": configurable},
):
yield
else:
with patch.object(
_lg_cfg,
"get_config",
return_value=configurable,
):
yield
def _make_request():
"""Build a minimal ``ModelRequest`` stub.
``request.override(model=...)`` returns a new request whose ``model``
field reflects the override.
"""
req = MagicMock()
def _override(**kwargs):
new = MagicMock()
new.model = kwargs.get("model", req.model)
return new
req.override = MagicMock(side_effect=_override)
return req
@pytest.fixture
def store(tmp_path):
return make_active_store(tmp_path / "model-runtime")
@pytest.fixture
def runtime(store):
return SnapshotRuntime(store)
def _configurable_for(snapshot, **overrides):
configurable = {
"runtime_snapshot_id": snapshot.snapshot_id,
"thread_id": snapshot.thread_id,
}
configurable.update(overrides)
return configurable
# =============================================================================
# 1. Outside-snapshot model configuration is rejected (section 8.2)
# =============================================================================
class TestOutsideSnapshotConfig:
@pytest.mark.parametrize("key", ["model", "model_provider"])
def test_model_keys_rejected(self, key):
with pytest.raises(ModelRegistryError) as excinfo:
check_no_outside_snapshot_model_config({key: "gpt-5"})
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
assert excinfo.value.http_status == 422
assert excinfo.value.details[0].path == key
def test_rejected_alongside_snapshot_id(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
with (
_patched_config(_configurable_for(snapshot, model="gpt-5")),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, MagicMock())
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
def test_rejected_without_snapshot_id(self, runtime):
"""The check applies even on the pass-through path (验收 13)."""
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
handler = MagicMock()
with (
_patched_config({"model_provider": "openai"}),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, handler)
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
handler.assert_not_called()
def test_none_values_treated_as_absent(self):
check_no_outside_snapshot_model_config({"model": None, "thread_id": "t"})
# =============================================================================
# 2. Snapshot binding parsing
# =============================================================================
class TestReadSnapshotBinding:
def test_full_binding(self):
binding = read_snapshot_binding(
{
"runtime_snapshot_id": "snap-1",
"workspace_deployment_id": "deploy-1",
"thread_id": "thread-1",
}
)
assert binding == ("snap-1", "deploy-1", "thread-1")
def test_missing_snapshot_id_returns_none(self):
assert read_snapshot_binding({}) is None
assert read_snapshot_binding({"runtime_snapshot_id": ""}) is None
assert read_snapshot_binding({"runtime_snapshot_id": 42}) is None
def test_missing_deployment_and_thread_fall_back(self):
assert read_snapshot_binding({"runtime_snapshot_id": "snap-1"}) == (
"snap-1",
None,
None,
)
# =============================================================================
# 3. Fail-closed behavior (no snapshot in configurable, section 8.1)
# =============================================================================
class TestFailClosed:
def test_sync_no_snapshot_no_thread_rejected(self, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
handler = MagicMock()
with (
_patched_config({}),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, handler)
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
handler.assert_not_called()
req.override.assert_not_called()
async def test_async_no_snapshot_no_thread_rejected(self, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
async def handler(r):
raise AssertionError("handler must not run") # pragma: no cover
with (
_patched_config({}),
pytest.raises(ModelRegistryError) as excinfo,
):
await mw.awrap_model_call(req, handler)
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
req.override.assert_not_called()
def test_outside_runnable_context_fails_closed(self, runtime):
"""No runnable context means no thread binding — never pass through."""
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
handler = MagicMock()
with (
_patched_config(None),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, handler)
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
handler.assert_not_called()
def test_bootstrap_registry_fails_closed_not_ready(self, tmp_path):
bootstrap_runtime = SnapshotRuntime(
ModelRuntimeStore(config_dir=tmp_path / "model-runtime")
)
mw = ConfigurableModelMiddleware(runtime=bootstrap_runtime)
req = _make_request()
with (
_patched_config({"thread_id": "cli-thread"}),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, MagicMock())
assert excinfo.value.code == MODEL_REGISTRY_NOT_READY
class TestLazyLocalSnapshot:
def test_thread_without_snapshot_lazily_creates_one(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
captured = {}
def handler(r):
captured["model"] = r.model
return "ok"
with _patched_config({"thread_id": "cron-thread-1"}):
assert mw.wrap_model_call(req, handler) == "ok"
# The override carries the registry-default primary model.
assert captured["model"] is not req.model
# A snapshot was created and bound to the run's own thread.
found = store.find_active_run_snapshot(
deployment_id=runtime.local_deployment_id,
thread_id="cron-thread-1",
run_request_id="auto:cron-thread-1",
)
assert found is not None
def test_lazy_creation_converges_within_a_thread(self, store, runtime):
"""Every middleware in the run lands on the same lazily created snapshot."""
first = ensure_snapshot_binding({"thread_id": "t-1"}, runtime)
second = ensure_snapshot_binding({"thread_id": "t-1"}, runtime)
assert first == second
assert first[1] == runtime.local_deployment_id
def test_explicit_snapshot_wins_over_lazy_creation(self, store, runtime):
snapshot = make_snapshot(store)
binding = ensure_snapshot_binding(
{"runtime_snapshot_id": snapshot.snapshot_id, "thread_id": "other"},
runtime,
)
assert binding[0] == snapshot.snapshot_id
# =============================================================================
# 4. Snapshot-driven model construction (full chain)
# =============================================================================
class TestSnapshotDrivenConstruction:
def test_sync_builds_snapshot_primary_model(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="response")
with _patched_config(_configurable_for(snapshot)):
mw.wrap_model_call(req, handler)
called_with = handler.call_args[0][0]
assert called_with is not req
model = called_with.model
assert isinstance(model, ChatOpenAI)
assert model.model_name == "glm-5.2"
assert model.openai_api_base == "https://open.bigmodel.cn/api/paas/v4"
assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET
assert model.max_retries == 2
async def test_async_path_parity(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
async def handler(r):
return r.model
with _patched_config(_configurable_for(snapshot)):
model = await mw.awrap_model_call(req, handler)
assert isinstance(model, ChatOpenAI)
assert model.model_name == "glm-5.2"
def test_auxiliary_role_maps_to_snapshot_auxiliary(self, store, runtime):
mw = ConfigurableModelMiddleware(role="auxiliary", runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="ok")
with _patched_config(_configurable_for(snapshot)):
mw.wrap_model_call(req, handler)
model = handler.call_args[0][0].model
assert isinstance(model, ChatOllama)
assert model.model == "qwen3"
def test_summary_role_falls_back_to_snapshot_primary(self, tmp_path):
store = make_active_store(tmp_path / "db", auxiliary_default=False)
mw = ConfigurableModelMiddleware(role="summary", runtime=SnapshotRuntime(store))
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="ok")
with _patched_config(_configurable_for(snapshot)):
mw.wrap_model_call(req, handler)
model = handler.call_args[0][0].model
assert isinstance(model, ChatOpenAI)
assert model.model_name == "glm-5.2"
def test_deployment_id_from_configurable_is_verified(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="ok")
with _patched_config(
_configurable_for(snapshot, workspace_deployment_id="local")
):
mw.wrap_model_call(req, handler)
handler.assert_called_once()
def test_wrong_thread_binding_fails_closed(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
with (
_patched_config(_configurable_for(snapshot, thread_id="other-thread")),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, MagicMock())
assert excinfo.value.code == SNAPSHOT_NOT_FOUND
def test_wrong_deployment_binding_fails_closed(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
with (
_patched_config(
_configurable_for(snapshot, workspace_deployment_id="other-deploy")
),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, MagicMock())
assert excinfo.value.code == SNAPSHOT_NOT_FOUND
def test_expired_snapshot_fails_loudly(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
store.abort_run_snapshot(snapshot.snapshot_id)
req = _make_request()
with (
_patched_config(_configurable_for(snapshot)),
pytest.raises(ModelRegistryError) as excinfo,
):
mw.wrap_model_call(req, MagicMock())
assert excinfo.value.code == SNAPSHOT_EXPIRED
# =============================================================================
# 5. Frozen credential revision resolution
# =============================================================================
class TestCredentialRevision:
def test_snapshot_keeps_its_frozen_credential_revision(self, store, runtime):
"""Rotating the credential after freezing must not affect the run."""
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
store.write_credential_version("zhipu-primary", "sk-rotated-0000")
req = _make_request()
handler = MagicMock(return_value="ok")
with _patched_config(_configurable_for(snapshot)):
mw.wrap_model_call(req, handler)
model = handler.call_args[0][0].model
assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET
# =============================================================================
# 6. Per-instance caching
# =============================================================================
class TestCache:
def test_second_call_hits_cache(self, store, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="ok")
with (
_patched_config(_configurable_for(snapshot)),
patch.object(
SnapshotRuntime, "build_role_model", wraps=runtime.build_role_model
) as build,
):
mw.wrap_model_call(req, handler)
mw.wrap_model_call(req, handler)
assert build.call_count == 1
def test_independent_instances_have_independent_caches(self, store, runtime):
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="ok")
with (
_patched_config(_configurable_for(snapshot)),
patch.object(
SnapshotRuntime, "build_role_model", wraps=runtime.build_role_model
) as build,
):
ConfigurableModelMiddleware(runtime=runtime).wrap_model_call(req, handler)
ConfigurableModelMiddleware(runtime=runtime).wrap_model_call(req, handler)
assert build.call_count == 2
# =============================================================================
# 7. Constructor validation
# =============================================================================
class TestRoleValidation:
def test_unknown_role_rejected(self):
with pytest.raises(ValueError, match="Unknown model role"):
ConfigurableModelMiddleware(role="bogus")
# =============================================================================
# 8. Fallback removal startup assertion (design doc 8.3)
# =============================================================================
class TestFallbackChainRemoved:
def test_fallback_middleware_gone_from_package(self):
import importlib
import EvoScientist.middleware as mw
assert not hasattr(mw, "ModelFallbackMiddleware")
assert not hasattr(mw, "load_fallback_chain")
with pytest.raises(ModuleNotFoundError):
importlib.import_module("EvoScientist.middleware.model_fallback")
# =============================================================================
# 9. Integration — real langgraph contextvar (no get_config mock)
# =============================================================================
class TestRunnableContextVarIntegration:
"""Set the actual ``var_child_runnable_config`` contextvar that LangGraph
populates per node, then verify the middleware reads through to it.
This catches breakage of the ``langgraph.config.get_config()`` contract
that pure-mock tests would miss (e.g. if get_config is moved to a
different module, or the contextvar mechanism changes).
"""
def test_real_contextvar_drives_snapshot_resolution(self, store, runtime):
from langchain_core.runnables.config import var_child_runnable_config
mw = ConfigurableModelMiddleware(runtime=runtime)
snapshot = make_snapshot(store)
req = _make_request()
handler = MagicMock(return_value="ok")
token = var_child_runnable_config.set(
{"configurable": _configurable_for(snapshot)}
)
try:
mw.wrap_model_call(req, handler)
finally:
var_child_runnable_config.reset(token)
model = handler.call_args[0][0].model
assert isinstance(model, ChatOpenAI)
assert model.model_name == "glm-5.2"
def test_real_contextvar_unset_fails_closed(self, runtime):
from langchain_core.runnables.config import var_child_runnable_config
# Defensive: ensure no leftover contextvar from another test.
token = var_child_runnable_config.set(None)
try:
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
handler = MagicMock(return_value="ok")
with pytest.raises(ModelRegistryError) as excinfo:
mw.wrap_model_call(req, handler)
finally:
var_child_runnable_config.reset(token)
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
handler.assert_not_called()
req.override.assert_not_called()
Reference in New Issue View Git Blame Copy Permalink