ouyangbo/EvoScientist
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity
Files
57176b359a7f578bc413a847eb4aed5d45df8018
EvoScientist/tests/test_configurable_model_middleware.py
T
m4 57176b359a feat(runtime)!: switch middleware and agent factories to snapshot-driven models 
Replace config.yaml-driven model selection with registry snapshot resolution
across the runtime chain:

- ConfigurableModelMiddleware reads configurable["runtime_snapshot_id"] only;
  model/model_provider overrides are rejected with MODEL_CONFIG_OUTSIDE_SNAPSHOT
- MessageBudgetMiddleware derives budgets from snapshot reserves
  (system/tools/attachments) and re-resolves the summarizer per snapshot
- Agent factory and subagent factory resolve models via SnapshotRuntime
  (auxiliary/tool_selector/scheduler -> defaults.auxiliary ?? defaults.primary)
- Remove ModelFallbackMiddleware, /model-fallback command, and fallback chain
- Add model_registry/runtime.py SnapshotRuntime glue layer

Legacy config.yaml LLM fields, /model command, and llm/models.py remain for
Task 7. Report: .superpowers/sdd/briefs/task-6-report.md
2026-07-21 12:54:35 +08:00

473 lines
17 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 passes through unchanged.
"""
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,
read_snapshot_binding,
)
from EvoScientist.model_registry.errors import (
MODEL_CONFIG_OUTSIDE_SNAPSHOT,
SNAPSHOT_EXPIRED,
SNAPSHOT_NOT_FOUND,
ModelRegistryError,
)
from EvoScientist.model_registry.runtime import SnapshotRuntime
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. Pass-through behavior (no snapshot in configurable)
# =============================================================================
class TestPassThrough:
def test_sync_no_snapshot_passes_request_unchanged(self, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
sentinel = object()
handler = MagicMock(return_value=sentinel)
with _patched_config({}):
result = mw.wrap_model_call(req, handler)
assert result is sentinel
handler.assert_called_once_with(req)
req.override.assert_not_called()
async def test_async_no_snapshot_passes_request_unchanged(self, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
async def handler(r):
assert r is req
return "ok"
with _patched_config({}):
result = await mw.awrap_model_call(req, handler)
assert result == "ok"
req.override.assert_not_called()
def test_outside_runnable_context_passes_through(self, runtime):
mw = ConfigurableModelMiddleware(runtime=runtime)
req = _make_request()
handler = MagicMock(return_value="ok")
with _patched_config(None):
mw.wrap_model_call(req, handler)
handler.assert_called_once_with(req)
# =============================================================================
# 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_passes_through(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")
mw.wrap_model_call(req, handler)
finally:
var_child_runnable_config.reset(token)
handler.assert_called_once_with(req)
req.override.assert_not_called()
Reference in New Issue View Git Blame Copy Permalink