mirror of
https://github.com/langgenius/dify.git
synced 2026-09-07 01:43:41 +08:00
1039 lines
42 KiB
Python
1039 lines
42 KiB
Python
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from core.entities.model_entities import ModelStatus
|
|
from core.entities.provider_entities import CredentialConfiguration
|
|
from core.plugin.entities.plugin import PluginInstallationSource
|
|
from core.plugin.entities.plugin_daemon import PluginModelProviderBinding
|
|
from enums import DeploymentEdition
|
|
from graphon.model_runtime.entities.common_entities import I18nObject
|
|
from graphon.model_runtime.entities.model_entities import FetchFrom, ModelType, ParameterRule, ParameterType
|
|
from graphon.model_runtime.entities.provider_entities import ConfigurateMethod
|
|
from models.provider import ProviderType
|
|
from services import model_provider_service as service_module
|
|
from services.errors.app_model_config import ProviderNotFoundError
|
|
from services.model_provider_service import ModelProviderService, _ProviderSummaryState
|
|
|
|
|
|
def _create_service_with_mocked_manager() -> tuple[ModelProviderService, MagicMock]:
|
|
manager = MagicMock()
|
|
service = ModelProviderService()
|
|
service._get_provider_manager = MagicMock(return_value=manager)
|
|
return service, manager
|
|
|
|
|
|
def _build_provider_configuration(
|
|
*,
|
|
provider_name: str = "openai",
|
|
supported_model_types: list[ModelType] | None = None,
|
|
custom_models: list[Any] | None = None,
|
|
custom_config_available: bool = True,
|
|
) -> SimpleNamespace:
|
|
if supported_model_types is None:
|
|
supported_model_types = [ModelType.LLM]
|
|
|
|
return SimpleNamespace(
|
|
provider=SimpleNamespace(
|
|
provider=provider_name,
|
|
label=I18nObject(en_US=provider_name),
|
|
description=None,
|
|
icon_small=None,
|
|
icon_small_dark=None,
|
|
background=None,
|
|
help=None,
|
|
supported_model_types=supported_model_types,
|
|
configurate_methods=[],
|
|
provider_credential_schema=None,
|
|
model_credential_schema=None,
|
|
),
|
|
preferred_provider_type=ProviderType.CUSTOM,
|
|
custom_configuration=SimpleNamespace(
|
|
provider=SimpleNamespace(
|
|
current_credential_id="cred-1",
|
|
current_credential_name="Credential 1",
|
|
available_credentials=[],
|
|
),
|
|
models=custom_models,
|
|
can_added_models=[],
|
|
),
|
|
system_configuration=SimpleNamespace(enabled=False, current_quota_type=None, quota_configurations=[]),
|
|
is_custom_configuration_available=lambda: custom_config_available,
|
|
)
|
|
|
|
|
|
def _build_model_provider_binding(
|
|
source: PluginInstallationSource,
|
|
*,
|
|
installation_id: str = "installation-1",
|
|
plugin_id: str = "langgenius/openai",
|
|
plugin_unique_identifier: str = "langgenius/openai:1.0.0@checksum",
|
|
verified: bool = True,
|
|
) -> PluginModelProviderBinding:
|
|
return PluginModelProviderBinding(
|
|
provider="openai",
|
|
installation_id=installation_id,
|
|
plugin_id=plugin_id,
|
|
plugin_unique_identifier=plugin_unique_identifier,
|
|
runtime_type="remote" if source == PluginInstallationSource.Remote else "local",
|
|
source=source,
|
|
version="1.0.0",
|
|
verified=verified,
|
|
)
|
|
|
|
|
|
class TestModelProviderServiceConfiguration:
|
|
def test__get_provider_configuration_should_return_configuration_when_provider_exists(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
provider_configuration = SimpleNamespace(name="provider-config")
|
|
manager.get_configurations.return_value = {"openai": provider_configuration}
|
|
|
|
result = service._get_provider_configuration(tenant_id="tenant-1", provider="openai")
|
|
|
|
assert result is provider_configuration
|
|
|
|
def test__get_provider_configuration_should_raise_error_when_provider_is_missing(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
manager.get_configurations.return_value = {}
|
|
|
|
with pytest.raises(ProviderNotFoundError, match="does not exist"):
|
|
service._get_provider_configuration(tenant_id="tenant-1", provider="missing")
|
|
|
|
def test_get_provider_list_should_filter_by_model_type_and_build_no_configure_status(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
allowed = _build_provider_configuration(
|
|
provider_name="openai",
|
|
supported_model_types=[ModelType.LLM],
|
|
custom_config_available=False,
|
|
)
|
|
filtered = _build_provider_configuration(
|
|
provider_name="embedding",
|
|
supported_model_types=[ModelType.TEXT_EMBEDDING],
|
|
custom_config_available=True,
|
|
)
|
|
manager.get_configurations.return_value = {"openai": allowed, "embedding": filtered}
|
|
|
|
result = service.get_provider_list(tenant_id="tenant-1", model_type=ModelType.LLM)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].provider == "openai"
|
|
assert result[0].custom_configuration.status.value == "no-configure"
|
|
|
|
def test_get_provider_summary_list_uses_lightweight_state_and_plugin_bindings(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider = SimpleNamespace(
|
|
provider="langgenius/openai/openai",
|
|
label=I18nObject(en_US="OpenAI"),
|
|
description=I18nObject(en_US="OpenAI models"),
|
|
icon_small=I18nObject(en_US="icon.svg"),
|
|
icon_small_dark=I18nObject(en_US="icon-dark.svg"),
|
|
supported_model_types=[ModelType.LLM],
|
|
configurate_methods=[ConfigurateMethod.PREDEFINED_MODEL],
|
|
)
|
|
binding = SimpleNamespace(
|
|
provider="openai",
|
|
plugin_id="langgenius/openai",
|
|
installation_id="installation-1",
|
|
plugin_unique_identifier="langgenius/openai:1.2.3@checksum",
|
|
runtime_type="local",
|
|
source=PluginInstallationSource.Marketplace,
|
|
version="1.2.3",
|
|
verified=True,
|
|
)
|
|
state = _ProviderSummaryState(
|
|
has_custom_provider=True,
|
|
available_credentials=[
|
|
CredentialConfiguration(
|
|
credential_id="credential-1",
|
|
credential_name="Production",
|
|
),
|
|
CredentialConfiguration(
|
|
credential_id="credential-2",
|
|
credential_name="Backup",
|
|
),
|
|
],
|
|
has_custom_models=True,
|
|
current_credential_id="credential-1",
|
|
current_credential_name="Production",
|
|
current_credential_usable=True,
|
|
preferred_provider_type=ProviderType.CUSTOM,
|
|
)
|
|
call_order: list[str] = []
|
|
manager_constructor = MagicMock(side_effect=AssertionError("summary must not construct ProviderManager"))
|
|
monkeypatch.setattr(service, "_get_provider_manager", manager_constructor)
|
|
monkeypatch.setattr(
|
|
service_module.PluginService,
|
|
"list_model_provider_bindings",
|
|
MagicMock(side_effect=lambda *_args, **_kwargs: call_order.append("bindings") or [binding]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service_module.PluginService,
|
|
"fetch_plugin_model_providers",
|
|
MagicMock(side_effect=lambda *_args, **_kwargs: call_order.append("providers") or [provider]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_load_provider_summary_states",
|
|
MagicMock(return_value={provider.provider: state}),
|
|
)
|
|
monkeypatch.setattr(
|
|
service_module.ext_hosting_provider.hosting_configuration,
|
|
"provider_map",
|
|
{provider.provider: SimpleNamespace(enabled=True, quotas=[SimpleNamespace()])},
|
|
)
|
|
|
|
providers, plugins = service.get_provider_summary_list(tenant_id="tenant-1")
|
|
|
|
assert len(providers) == 1
|
|
assert providers[0].provider == provider.provider
|
|
assert providers[0].plugin_id == "langgenius/openai"
|
|
assert providers[0].is_configured is True
|
|
assert providers[0].custom_configuration.available_credentials == [
|
|
CredentialConfiguration(
|
|
credential_id="credential-1",
|
|
credential_name="Production",
|
|
),
|
|
CredentialConfiguration(
|
|
credential_id="credential-2",
|
|
credential_name="Backup",
|
|
),
|
|
]
|
|
assert providers[0].custom_configuration.has_custom_models is True
|
|
assert providers[0].custom_configuration.current_credential_name == "Production"
|
|
assert providers[0].custom_configuration.current_credential_usable is True
|
|
assert providers[0].system_configuration.enabled is True
|
|
assert plugins["langgenius/openai"].installation_id == "installation-1"
|
|
assert plugins["langgenius/openai"].version == "1.2.3"
|
|
assert call_order == ["bindings", "providers"]
|
|
manager_constructor.assert_not_called()
|
|
|
|
def test_get_provider_summary_list_enables_system_only_for_verified_hosted_non_package_bindings(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
marketplace_binding = _build_model_provider_binding(PluginInstallationSource.Marketplace)
|
|
package_binding = _build_model_provider_binding(
|
|
PluginInstallationSource.Package,
|
|
installation_id="installation-package",
|
|
plugin_id="langgenius/package",
|
|
plugin_unique_identifier="langgenius/package:1.0.0@checksum",
|
|
)
|
|
unhosted_binding = _build_model_provider_binding(
|
|
PluginInstallationSource.Marketplace,
|
|
installation_id="installation-unhosted",
|
|
plugin_id="langgenius/unhosted",
|
|
plugin_unique_identifier="langgenius/unhosted:1.0.0@checksum",
|
|
)
|
|
remote_binding = _build_model_provider_binding(
|
|
PluginInstallationSource.Remote,
|
|
installation_id="installation-remote",
|
|
plugin_id="langgenius/remote",
|
|
plugin_unique_identifier="langgenius/remote:1.0.0@checksum",
|
|
)
|
|
unverified_binding = _build_model_provider_binding(
|
|
PluginInstallationSource.Marketplace,
|
|
installation_id="installation-unverified",
|
|
plugin_id="langgenius/unverified",
|
|
plugin_unique_identifier="langgenius/unverified:1.0.0@checksum",
|
|
verified=False,
|
|
)
|
|
bindings = [
|
|
marketplace_binding,
|
|
package_binding,
|
|
unhosted_binding,
|
|
remote_binding,
|
|
unverified_binding,
|
|
]
|
|
provider_entities = [
|
|
SimpleNamespace(
|
|
provider=f"{binding.plugin_id}/openai",
|
|
label=I18nObject(en_US=binding.plugin_id),
|
|
description=None,
|
|
icon_small=None,
|
|
icon_small_dark=None,
|
|
supported_model_types=[ModelType.LLM],
|
|
configurate_methods=[],
|
|
)
|
|
for binding in bindings
|
|
]
|
|
monkeypatch.setattr(
|
|
service_module.ext_hosting_provider.hosting_configuration,
|
|
"provider_map",
|
|
{
|
|
"langgenius/openai/openai": SimpleNamespace(enabled=True, quotas=[SimpleNamespace()]),
|
|
"langgenius/package/openai": SimpleNamespace(enabled=True, quotas=[SimpleNamespace()]),
|
|
"langgenius/remote/openai": SimpleNamespace(enabled=True, quotas=[SimpleNamespace()]),
|
|
"langgenius/unverified/openai": SimpleNamespace(enabled=True, quotas=[SimpleNamespace()]),
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
service_module.PluginService,
|
|
"list_model_provider_bindings",
|
|
MagicMock(return_value=bindings),
|
|
)
|
|
monkeypatch.setattr(
|
|
service_module.PluginService,
|
|
"fetch_plugin_model_providers",
|
|
MagicMock(return_value=provider_entities),
|
|
)
|
|
monkeypatch.setattr(ModelProviderService, "_load_provider_summary_states", MagicMock(return_value={}))
|
|
monkeypatch.setattr(service_module, "is_filtered", MagicMock(return_value=False))
|
|
|
|
providers, _ = ModelProviderService().get_provider_summary_list("tenant-1")
|
|
|
|
assert {provider.provider: provider.system_configuration.enabled for provider in providers} == {
|
|
"langgenius/openai/openai": True,
|
|
"langgenius/package/openai": False,
|
|
"langgenius/unhosted/openai": False,
|
|
"langgenius/remote/openai": True,
|
|
"langgenius/unverified/openai": False,
|
|
}
|
|
|
|
def test_model_provider_binding_without_verified_field_fails_closed(self) -> None:
|
|
binding = PluginModelProviderBinding.model_validate(
|
|
{
|
|
"provider": "openai",
|
|
"installation_id": "installation-1",
|
|
"plugin_id": "langgenius/openai",
|
|
"plugin_unique_identifier": "langgenius/openai:1.0.0@checksum",
|
|
"runtime_type": "local",
|
|
"source": "marketplace",
|
|
"version": "1.0.0",
|
|
}
|
|
)
|
|
|
|
assert binding.verified is False
|
|
|
|
def test_get_provider_summary_list_returns_all_unique_provider_metadata(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
llm_provider = SimpleNamespace(
|
|
provider="langgenius/openai/openai",
|
|
label=I18nObject(en_US="OpenAI"),
|
|
description=None,
|
|
icon_small=None,
|
|
icon_small_dark=None,
|
|
supported_model_types=[ModelType.LLM],
|
|
configurate_methods=[],
|
|
)
|
|
embedding_provider = SimpleNamespace(
|
|
provider="langgenius/embedding/embedding",
|
|
label=I18nObject(en_US="Embedding"),
|
|
description=None,
|
|
icon_small=None,
|
|
icon_small_dark=None,
|
|
supported_model_types=[ModelType.TEXT_EMBEDDING],
|
|
configurate_methods=[],
|
|
)
|
|
llm_binding = SimpleNamespace(
|
|
provider="openai",
|
|
plugin_id="langgenius/openai",
|
|
installation_id="installation-openai",
|
|
plugin_unique_identifier="langgenius/openai:1.0.0@checksum",
|
|
runtime_type="local",
|
|
source=service_module.PluginInstallationSource.Marketplace,
|
|
version="1.0.0",
|
|
verified=False,
|
|
)
|
|
embedding_binding = SimpleNamespace(
|
|
provider="embedding",
|
|
plugin_id="langgenius/embedding",
|
|
installation_id="installation-embedding",
|
|
plugin_unique_identifier="langgenius/embedding:1.0.0@checksum",
|
|
runtime_type="local",
|
|
source=service_module.PluginInstallationSource.Marketplace,
|
|
version="1.0.0",
|
|
verified=False,
|
|
)
|
|
monkeypatch.setattr(
|
|
service_module.PluginService,
|
|
"list_model_provider_bindings",
|
|
MagicMock(return_value=[llm_binding, embedding_binding]),
|
|
)
|
|
monkeypatch.setattr(
|
|
service_module.PluginService,
|
|
"fetch_plugin_model_providers",
|
|
MagicMock(return_value=[llm_provider, llm_provider, embedding_provider]),
|
|
)
|
|
monkeypatch.setattr(service, "_load_provider_summary_states", MagicMock(return_value={}))
|
|
monkeypatch.setattr(service_module, "is_filtered", MagicMock(return_value=False))
|
|
|
|
providers, plugins = service.get_provider_summary_list(tenant_id="tenant-1")
|
|
|
|
assert [provider.provider for provider in providers] == [
|
|
"langgenius/openai/openai",
|
|
"langgenius/embedding/embedding",
|
|
]
|
|
assert providers[0].is_configured is False
|
|
assert providers[0].custom_configuration.status.value == "no-configure"
|
|
assert providers[0].custom_configuration.has_custom_models is False
|
|
assert providers[0].custom_configuration.available_credentials == []
|
|
assert set(plugins) == {"langgenius/openai", "langgenius/embedding"}
|
|
|
|
def test_preferred_provider_fallback_uses_custom_presence_not_configuration_status(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(service_module.dify_config, "DEPLOYMENT_EDITION", DeploymentEdition.COMMUNITY)
|
|
state = _ProviderSummaryState(has_custom_provider=True)
|
|
|
|
preferred_provider_type = ModelProviderService._get_preferred_provider_type(
|
|
state,
|
|
custom_present=True,
|
|
system_enabled=True,
|
|
)
|
|
|
|
assert preferred_provider_type == ProviderType.CUSTOM
|
|
|
|
def test_load_provider_summary_states_reads_only_lightweight_columns(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
canonical_provider = "langgenius/openai/openai"
|
|
session = MagicMock()
|
|
session.execute.side_effect = [
|
|
SimpleNamespace(
|
|
all=lambda: [
|
|
SimpleNamespace(
|
|
provider_name="openai",
|
|
credential_id="credential-legacy",
|
|
credential_provider_name="openai",
|
|
credential_name="Legacy",
|
|
),
|
|
SimpleNamespace(
|
|
provider_name=canonical_provider,
|
|
credential_id="credential-current",
|
|
credential_provider_name=canonical_provider,
|
|
credential_name="Production",
|
|
),
|
|
]
|
|
),
|
|
SimpleNamespace(
|
|
all=lambda: [
|
|
SimpleNamespace(
|
|
id="credential-legacy",
|
|
provider_name="openai",
|
|
credential_name="Legacy",
|
|
),
|
|
SimpleNamespace(
|
|
id="credential-current",
|
|
provider_name=canonical_provider,
|
|
credential_name="Production",
|
|
),
|
|
]
|
|
),
|
|
SimpleNamespace(all=lambda: [SimpleNamespace(provider_name="openai")]),
|
|
SimpleNamespace(
|
|
all=lambda: [
|
|
SimpleNamespace(
|
|
provider_name=canonical_provider,
|
|
preferred_provider_type=ProviderType.SYSTEM,
|
|
)
|
|
]
|
|
),
|
|
]
|
|
session_context = MagicMock()
|
|
session_context.__enter__.return_value = session
|
|
create_session = MagicMock(return_value=session_context)
|
|
monkeypatch.setattr(service_module.session_factory, "create_session", create_session)
|
|
|
|
states = ModelProviderService._load_provider_summary_states("tenant-1")
|
|
|
|
state = states[canonical_provider]
|
|
assert state.has_custom_provider is True
|
|
assert state.available_credentials == [
|
|
CredentialConfiguration(
|
|
credential_id="credential-legacy",
|
|
credential_name="Legacy",
|
|
),
|
|
CredentialConfiguration(
|
|
credential_id="credential-current",
|
|
credential_name="Production",
|
|
),
|
|
]
|
|
assert state.has_custom_models is True
|
|
assert state.current_credential_id == "credential-current"
|
|
assert state.current_credential_name == "Production"
|
|
assert state.current_credential_usable is True
|
|
assert state.preferred_provider_type == ProviderType.SYSTEM
|
|
|
|
statements = [str(execute_call.args[0]) for execute_call in session.execute.call_args_list]
|
|
assert len(statements) == 4
|
|
assert all("encrypted_config" not in statement for statement in statements)
|
|
assert "count(" not in statements[1].lower()
|
|
assert "provider_credentials.id" in statements[1]
|
|
assert "provider_credentials.credential_name" in statements[1]
|
|
assert "ORDER BY provider_credentials.created_at DESC, provider_credentials.id DESC" in statements[1]
|
|
assert "provider_model_credentials" in statements[2]
|
|
|
|
def test_get_models_by_provider_should_wrap_model_entities_with_tenant_context(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
|
|
class _Model:
|
|
def __init__(self, model_name: str) -> None:
|
|
self.model_name = model_name
|
|
|
|
def model_dump(self) -> dict[str, Any]:
|
|
return {
|
|
"model": self.model_name,
|
|
"label": {"en_US": self.model_name},
|
|
"model_type": ModelType.LLM,
|
|
"features": [],
|
|
"fetch_from": FetchFrom.PREDEFINED_MODEL,
|
|
"model_properties": {},
|
|
"deprecated": False,
|
|
"status": ModelStatus.ACTIVE,
|
|
"load_balancing_enabled": False,
|
|
"has_invalid_load_balancing_configs": False,
|
|
"provider": {
|
|
"provider": "openai",
|
|
"label": {"en_US": "OpenAI"},
|
|
"icon_small": None,
|
|
"icon_small_dark": None,
|
|
"supported_model_types": [ModelType.LLM],
|
|
},
|
|
}
|
|
|
|
provider_configurations = SimpleNamespace(
|
|
get_models=MagicMock(return_value=[_Model("gpt-4o"), _Model("gpt-4o-mini")])
|
|
)
|
|
manager.get_configurations.return_value = provider_configurations
|
|
|
|
result = service.get_models_by_provider(tenant_id="tenant-1", provider="openai")
|
|
|
|
assert len(result) == 2
|
|
assert result[0].model == "gpt-4o"
|
|
assert result[1].provider.provider == "openai"
|
|
provider_configurations.get_models.assert_called_once_with(provider="openai")
|
|
|
|
|
|
class TestModelProviderServiceDelegation:
|
|
@pytest.mark.parametrize(
|
|
("method_name", "method_kwargs", "provider_method_name", "provider_call_kwargs", "provider_return"),
|
|
[
|
|
(
|
|
"get_provider_credential",
|
|
{"tenant_id": "tenant-1", "provider": "openai", "credential_id": "cred-1"},
|
|
"get_provider_credential",
|
|
{"credential_id": "cred-1"},
|
|
{"token": "abc"},
|
|
),
|
|
(
|
|
"validate_provider_credentials",
|
|
{"tenant_id": "tenant-1", "provider": "openai", "credentials": {"token": "abc"}},
|
|
"validate_provider_credentials",
|
|
({"token": "abc"},),
|
|
None,
|
|
),
|
|
(
|
|
"create_provider_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"credentials": {"token": "abc"},
|
|
"credential_name": "A",
|
|
},
|
|
"create_provider_credential",
|
|
({"token": "abc"}, "A"),
|
|
None,
|
|
),
|
|
(
|
|
"update_provider_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"credentials": {"token": "abc"},
|
|
"credential_id": "cred-1",
|
|
"credential_name": "B",
|
|
},
|
|
"update_provider_credential",
|
|
{"credential_id": "cred-1", "credentials": {"token": "abc"}, "credential_name": "B"},
|
|
None,
|
|
),
|
|
(
|
|
"remove_provider_credential",
|
|
{"tenant_id": "tenant-1", "provider": "openai", "credential_id": "cred-1"},
|
|
"delete_provider_credential",
|
|
{"credential_id": "cred-1"},
|
|
None,
|
|
),
|
|
(
|
|
"switch_active_provider_credential",
|
|
{"tenant_id": "tenant-1", "provider": "openai", "credential_id": "cred-1"},
|
|
"switch_active_provider_credential",
|
|
{"credential_id": "cred-1"},
|
|
None,
|
|
),
|
|
],
|
|
)
|
|
def test_provider_credential_methods_should_delegate_to_provider_configuration(
|
|
self,
|
|
method_name: str,
|
|
method_kwargs: dict[str, Any],
|
|
provider_method_name: str,
|
|
provider_call_kwargs: Any,
|
|
provider_return: Any,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider_configuration = MagicMock()
|
|
getattr(provider_configuration, provider_method_name).return_value = provider_return
|
|
get_provider_config_mock = MagicMock(return_value=provider_configuration)
|
|
monkeypatch.setattr(service, "_get_provider_configuration", get_provider_config_mock)
|
|
|
|
result = getattr(service, method_name)(**method_kwargs)
|
|
|
|
get_provider_config_mock.assert_called_once_with("tenant-1", "openai")
|
|
provider_method = getattr(provider_configuration, provider_method_name)
|
|
match provider_call_kwargs:
|
|
case tuple():
|
|
provider_method.assert_called_once_with(*provider_call_kwargs)
|
|
case dict():
|
|
provider_method.assert_called_once_with(**provider_call_kwargs)
|
|
case _:
|
|
provider_method.assert_called_once_with(provider_call_kwargs)
|
|
if method_name == "get_provider_credential":
|
|
assert result == {"token": "abc"}
|
|
|
|
@pytest.mark.parametrize(
|
|
("method_name", "method_kwargs", "provider_method_name", "expected_kwargs", "provider_return"),
|
|
[
|
|
(
|
|
"get_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credential_id": "cred-1",
|
|
},
|
|
"get_custom_model_credential",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o", "credential_id": "cred-1"},
|
|
{"api_key": "x"},
|
|
),
|
|
(
|
|
"validate_model_credentials",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
},
|
|
"validate_custom_model_credentials",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o", "credentials": {"api_key": "x"}},
|
|
None,
|
|
),
|
|
(
|
|
"create_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
"credential_name": "cred-a",
|
|
},
|
|
"create_custom_model_credential",
|
|
{
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
"credential_name": "cred-a",
|
|
},
|
|
None,
|
|
),
|
|
(
|
|
"update_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
"credential_id": "cred-1",
|
|
"credential_name": "cred-b",
|
|
},
|
|
"update_custom_model_credential",
|
|
{
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
"credential_id": "cred-1",
|
|
"credential_name": "cred-b",
|
|
},
|
|
None,
|
|
),
|
|
(
|
|
"remove_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credential_id": "cred-1",
|
|
},
|
|
"delete_custom_model_credential",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o", "credential_id": "cred-1"},
|
|
None,
|
|
),
|
|
(
|
|
"switch_active_custom_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credential_id": "cred-1",
|
|
},
|
|
"switch_custom_model_credential",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o", "credential_id": "cred-1"},
|
|
None,
|
|
),
|
|
(
|
|
"add_model_credential_to_model_list",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credential_id": "cred-1",
|
|
},
|
|
"add_model_credential_to_model",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o", "credential_id": "cred-1"},
|
|
None,
|
|
),
|
|
(
|
|
"remove_model",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
},
|
|
"delete_custom_model",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o"},
|
|
None,
|
|
),
|
|
],
|
|
)
|
|
def test_custom_model_methods_should_convert_model_type_and_delegate(
|
|
self,
|
|
method_name: str,
|
|
method_kwargs: dict[str, Any],
|
|
provider_method_name: str,
|
|
expected_kwargs: dict[str, Any],
|
|
provider_return: Any,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider_configuration = MagicMock()
|
|
getattr(provider_configuration, provider_method_name).return_value = provider_return
|
|
get_provider_config_mock = MagicMock(return_value=provider_configuration)
|
|
monkeypatch.setattr(service, "_get_provider_configuration", get_provider_config_mock)
|
|
|
|
result = getattr(service, method_name)(**method_kwargs)
|
|
|
|
get_provider_config_mock.assert_called_once_with("tenant-1", "openai")
|
|
getattr(provider_configuration, provider_method_name).assert_called_once_with(**expected_kwargs)
|
|
if method_name == "get_model_credential":
|
|
assert result == {"api_key": "x"}
|
|
|
|
@pytest.mark.parametrize(
|
|
("method_name", "method_kwargs", "provider_method_name", "expected_kwargs"),
|
|
[
|
|
(
|
|
"get_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": "llm",
|
|
"model": "gpt-4o",
|
|
"credential_id": "cred-1",
|
|
},
|
|
"get_custom_model_credential",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o", "credential_id": "cred-1"},
|
|
),
|
|
(
|
|
"create_model_credential",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": "llm",
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
"credential_name": "cred-a",
|
|
},
|
|
"create_custom_model_credential",
|
|
{
|
|
"model_type": ModelType.LLM,
|
|
"model": "gpt-4o",
|
|
"credentials": {"api_key": "x"},
|
|
"credential_name": "cred-a",
|
|
},
|
|
),
|
|
(
|
|
"remove_model",
|
|
{
|
|
"tenant_id": "tenant-1",
|
|
"provider": "openai",
|
|
"model_type": "llm",
|
|
"model": "gpt-4o",
|
|
},
|
|
"delete_custom_model",
|
|
{"model_type": ModelType.LLM, "model": "gpt-4o"},
|
|
),
|
|
],
|
|
)
|
|
def test_custom_model_methods_use_model_type_constructor_directly(
|
|
self,
|
|
method_name: str,
|
|
method_kwargs: dict[str, Any],
|
|
provider_method_name: str,
|
|
expected_kwargs: dict[str, Any],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider_configuration = MagicMock()
|
|
get_provider_config_mock = MagicMock(return_value=provider_configuration)
|
|
monkeypatch.setattr(service, "_get_provider_configuration", get_provider_config_mock)
|
|
|
|
getattr(service, method_name)(**method_kwargs)
|
|
|
|
get_provider_config_mock.assert_called_once_with("tenant-1", "openai")
|
|
getattr(provider_configuration, provider_method_name).assert_called_once_with(**expected_kwargs)
|
|
|
|
|
|
class TestModelProviderServiceListingsAndDefaults:
|
|
def test_get_models_by_model_type_should_group_active_non_deprecated_models(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
openai_provider = SimpleNamespace(
|
|
provider="openai",
|
|
label=I18nObject(en_US="OpenAI"),
|
|
icon_small=None,
|
|
icon_small_dark=None,
|
|
)
|
|
anthropic_provider = SimpleNamespace(
|
|
provider="anthropic",
|
|
label=I18nObject(en_US="Anthropic"),
|
|
icon_small=None,
|
|
icon_small_dark=None,
|
|
)
|
|
models = [
|
|
SimpleNamespace(
|
|
provider=openai_provider,
|
|
model="gpt-4o",
|
|
label=I18nObject(en_US="GPT-4o"),
|
|
model_type=ModelType.LLM,
|
|
features=[],
|
|
fetch_from=FetchFrom.PREDEFINED_MODEL,
|
|
model_properties={},
|
|
status=ModelStatus.ACTIVE,
|
|
load_balancing_enabled=False,
|
|
deprecated=False,
|
|
),
|
|
SimpleNamespace(
|
|
provider=openai_provider,
|
|
model="old-openai",
|
|
label=I18nObject(en_US="Old OpenAI"),
|
|
model_type=ModelType.LLM,
|
|
features=[],
|
|
fetch_from=FetchFrom.PREDEFINED_MODEL,
|
|
model_properties={},
|
|
status=ModelStatus.ACTIVE,
|
|
load_balancing_enabled=False,
|
|
deprecated=True,
|
|
),
|
|
SimpleNamespace(
|
|
provider=anthropic_provider,
|
|
model="old-anthropic",
|
|
label=I18nObject(en_US="Old Anthropic"),
|
|
model_type=ModelType.LLM,
|
|
features=[],
|
|
fetch_from=FetchFrom.PREDEFINED_MODEL,
|
|
model_properties={},
|
|
status=ModelStatus.ACTIVE,
|
|
load_balancing_enabled=False,
|
|
deprecated=True,
|
|
),
|
|
]
|
|
provider_configurations = SimpleNamespace(get_models=MagicMock(return_value=models))
|
|
manager.get_configurations.return_value = provider_configurations
|
|
|
|
result = service.get_models_by_model_type(tenant_id="tenant-1", model_type=ModelType.LLM)
|
|
|
|
provider_configurations.get_models.assert_called_once_with(model_type=ModelType.LLM, only_active=True)
|
|
assert len(result) == 1
|
|
assert result[0].provider == "openai"
|
|
assert len(result[0].models) == 1
|
|
assert result[0].models[0].model == "gpt-4o"
|
|
|
|
@pytest.mark.parametrize(
|
|
("credentials", "schema", "expected_count"),
|
|
[
|
|
(None, None, 0),
|
|
({"api_key": "x"}, None, 0),
|
|
(
|
|
{"api_key": "x"},
|
|
SimpleNamespace(
|
|
parameter_rules=[
|
|
ParameterRule(
|
|
name="temperature",
|
|
label=I18nObject(en_US="Temperature"),
|
|
type=ParameterType.FLOAT,
|
|
)
|
|
]
|
|
),
|
|
1,
|
|
),
|
|
],
|
|
)
|
|
def test_get_model_parameter_rules_should_handle_missing_credentials_and_schema(
|
|
self,
|
|
credentials: dict[str, Any] | None,
|
|
schema: Any,
|
|
expected_count: int,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider_configuration = MagicMock()
|
|
provider_configuration.get_current_credentials.return_value = credentials
|
|
provider_configuration.get_model_schema.return_value = schema
|
|
monkeypatch.setattr(service, "_get_provider_configuration", MagicMock(return_value=provider_configuration))
|
|
|
|
result = service.get_model_parameter_rules(tenant_id="tenant-1", provider="openai", model="gpt-4o")
|
|
|
|
assert len(result) == expected_count
|
|
provider_configuration.get_current_credentials.assert_called_once_with(
|
|
model_type=ModelType.LLM,
|
|
model="gpt-4o",
|
|
)
|
|
if credentials:
|
|
provider_configuration.get_model_schema.assert_called_once_with(
|
|
model_type=ModelType.LLM,
|
|
model="gpt-4o",
|
|
credentials=credentials,
|
|
)
|
|
else:
|
|
provider_configuration.get_model_schema.assert_not_called()
|
|
|
|
def test_get_default_model_of_model_type_should_return_response_when_manager_returns_model(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
manager.get_default_model.return_value = SimpleNamespace(
|
|
model="gpt-4o",
|
|
model_type=ModelType.LLM,
|
|
provider=SimpleNamespace(
|
|
provider="openai",
|
|
label=I18nObject(en_US="OpenAI"),
|
|
icon_small=None,
|
|
supported_model_types=[ModelType.LLM],
|
|
),
|
|
)
|
|
|
|
result = service.get_default_model_of_model_type(tenant_id="tenant-1", model_type=ModelType.LLM)
|
|
|
|
assert result is not None
|
|
assert result.model == "gpt-4o"
|
|
assert result.provider.provider == "openai"
|
|
manager.get_default_model.assert_called_once_with(tenant_id="tenant-1", model_type=ModelType.LLM)
|
|
|
|
def test_get_default_model_of_model_type_should_return_none_when_manager_returns_none(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
manager.get_default_model.return_value = None
|
|
|
|
result = service.get_default_model_of_model_type(tenant_id="tenant-1", model_type=ModelType.LLM)
|
|
|
|
assert result is None
|
|
|
|
def test_get_default_model_of_model_type_should_return_none_when_manager_raises_exception(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
manager.get_default_model.side_effect = RuntimeError("boom")
|
|
|
|
result = service.get_default_model_of_model_type(tenant_id="tenant-1", model_type=ModelType.LLM)
|
|
|
|
assert result is None
|
|
|
|
def test_update_default_model_of_model_type_should_delegate_to_provider_manager(self) -> None:
|
|
service, manager = _create_service_with_mocked_manager()
|
|
|
|
service.update_default_model_of_model_type(
|
|
tenant_id="tenant-1",
|
|
model_type=ModelType.LLM,
|
|
provider="openai",
|
|
model="gpt-4o",
|
|
)
|
|
|
|
manager.update_default_model_record.assert_called_once_with(
|
|
tenant_id="tenant-1",
|
|
model_type=ModelType.LLM,
|
|
provider="openai",
|
|
model="gpt-4o",
|
|
)
|
|
|
|
def test_get_model_provider_icon_should_fetch_icon_bytes_from_factory(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
factory_instance = MagicMock()
|
|
factory_instance.get_provider_icon.return_value = (b"icon-bytes", "image/png")
|
|
factory_constructor = MagicMock(return_value=factory_instance)
|
|
monkeypatch.setattr(service_module, "create_plugin_model_provider_factory", factory_constructor)
|
|
|
|
result = service.get_model_provider_icon(
|
|
tenant_id="tenant-1",
|
|
provider="openai",
|
|
icon_type="icon_small",
|
|
lang="en_US",
|
|
)
|
|
|
|
factory_constructor.assert_called_once_with(tenant_id="tenant-1")
|
|
factory_instance.get_provider_icon.assert_called_once_with("openai", "icon_small", "en_US")
|
|
assert result == (b"icon-bytes", "image/png")
|
|
|
|
def test_switch_preferred_provider_should_convert_enum_and_delegate(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider_configuration = MagicMock()
|
|
monkeypatch.setattr(service, "_get_provider_configuration", MagicMock(return_value=provider_configuration))
|
|
|
|
service.switch_preferred_provider(
|
|
tenant_id="tenant-1",
|
|
provider="openai",
|
|
preferred_provider_type=ProviderType.SYSTEM.value,
|
|
)
|
|
|
|
provider_configuration.switch_preferred_provider_type.assert_called_once_with(ProviderType.SYSTEM)
|
|
|
|
@pytest.mark.parametrize(
|
|
("method_name", "provider_method_name"),
|
|
[
|
|
("enable_model", "enable_model"),
|
|
("disable_model", "disable_model"),
|
|
],
|
|
)
|
|
def test_model_enablement_methods_should_convert_model_type_and_delegate(
|
|
self,
|
|
method_name: str,
|
|
provider_method_name: str,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = ModelProviderService()
|
|
provider_configuration = MagicMock()
|
|
monkeypatch.setattr(service, "_get_provider_configuration", MagicMock(return_value=provider_configuration))
|
|
|
|
getattr(service, method_name)(
|
|
tenant_id="tenant-1",
|
|
provider="openai",
|
|
model="gpt-4o",
|
|
model_type=ModelType.LLM,
|
|
)
|
|
|
|
getattr(provider_configuration, provider_method_name).assert_called_once_with(
|
|
model="gpt-4o",
|
|
model_type=ModelType.LLM,
|
|
)
|