"""Unit tests for the plugin-backed model runtime adapter.""" import datetime import uuid from types import SimpleNamespace from unittest.mock import Mock, patch, sentinel import pytest from core.plugin.entities.plugin_daemon import PluginModelProviderEntity from core.plugin.impl import model_runtime as model_runtime_module from core.plugin.impl.model import PluginModelClient from core.plugin.impl.model_runtime import TENANT_SCOPE_SCHEMA_CACHE_USER_ID, PluginModelRuntime from core.plugin.impl.model_runtime_factory import create_plugin_model_runtime from core.plugin.plugin_service import PluginService from graphon.model_runtime.entities.common_entities import I18nObject from graphon.model_runtime.entities.llm_entities import LLMResultChunk, LLMResultChunkDelta, LLMUsage from graphon.model_runtime.entities.message_entities import AssistantPromptMessage from graphon.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType from graphon.model_runtime.entities.provider_entities import ConfigurateMethod, ProviderEntity class _FakeRedis: def __init__(self) -> None: self._values: dict[str, str] = {} self.setex_calls: list[tuple[str, int, str]] = [] def get(self, key: str) -> str | None: return self._values.get(key) def setex(self, key: str, ttl: int, value: str) -> None: self._values[key] = value self.setex_calls.append((key, ttl, value)) def delete(self, key: str) -> None: self._values.pop(key, None) def _build_model_schema() -> AIModelEntity: return AIModelEntity( model="gpt-4o-mini", label=I18nObject(en_US="GPT-4o mini"), model_type=ModelType.LLM, fetch_from=FetchFrom.PREDEFINED_MODEL, model_properties={}, ) def _build_plugin_model_provider(*, tenant_id: str, provider: str = "openai") -> PluginModelProviderEntity: return PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider=provider, tenant_id=tenant_id, plugin_unique_identifier=f"langgenius/{provider}/{provider}", plugin_id=f"langgenius/{provider}", declaration=ProviderEntity( provider=provider, label=I18nObject(en_US=provider.title()), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ) class TestPluginModelRuntime: """Validate the adapter keeps plugin-specific routing out of the runtime port.""" def test_fetch_model_providers_returns_runtime_entities(self) -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [ PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="openai", tenant_id="tenant", plugin_unique_identifier="langgenius/openai/openai", plugin_id="langgenius/openai", declaration=ProviderEntity( provider="openai", label=I18nObject(en_US="OpenAI"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ) ] runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) providers = runtime.fetch_model_providers() assert len(providers) == 1 assert providers[0].provider == "langgenius/openai/openai" assert providers[0].provider_name == "openai" assert providers[0].label.en_us == "OpenAI" client.fetch_model_providers.assert_called_once_with("tenant") def test_fetch_model_providers_only_exposes_short_name_for_canonical_provider(self) -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [ PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="openai", tenant_id="tenant", plugin_unique_identifier="acme/openai/openai", plugin_id="acme/openai", declaration=ProviderEntity( provider="openai", label=I18nObject(en_US="Acme OpenAI"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ), PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="openai", tenant_id="tenant", plugin_unique_identifier="langgenius/openai/openai", plugin_id="langgenius/openai", declaration=ProviderEntity( provider="openai", label=I18nObject(en_US="OpenAI"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ), ] runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) providers = runtime.fetch_model_providers() provider_aliases = {provider.provider: provider.provider_name for provider in providers} assert provider_aliases["acme/openai/openai"] == "" assert provider_aliases["langgenius/openai/openai"] == "openai" def test_fetch_model_providers_keeps_google_alias_on_canonical_gemini_provider(self) -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [ PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="google", tenant_id="tenant", plugin_unique_identifier="langgenius/gemini/google", plugin_id="langgenius/gemini", declaration=ProviderEntity( provider="google", label=I18nObject(en_US="Google"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ) ] runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) providers = runtime.fetch_model_providers() assert providers[0].provider == "langgenius/gemini/google" assert providers[0].provider_name == "google" def test_validate_provider_credentials_resolves_plugin_fields(self) -> None: client = Mock(spec=PluginModelClient) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) runtime.validate_provider_credentials( provider="langgenius/openai/openai", credentials={"api_key": "secret"}, ) client.validate_provider_credentials.assert_called_once_with( tenant_id="tenant", user_id="user", plugin_id="langgenius/openai", provider="openai", credentials={"api_key": "secret"}, ) def test_invoke_llm_resolves_plugin_fields(self) -> None: client = Mock(spec=PluginModelClient) usage = LLMUsage.empty_usage() client.invoke_llm.return_value = iter( [ LLMResultChunk( model="gpt-4o-mini", prompt_messages=[], system_fingerprint="fp-plugin", delta=LLMResultChunkDelta( index=0, message=AssistantPromptMessage(content="plugin "), ), ), LLMResultChunk( model="gpt-4o-mini", prompt_messages=[], system_fingerprint="fp-plugin", delta=LLMResultChunkDelta( index=1, message=AssistantPromptMessage(content="response"), usage=usage, finish_reason="stop", ), ), ] ) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) result = runtime.invoke_llm( provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={"temperature": 0.3}, prompt_messages=[], tools=None, stop=None, stream=False, ) assert result.model == "gpt-4o-mini" assert result.prompt_messages == [] assert result.message.content == "plugin response" assert result.usage == usage assert result.system_fingerprint == "fp-plugin" client.invoke_llm.assert_called_once_with( tenant_id="tenant", user_id="user", plugin_id="langgenius/openai", provider="openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={"temperature": 0.3}, prompt_messages=[], tools=None, stop=None, stream=False, ) def test_invoke_llm_returns_plugin_stream_directly(self) -> None: client = Mock(spec=PluginModelClient) stream_result = iter([]) client.invoke_llm.return_value = stream_result runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) result = runtime.invoke_llm( provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={"temperature": 0.3}, prompt_messages=[], tools=None, stop=("END",), stream=True, ) assert result is stream_result client.invoke_llm.assert_called_once_with( tenant_id="tenant", user_id="user", plugin_id="langgenius/openai", provider="openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={"temperature": 0.3}, prompt_messages=[], tools=None, stop=["END"], stream=True, ) def test_invoke_llm_rejects_per_call_user_override(self) -> None: client = Mock(spec=PluginModelClient) client.invoke_llm.return_value = sentinel.result runtime = PluginModelRuntime( tenant_id="tenant", user_id="bound-user", client=client, plugin_service=PluginService ) with pytest.raises(TypeError, match="unexpected keyword argument 'user_id'"): runtime.invoke_llm( # type: ignore[call-arg] provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={"temperature": 0.3}, prompt_messages=[], tools=None, stop=None, stream=False, user_id="request-user", ) client.invoke_llm.assert_not_called() def test_invoke_tts_uses_bound_runtime_user_when_runtime_is_unbound(self) -> None: client = Mock(spec=PluginModelClient) client.invoke_tts.return_value = iter([b"chunk"]) runtime = PluginModelRuntime(tenant_id="tenant", user_id=None, client=client, plugin_service=PluginService) result = runtime.invoke_tts( provider="langgenius/openai/openai", model="tts-1", credentials={"api_key": "secret"}, content_text="hello", voice="alloy", ) assert list(result) == [b"chunk"] client.invoke_tts.assert_called_once_with( tenant_id="tenant", user_id=None, plugin_id="langgenius/openai", provider="openai", model="tts-1", credentials={"api_key": "secret"}, content_text="hello", voice="alloy", ) def test_fetch_model_providers_does_not_keep_bound_runtime_cache(self, monkeypatch: pytest.MonkeyPatch) -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [] from core.plugin import plugin_service as plugin_service_module monkeypatch.setattr( plugin_service_module, "redis_client", SimpleNamespace( get=Mock(return_value=None), delete=Mock(), setex=Mock(), ), ) monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 300) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) runtime.fetch_model_providers() runtime.fetch_model_providers() assert client.fetch_model_providers.call_count == 2 def test_fetch_model_providers_uses_tenant_ttl_cache_across_runtime_instances( self, monkeypatch: pytest.MonkeyPatch ) -> None: redis = _FakeRedis() from core.plugin import plugin_service as plugin_service_module monkeypatch.setattr(plugin_service_module, "redis_client", redis) monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 300) first_client = Mock(spec=PluginModelClient) first_client.fetch_model_providers.return_value = [_build_plugin_model_provider(tenant_id="tenant")] second_client = Mock(spec=PluginModelClient) first_runtime = PluginModelRuntime( tenant_id="tenant", user_id="user-a", client=first_client, plugin_service=PluginService ) second_runtime = PluginModelRuntime( tenant_id="tenant", user_id="user-b", client=second_client, plugin_service=PluginService ) first_providers = first_runtime.fetch_model_providers() second_providers = second_runtime.fetch_model_providers() assert [provider.provider for provider in first_providers] == ["langgenius/openai/openai"] assert [provider.provider for provider in second_providers] == ["langgenius/openai/openai"] first_client.fetch_model_providers.assert_called_once_with("tenant") second_client.fetch_model_providers.assert_not_called() assert redis.setex_calls[0][1] == 300 def test_fetch_model_providers_cache_is_tenant_isolated(self, monkeypatch: pytest.MonkeyPatch) -> None: redis = _FakeRedis() from core.plugin import plugin_service as plugin_service_module monkeypatch.setattr(plugin_service_module, "redis_client", redis) monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 300) first_client = Mock(spec=PluginModelClient) first_client.fetch_model_providers.return_value = [_build_plugin_model_provider(tenant_id="tenant-a")] second_client = Mock(spec=PluginModelClient) second_client.fetch_model_providers.return_value = [_build_plugin_model_provider(tenant_id="tenant-b")] first_runtime = PluginModelRuntime( tenant_id="tenant-a", user_id="user", client=first_client, plugin_service=PluginService ) second_runtime = PluginModelRuntime( tenant_id="tenant-b", user_id="user", client=second_client, plugin_service=PluginService ) first_providers = first_runtime.fetch_model_providers() second_providers = second_runtime.fetch_model_providers() assert [provider.provider for provider in first_providers] == ["langgenius/openai/openai"] assert [provider.provider for provider in second_providers] == ["langgenius/openai/openai"] first_client.fetch_model_providers.assert_called_once_with("tenant-a") second_client.fetch_model_providers.assert_called_once_with("tenant-b") assert len(redis.setex_calls) == 2 def test_fetch_model_providers_delegates_cache_to_injected_plugin_service(self) -> None: client = Mock(spec=PluginModelClient) service_result = [ ProviderEntity( provider="langgenius/openai/openai", label=I18nObject(en_US="OpenAI"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ) ] fetch_plugin_model_providers = Mock(return_value=service_result) class TestPluginService(PluginService): pass TestPluginService.fetch_plugin_model_providers = fetch_plugin_model_providers runtime = PluginModelRuntime( tenant_id="tenant", user_id="user", client=client, plugin_service=TestPluginService ) result = runtime.fetch_model_providers() assert result is service_result fetch_plugin_model_providers.assert_called_once_with(tenant_id="tenant", client=client) client.fetch_model_providers.assert_not_called() def test_create_plugin_model_runtime_without_user_context() -> None: runtime = create_plugin_model_runtime(tenant_id="tenant") assert runtime.user_id is None def test_plugin_model_runtime_requires_client() -> None: with pytest.raises(ValueError, match="client is required"): PluginModelRuntime(tenant_id="tenant", user_id="user", client=None, plugin_service=PluginService) # type: ignore[arg-type] def test_plugin_model_runtime_requires_plugin_service() -> None: with pytest.raises(ValueError, match="plugin_service is required"): PluginModelRuntime( tenant_id="tenant", user_id="user", client=Mock(spec=PluginModelClient), plugin_service=None, # type: ignore[arg-type] ) def test_get_model_schema_uses_cached_schema_without_hitting_client(monkeypatch: pytest.MonkeyPatch) -> None: client = Mock(spec=PluginModelClient) schema = _build_model_schema() monkeypatch.setattr( model_runtime_module, "redis_client", SimpleNamespace( get=Mock(return_value=schema.model_dump_json()), delete=Mock(), setex=Mock(), ), ) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) result = runtime.get_model_schema( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"api_key": "secret"}, ) assert result == schema client.get_model_schema.assert_not_called() def test_structured_output_adapter_invokes_bound_runtime_streaming() -> None: runtime = Mock() runtime.invoke_llm.return_value = sentinel.stream_result adapter = model_runtime_module._PluginStructuredOutputModelInstance( runtime=runtime, provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, ) tool = Mock() result = adapter.invoke_llm( prompt_messages=[], model_parameters=None, tools=[tool], stop=["END"], stream=True, callbacks=sentinel.callbacks, ) assert result is sentinel.stream_result runtime.invoke_llm.assert_called_once_with( provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={}, prompt_messages=[], tools=[tool], stop=["END"], stream=True, ) def test_structured_output_adapter_invokes_bound_runtime_non_streaming() -> None: runtime = Mock() runtime.invoke_llm.return_value = sentinel.result adapter = model_runtime_module._PluginStructuredOutputModelInstance( runtime=runtime, provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, ) result = adapter.invoke_llm( prompt_messages=[], model_parameters={"temperature": 0}, tools=None, stop=None, stream=False, ) assert result is sentinel.result runtime.invoke_llm.assert_called_once_with( provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, model_parameters={"temperature": 0}, prompt_messages=[], tools=None, stop=None, stream=False, ) def test_invoke_llm_with_structured_output_delegates_with_bound_adapter() -> None: client = Mock(spec=PluginModelClient) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) schema = _build_model_schema() runtime.get_model_schema = Mock(return_value=schema) # type: ignore[method-assign] with patch.object( model_runtime_module, "invoke_llm_with_structured_output_helper", return_value=sentinel.structured_result, ) as mock_helper: result = runtime.invoke_llm_with_structured_output( provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, json_schema={"type": "object"}, model_parameters={"temperature": 0}, prompt_messages=[], stop=("END",), stream=False, ) assert result is sentinel.structured_result runtime.get_model_schema.assert_called_once_with( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"api_key": "secret"}, ) helper_kwargs = mock_helper.call_args.kwargs assert helper_kwargs["provider"] == "langgenius/openai/openai" assert helper_kwargs["model_schema"] == schema assert helper_kwargs["json_schema"] == {"type": "object"} assert helper_kwargs["model_parameters"] == {"temperature": 0} assert helper_kwargs["prompt_messages"] == [] assert helper_kwargs["tools"] is None assert helper_kwargs["stop"] == ["END"] assert helper_kwargs["stream"] is False assert isinstance(helper_kwargs["model_instance"], model_runtime_module._PluginStructuredOutputModelInstance) def test_invoke_llm_with_structured_output_raises_when_model_schema_is_missing() -> None: client = Mock(spec=PluginModelClient) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) runtime.get_model_schema = Mock(return_value=None) # type: ignore[method-assign] with pytest.raises(ValueError, match="Model schema not found for gpt-4o-mini"): runtime.invoke_llm_with_structured_output( provider="langgenius/openai/openai", model="gpt-4o-mini", credentials={"api_key": "secret"}, json_schema={"type": "object"}, model_parameters={}, prompt_messages=[], stop=None, stream=False, ) def test_get_model_schema_deletes_invalid_cache_and_refetches(monkeypatch: pytest.MonkeyPatch) -> None: client = Mock(spec=PluginModelClient) schema = _build_model_schema() delete = Mock() setex = Mock() monkeypatch.setattr( model_runtime_module, "redis_client", SimpleNamespace( get=Mock(return_value="not-json"), delete=delete, setex=setex, ), ) monkeypatch.setattr(model_runtime_module.dify_config, "PLUGIN_MODEL_SCHEMA_CACHE_TTL", 300) client.get_model_schema.return_value = schema runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) result = runtime.get_model_schema( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"api_key": "secret"}, ) assert result == schema delete.assert_called_once() client.get_model_schema.assert_called_once_with( tenant_id="tenant", user_id="user", plugin_id="langgenius/openai", provider="openai", model_type=ModelType.LLM.value, model="gpt-4o-mini", credentials={"api_key": "secret"}, ) setex.assert_called_once() def test_get_llm_num_tokens_returns_zero_when_plugin_counting_is_disabled(monkeypatch: pytest.MonkeyPatch) -> None: client = Mock(spec=PluginModelClient) monkeypatch.setattr(model_runtime_module.dify_config, "PLUGIN_BASED_TOKEN_COUNTING_ENABLED", False) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) assert ( runtime.get_llm_num_tokens( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"api_key": "secret"}, prompt_messages=[], tools=None, ) == 0 ) client.get_llm_num_tokens.assert_not_called() def test_get_provider_icon_reads_requested_variant_and_detects_svg_mime(monkeypatch: pytest.MonkeyPatch) -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [ PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="openai", tenant_id="tenant", plugin_unique_identifier="langgenius/openai/openai", plugin_id="langgenius/openai", declaration=ProviderEntity( provider="openai", label=I18nObject(en_US="OpenAI"), icon_small=I18nObject(en_US="logo.svg"), icon_small_dark=I18nObject(en_US="logo-dark.png"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ) ] fetch_asset = Mock(return_value=b"") monkeypatch.setattr(model_runtime_module.PluginAssetManager, "fetch_asset", fetch_asset) runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) icon_bytes, mime_type = runtime.get_provider_icon( provider="langgenius/openai/openai", icon_type="icon_small", lang="en_US", ) assert icon_bytes == b"" assert mime_type == "image/svg+xml" fetch_asset.assert_called_once_with(tenant_id="tenant", id="logo.svg") def test_get_provider_icon_rejects_unsupported_types_and_missing_variants() -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [ PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="openai", tenant_id="tenant", plugin_unique_identifier="langgenius/openai/openai", plugin_id="langgenius/openai", declaration=ProviderEntity( provider="openai", label=I18nObject(en_US="OpenAI"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ) ] runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) with pytest.raises(ValueError, match="does not have small dark icon"): runtime.get_provider_icon( provider="langgenius/openai/openai", icon_type="icon_small_dark", lang="en_US", ) with pytest.raises(ValueError, match="Unsupported icon type"): runtime.get_provider_icon( provider="langgenius/openai/openai", icon_type="icon_large", lang="en_US", ) def test_get_schema_cache_key_is_stable_across_credential_order() -> None: runtime = PluginModelRuntime( tenant_id="tenant", user_id="user", client=Mock(spec=PluginModelClient), plugin_service=PluginService ) first = runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"b": "2", "a": "1"}, ) second = runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"a": "1", "b": "2"}, ) assert first == second def test_get_schema_cache_key_separates_distinct_user_scopes() -> None: first_runtime = PluginModelRuntime( tenant_id="tenant", user_id="user-a", client=Mock(spec=PluginModelClient), plugin_service=PluginService ) second_runtime = PluginModelRuntime( tenant_id="tenant", user_id="user-b", client=Mock(spec=PluginModelClient), plugin_service=PluginService ) first = first_runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"a": "1"}, ) second = second_runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"a": "1"}, ) assert first != second def test_get_schema_cache_key_separates_tenant_scope_from_user_scope() -> None: tenant_runtime = PluginModelRuntime( tenant_id="tenant", user_id=None, client=Mock(spec=PluginModelClient), plugin_service=PluginService ) user_runtime = PluginModelRuntime( tenant_id="tenant", user_id="user-a", client=Mock(spec=PluginModelClient), plugin_service=PluginService ) tenant_key = tenant_runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"a": "1"}, ) user_key = user_runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={"a": "1"}, ) assert tenant_key != user_key assert f":{TENANT_SCOPE_SCHEMA_CACHE_USER_ID}" in tenant_key def test_get_schema_cache_key_separates_tenant_scope_from_empty_string_user_scope() -> None: tenant_runtime = PluginModelRuntime( tenant_id="tenant", user_id=None, client=Mock(spec=PluginModelClient), plugin_service=PluginService ) empty_user_runtime = PluginModelRuntime( tenant_id="tenant", user_id="", client=Mock(spec=PluginModelClient), plugin_service=PluginService ) tenant_key = tenant_runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={}, ) empty_user_key = empty_user_runtime._get_schema_cache_key( provider="langgenius/openai/openai", model_type=ModelType.LLM, model="gpt-4o-mini", credentials={}, ) assert tenant_key != empty_user_key assert empty_user_key.endswith(":") assert TENANT_SCOPE_SCHEMA_CACHE_USER_ID not in empty_user_key def test_get_provider_schema_supports_short_alias_and_rejects_invalid_provider() -> None: client = Mock(spec=PluginModelClient) client.fetch_model_providers.return_value = [ PluginModelProviderEntity( id=uuid.uuid4().hex, created_at=datetime.datetime.now(), updated_at=datetime.datetime.now(), provider="openai", tenant_id="tenant", plugin_unique_identifier="langgenius/openai/openai", plugin_id="langgenius/openai", declaration=ProviderEntity( provider="openai", label=I18nObject(en_US="OpenAI"), supported_model_types=[], configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL], ), ) ] runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService) assert runtime._get_provider_schema("openai").provider == "langgenius/openai/openai" with pytest.raises(ValueError, match="Invalid provider"): runtime._get_provider_schema("missing")