"""Tests for ``EvoScientist.model_registry.runtime`` (design doc 8.3). Covers the runtime-link glue: registry-default primary resolution, snapshot role construction with per-call credential resolution, safe-client wiring (ollama carries retries on the transports), the shared default-runtime accessor, and the tolerant local platform field reader. """ from __future__ import annotations import pytest from langchain_ollama import ChatOllama from langchain_openai import ChatOpenAI from EvoScientist.model_registry.endpoint_policy import EndpointPolicy from EvoScientist.model_registry.errors import ( MODEL_REGISTRY_NOT_READY, ModelRegistryError, ) from EvoScientist.model_registry.platform import ( DEFAULT_LOCAL_DEPLOYMENT_ID, PlatformConfigError, ) from EvoScientist.model_registry.runtime import ( SnapshotRuntime, _read_local_platform_fields, get_snapshot_runtime, set_snapshot_runtime_for_tests, ) from EvoScientist.model_registry.schemas import DevelopmentEndpoint from EvoScientist.model_registry.store import ModelRuntimeStore from tests.registry_fixtures import ( OLLAMA_REF, ZHIPU_REF, ZHIPU_SECRET, make_active_store, make_snapshot, ) @pytest.fixture def store(tmp_path): return make_active_store(tmp_path / "model-runtime") @pytest.fixture def runtime(store): return SnapshotRuntime(store) @pytest.fixture(autouse=True) def _reset_default_runtime(): set_snapshot_runtime_for_tests(None) yield set_snapshot_runtime_for_tests(None) class TestRegistryDefaults: def test_bootstrap_registry_is_not_ready(self, tmp_path): runtime = SnapshotRuntime(ModelRuntimeStore(config_dir=tmp_path / "db")) with pytest.raises(ModelRegistryError) as excinfo: runtime.registry_default() assert excinfo.value.code == MODEL_REGISTRY_NOT_READY assert excinfo.value.http_status == 422 def test_active_registry_returns_default(self, runtime): primary, revision = runtime.registry_default() assert primary == ZHIPU_REF assert revision == 2 class TestResolveDefaultConfig: def test_resolves_primary_default(self, runtime): config = runtime.resolve_default_config() assert config.model_ref == ZHIPU_REF assert config.role == "primary" class TestBuildDefaultRoleModel: def test_primary_builds_chat_openai_with_frozen_options(self, runtime): model = runtime.build_default_role_model("primary") 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.max_retries == 2 assert model.temperature == 0.7 assert model.top_p == 0.95 def test_openai_compatible_never_reads_env_api_key(self, runtime, monkeypatch): """The credential comes from the frozen auth_ref, never OPENAI_API_KEY.""" monkeypatch.setenv("OPENAI_API_KEY", "sk-env-must-not-leak") model = runtime.build_default_role_model("primary") assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET class TestBuildRoleModelFromSnapshot: def test_primary_role_uses_snapshot_primary(self, store, runtime): snapshot = make_snapshot(store) model = runtime.build_role_model(snapshot, "primary") assert isinstance(model, ChatOpenAI) assert model.model_name == "glm-5.2" assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET class TestSharedDefaultRuntime: def test_set_and_get_runtime(self, runtime): set_snapshot_runtime_for_tests(runtime) assert get_snapshot_runtime() is runtime def test_builds_from_missing_config_yaml(self, tmp_path, monkeypatch): monkeypatch.setattr( "EvoScientist.model_registry.runtime.get_config_path", lambda: tmp_path / "config.yaml", ) monkeypatch.setattr( "EvoScientist.model_registry.store.DEFAULT_CONFIG_DIR", tmp_path / "config-dir", ) runtime = get_snapshot_runtime() assert runtime.local_deployment_id == DEFAULT_LOCAL_DEPLOYMENT_ID assert get_snapshot_runtime() is runtime class TestLocalPlatformFields: def _read(self, tmp_path, monkeypatch, text: str | None): config_path = tmp_path / "config.yaml" if text is not None: config_path.write_text(text, encoding="utf-8") monkeypatch.setattr( "EvoScientist.model_registry.runtime.get_config_path", lambda: config_path ) return _read_local_platform_fields() def test_missing_config_uses_defaults(self, tmp_path, monkeypatch): deployment_id, db_path, endpoints, webui_ids = self._read( tmp_path, monkeypatch, None ) assert deployment_id == DEFAULT_LOCAL_DEPLOYMENT_ID assert db_path is None assert endpoints == () assert webui_ids == () def test_reads_runtime_fields(self, tmp_path, monkeypatch): deployment_id, db_path, endpoints, webui_ids = self._read( tmp_path, monkeypatch, "local_deployment_id: dev-deploy\n" "model_runtime_db: /tmp/mr.sqlite3\n" "development_endpoints:\n" " - {id: ollama, url: 'http://localhost:11434', label: Local}\n" "webui_delegation_public_keys:\n" " - {deployment_id: webui-local, public_key: '-----BEGIN PUBLIC KEY-----\\nMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE\\n-----END PUBLIC KEY-----\\n'}\n", ) assert deployment_id == "dev-deploy" assert db_path == __import__("pathlib").Path("/tmp/mr.sqlite3") assert endpoints == ( DevelopmentEndpoint( id="ollama", url="http://localhost:11434", label="Local" ), ) assert webui_ids == ("webui-local",) def test_invalid_yaml_raises(self, tmp_path, monkeypatch): with pytest.raises(PlatformConfigError): self._read(tmp_path, monkeypatch, "local_deployment_id: [unclosed") def test_invalid_field_type_raises(self, tmp_path, monkeypatch): with pytest.raises(PlatformConfigError): self._read(tmp_path, monkeypatch, "local_deployment_id: 42") class TestEndpointPolicyWiring: def test_development_endpoints_allow_loopback_ollama(self, store): policy = EndpointPolicy( ( DevelopmentEndpoint( id="ollama", url="http://localhost:11434", label="Local" ), ) ) runtime = SnapshotRuntime(store, endpoint_policy=policy) # The loopback ollama provider builds only because its base URL is a # registered development endpoint. config = runtime.resolver.resolve(OLLAMA_REF, "primary") model = runtime._build(config, "") assert isinstance(model, ChatOllama) class TestCreateLocalSnapshot: """Section 8.1 local entry: fixed binding convention through SnapshotService.""" def test_bootstrap_registry_fails_closed(self, tmp_path): runtime = SnapshotRuntime(ModelRuntimeStore(config_dir=tmp_path / "db")) with pytest.raises(ModelRegistryError) as excinfo: runtime.create_local_snapshot("thread-1") assert excinfo.value.code == MODEL_REGISTRY_NOT_READY def test_binding_convention_is_fixed(self, store, runtime): from EvoScientist.model_registry.snapshots import config_for_role snapshot = runtime.create_local_snapshot("cli-thread-1") assert snapshot.thread_id == "cli-thread-1" assert snapshot.deployment_id == runtime.local_deployment_id assert snapshot.payload.model_selection_revision == 0 # primary=None inherits — the registry default is frozen at creation. assert config_for_role(snapshot, "primary").model_ref == ZHIPU_REF def test_same_run_request_id_reuses_snapshot(self, store, runtime): first = runtime.create_local_snapshot("t-1", run_request_id="req-1") second = runtime.create_local_snapshot("t-1", run_request_id="req-1") assert second.snapshot_id == first.snapshot_id def test_default_run_request_id_freezes_per_call(self, store, runtime): first = runtime.create_local_snapshot("t-1") second = runtime.create_local_snapshot("t-1") assert second.snapshot_id != first.snapshot_id