mirror of
https://github.com/langgenius/dify.git
synced 2026-05-13 08:57:28 +08:00
test: migrate recommended_app_service tests to testcontainers (#34751)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
volcano303
GitHub
parent
9c4f897b9a
commit
1898a3f8a5
@ -0,0 +1,388 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from models.model import AccountTrialAppRecord, TrialApp
|
||||
from services import recommended_app_service as service_module
|
||||
from services.recommended_app_service import RecommendedAppService
|
||||
|
||||
# ── Helpers ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _apps_response(
|
||||
recommended_apps: list[dict] | None = None,
|
||||
categories: list[str] | None = None,
|
||||
) -> dict:
|
||||
if recommended_apps is None:
|
||||
recommended_apps = [
|
||||
{"id": "app-1", "name": "Test App 1", "description": "d1", "category": "productivity"},
|
||||
{"id": "app-2", "name": "Test App 2", "description": "d2", "category": "communication"},
|
||||
]
|
||||
if categories is None:
|
||||
categories = ["productivity", "communication", "utilities"]
|
||||
return {"recommended_apps": recommended_apps, "categories": categories}
|
||||
|
||||
|
||||
def _app_detail(
|
||||
app_id: str = "app-123",
|
||||
name: str = "Test App",
|
||||
description: str = "Test description",
|
||||
**kwargs: Any,
|
||||
) -> dict:
|
||||
detail: dict[str, Any] = {
|
||||
"id": app_id,
|
||||
"name": name,
|
||||
"description": description,
|
||||
"category": kwargs.get("category", "productivity"),
|
||||
"icon": kwargs.get("icon", "🚀"),
|
||||
"model_config": kwargs.get("model_config", {}),
|
||||
}
|
||||
detail.update(kwargs)
|
||||
return detail
|
||||
|
||||
|
||||
def _recommendation_detail(result: dict[str, Any] | None) -> dict[str, Any] | None:
|
||||
return cast("dict[str, Any] | None", result)
|
||||
|
||||
|
||||
def _mock_factory_for_apps(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
*,
|
||||
mode: str,
|
||||
result: dict[str, Any],
|
||||
fallback_result: dict[str, Any] | None = None,
|
||||
) -> tuple[MagicMock, MagicMock]:
|
||||
retrieval_instance = MagicMock()
|
||||
retrieval_instance.get_recommended_apps_and_categories.return_value = result
|
||||
retrieval_factory = MagicMock(return_value=retrieval_instance)
|
||||
monkeypatch.setattr(service_module.dify_config, "HOSTED_FETCH_APP_TEMPLATES_MODE", mode, raising=False)
|
||||
monkeypatch.setattr(
|
||||
service_module.RecommendAppRetrievalFactory,
|
||||
"get_recommend_app_factory",
|
||||
MagicMock(return_value=retrieval_factory),
|
||||
)
|
||||
builtin_instance = MagicMock()
|
||||
if fallback_result is not None:
|
||||
builtin_instance.fetch_recommended_apps_from_builtin.return_value = fallback_result
|
||||
monkeypatch.setattr(
|
||||
service_module.RecommendAppRetrievalFactory,
|
||||
"get_buildin_recommend_app_retrieval",
|
||||
MagicMock(return_value=builtin_instance),
|
||||
)
|
||||
return retrieval_instance, builtin_instance
|
||||
|
||||
|
||||
# ── Pure logic tests: get_recommended_apps_and_categories ──────────────
|
||||
|
||||
|
||||
class TestRecommendedAppServiceGetApps:
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_success_with_apps(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
expected = _apps_response()
|
||||
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommended_apps_and_categories.return_value = expected
|
||||
mock_factory = MagicMock(return_value=mock_instance)
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
assert result == expected
|
||||
assert len(result["recommended_apps"]) == 2
|
||||
assert len(result["categories"]) == 3
|
||||
mock_factory_class.get_recommend_app_factory.assert_called_once_with("remote")
|
||||
mock_instance.get_recommended_apps_and_categories.assert_called_once_with("en-US")
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_fallback_to_builtin_when_empty(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
empty_response = {"recommended_apps": [], "categories": []}
|
||||
builtin_response = _apps_response(
|
||||
recommended_apps=[{"id": "builtin-1", "name": "Builtin App", "category": "default"}]
|
||||
)
|
||||
|
||||
mock_remote_instance = MagicMock()
|
||||
mock_remote_instance.get_recommended_apps_and_categories.return_value = empty_response
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_remote_instance)
|
||||
|
||||
mock_builtin_instance = MagicMock()
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.return_value = builtin_response
|
||||
mock_factory_class.get_buildin_recommend_app_retrieval.return_value = mock_builtin_instance
|
||||
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("zh-CN")
|
||||
|
||||
assert result == builtin_response
|
||||
assert result["recommended_apps"][0]["id"] == "builtin-1"
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.assert_called_once_with("en-US")
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_fallback_when_none_recommended_apps(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "db"
|
||||
none_response = {"recommended_apps": None, "categories": ["test"]}
|
||||
builtin_response = _apps_response()
|
||||
|
||||
mock_db_instance = MagicMock()
|
||||
mock_db_instance.get_recommended_apps_and_categories.return_value = none_response
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_db_instance)
|
||||
|
||||
mock_builtin_instance = MagicMock()
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.return_value = builtin_response
|
||||
mock_factory_class.get_buildin_recommend_app_retrieval.return_value = mock_builtin_instance
|
||||
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
assert result == builtin_response
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.assert_called_once()
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_different_languages(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "builtin"
|
||||
|
||||
for language in ["en-US", "zh-CN", "ja-JP", "fr-FR"]:
|
||||
lang_response = _apps_response(
|
||||
recommended_apps=[{"id": f"app-{language}", "name": f"App {language}", "category": "test"}]
|
||||
)
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommended_apps_and_categories.return_value = lang_response
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories(language)
|
||||
|
||||
assert result["recommended_apps"][0]["id"] == f"app-{language}"
|
||||
mock_instance.get_recommended_apps_and_categories.assert_called_with(language)
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_uses_correct_factory_mode(self, mock_config, mock_factory_class):
|
||||
for mode in ["remote", "builtin", "db"]:
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = mode
|
||||
response = _apps_response()
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommended_apps_and_categories.return_value = response
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
mock_factory_class.get_recommend_app_factory.assert_called_with(mode)
|
||||
|
||||
|
||||
# ── Pure logic tests: get_recommend_app_detail ─────────────────────────
|
||||
|
||||
|
||||
class TestRecommendedAppServiceGetDetail:
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_success(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
expected = _app_detail(app_id="app-123", name="Productivity App", description="A great app")
|
||||
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = expected
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail("app-123"))
|
||||
|
||||
assert result == expected
|
||||
assert result["id"] == "app-123"
|
||||
mock_instance.get_recommend_app_detail.assert_called_once_with("app-123")
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_different_modes(self, mock_config, mock_factory_class):
|
||||
for mode in ["remote", "builtin", "db"]:
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = mode
|
||||
detail = _app_detail(app_id="test-app", name=f"App from {mode}")
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = detail
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail("test-app"))
|
||||
|
||||
assert result["name"] == f"App from {mode}"
|
||||
mock_factory_class.get_recommend_app_factory.assert_called_with(mode)
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_returns_none_when_not_found(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = None
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail("nonexistent"))
|
||||
|
||||
assert result is None
|
||||
mock_instance.get_recommend_app_detail.assert_called_once_with("nonexistent")
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_returns_empty_dict(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "builtin"
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = {}
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail("app-empty"))
|
||||
|
||||
assert result == {}
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_complex_model_config(self, mock_config, mock_factory_class):
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
complex_config = {
|
||||
"provider": "openai",
|
||||
"model": "gpt-4",
|
||||
"parameters": {"temperature": 0.7, "max_tokens": 2000, "top_p": 1.0},
|
||||
}
|
||||
expected = _app_detail(
|
||||
app_id="complex-app",
|
||||
name="Complex App",
|
||||
model_config=complex_config,
|
||||
workflows=["workflow-1", "workflow-2"],
|
||||
tools=["tool-1", "tool-2", "tool-3"],
|
||||
)
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = expected
|
||||
mock_factory_class.get_recommend_app_factory.return_value = MagicMock(return_value=mock_instance)
|
||||
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail("complex-app"))
|
||||
|
||||
assert result["model_config"] == complex_config
|
||||
assert len(result["workflows"]) == 2
|
||||
assert len(result["tools"]) == 3
|
||||
|
||||
|
||||
# ── Integration tests: trial app features (real DB) ────────────────────
|
||||
|
||||
|
||||
class TestRecommendedAppServiceTrialFeatures:
|
||||
def test_get_apps_should_not_query_trial_table_when_disabled(
|
||||
self, db_session_with_containers: Session, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
expected = {"recommended_apps": [{"app_id": "app-1"}], "categories": ["all"]}
|
||||
retrieval_instance, builtin_instance = _mock_factory_for_apps(monkeypatch, mode="remote", result=expected)
|
||||
monkeypatch.setattr(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
MagicMock(return_value=SimpleNamespace(enable_trial_app=False)),
|
||||
)
|
||||
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
assert result == expected
|
||||
retrieval_instance.get_recommended_apps_and_categories.assert_called_once_with("en-US")
|
||||
builtin_instance.fetch_recommended_apps_from_builtin.assert_not_called()
|
||||
|
||||
def test_get_apps_should_enrich_can_trial_when_enabled(
|
||||
self, db_session_with_containers: Session, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
app_id_1 = str(uuid.uuid4())
|
||||
app_id_2 = str(uuid.uuid4())
|
||||
tenant_id = str(uuid.uuid4())
|
||||
|
||||
# app_id_1 has a TrialApp record; app_id_2 does not
|
||||
db_session_with_containers.add(TrialApp(app_id=app_id_1, tenant_id=tenant_id))
|
||||
db_session_with_containers.commit()
|
||||
|
||||
remote_result = {"recommended_apps": [], "categories": []}
|
||||
fallback_result = {
|
||||
"recommended_apps": [{"app_id": app_id_1}, {"app_id": app_id_2}],
|
||||
"categories": ["all"],
|
||||
}
|
||||
_, builtin_instance = _mock_factory_for_apps(
|
||||
monkeypatch, mode="remote", result=remote_result, fallback_result=fallback_result
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
MagicMock(return_value=SimpleNamespace(enable_trial_app=True)),
|
||||
)
|
||||
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("ja-JP")
|
||||
|
||||
builtin_instance.fetch_recommended_apps_from_builtin.assert_called_once_with("en-US")
|
||||
assert result["recommended_apps"][0]["can_trial"] is True
|
||||
assert result["recommended_apps"][1]["can_trial"] is False
|
||||
|
||||
@pytest.mark.parametrize("has_trial_app", [True, False])
|
||||
def test_get_detail_should_set_can_trial_when_enabled(
|
||||
self,
|
||||
db_session_with_containers: Session,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
has_trial_app: bool,
|
||||
):
|
||||
app_id = str(uuid.uuid4())
|
||||
tenant_id = str(uuid.uuid4())
|
||||
|
||||
if has_trial_app:
|
||||
db_session_with_containers.add(TrialApp(app_id=app_id, tenant_id=tenant_id))
|
||||
db_session_with_containers.commit()
|
||||
|
||||
detail = {"id": app_id, "name": "Test App"}
|
||||
retrieval_instance = MagicMock()
|
||||
retrieval_instance.get_recommend_app_detail.return_value = detail
|
||||
retrieval_factory = MagicMock(return_value=retrieval_instance)
|
||||
monkeypatch.setattr(service_module.dify_config, "HOSTED_FETCH_APP_TEMPLATES_MODE", "remote", raising=False)
|
||||
monkeypatch.setattr(
|
||||
service_module.RecommendAppRetrievalFactory,
|
||||
"get_recommend_app_factory",
|
||||
MagicMock(return_value=retrieval_factory),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
MagicMock(return_value=SimpleNamespace(enable_trial_app=True)),
|
||||
)
|
||||
|
||||
result = cast(dict[str, Any], RecommendedAppService.get_recommend_app_detail(app_id))
|
||||
|
||||
assert result["id"] == app_id
|
||||
assert result["can_trial"] is has_trial_app
|
||||
|
||||
def test_add_trial_app_record_increments_count_for_existing(self, db_session_with_containers: Session):
|
||||
app_id = str(uuid.uuid4())
|
||||
account_id = str(uuid.uuid4())
|
||||
|
||||
db_session_with_containers.add(AccountTrialAppRecord(app_id=app_id, account_id=account_id, count=3))
|
||||
db_session_with_containers.commit()
|
||||
|
||||
RecommendedAppService.add_trial_app_record(app_id, account_id)
|
||||
|
||||
db_session_with_containers.expire_all()
|
||||
record = db_session_with_containers.scalar(
|
||||
select(AccountTrialAppRecord)
|
||||
.where(AccountTrialAppRecord.app_id == app_id, AccountTrialAppRecord.account_id == account_id)
|
||||
.limit(1)
|
||||
)
|
||||
assert record is not None
|
||||
assert record.count == 4
|
||||
|
||||
def test_add_trial_app_record_creates_new_record(self, db_session_with_containers: Session):
|
||||
app_id = str(uuid.uuid4())
|
||||
account_id = str(uuid.uuid4())
|
||||
|
||||
RecommendedAppService.add_trial_app_record(app_id, account_id)
|
||||
|
||||
db_session_with_containers.expire_all()
|
||||
record = db_session_with_containers.scalar(
|
||||
select(AccountTrialAppRecord)
|
||||
.where(AccountTrialAppRecord.app_id == app_id, AccountTrialAppRecord.account_id == account_id)
|
||||
.limit(1)
|
||||
)
|
||||
assert record is not None
|
||||
assert record.app_id == app_id
|
||||
assert record.account_id == account_id
|
||||
assert record.count == 1
|
||||
@ -1,628 +0,0 @@
|
||||
"""
|
||||
Comprehensive unit tests for RecommendedAppService.
|
||||
|
||||
This test suite provides complete coverage of recommended app operations in Dify,
|
||||
following TDD principles with the Arrange-Act-Assert pattern.
|
||||
|
||||
## Test Coverage
|
||||
|
||||
### 1. Get Recommended Apps and Categories (TestRecommendedAppServiceGetApps)
|
||||
Tests fetching recommended apps with categories:
|
||||
- Successful retrieval with recommended apps
|
||||
- Fallback to builtin when no recommended apps
|
||||
- Different language support
|
||||
- Factory mode selection (remote, builtin, db)
|
||||
- Empty result handling
|
||||
|
||||
### 2. Get Recommend App Detail (TestRecommendedAppServiceGetDetail)
|
||||
Tests fetching individual app details:
|
||||
- Successful app detail retrieval
|
||||
- Different factory modes
|
||||
- App not found scenarios
|
||||
- Language-specific details
|
||||
|
||||
## Testing Approach
|
||||
|
||||
- **Mocking Strategy**: All external dependencies (dify_config, RecommendAppRetrievalFactory)
|
||||
are mocked for fast, isolated unit tests
|
||||
- **Factory Pattern**: Tests verify correct factory selection based on mode
|
||||
- **Fixtures**: Mock objects are configured per test method
|
||||
- **Assertions**: Each test verifies return values and factory method calls
|
||||
|
||||
## Key Concepts
|
||||
|
||||
**Factory Modes:**
|
||||
- remote: Fetch from remote API
|
||||
- builtin: Use built-in templates
|
||||
- db: Fetch from database
|
||||
|
||||
**Fallback Logic:**
|
||||
- If remote/db returns no apps, fallback to builtin en-US templates
|
||||
- Ensures users always see some recommended apps
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from services.recommended_app_service import RecommendedAppService
|
||||
|
||||
|
||||
class RecommendedAppServiceTestDataFactory:
|
||||
"""
|
||||
Factory for creating test data and mock objects.
|
||||
|
||||
Provides reusable methods to create consistent mock objects for testing
|
||||
recommended app operations.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def create_recommended_apps_response(
|
||||
recommended_apps: list[dict] | None = None,
|
||||
categories: list[str] | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Create a mock response for recommended apps.
|
||||
|
||||
Args:
|
||||
recommended_apps: List of recommended app dictionaries
|
||||
categories: List of category names
|
||||
|
||||
Returns:
|
||||
Dictionary with recommended_apps and categories
|
||||
"""
|
||||
if recommended_apps is None:
|
||||
recommended_apps = [
|
||||
{
|
||||
"id": "app-1",
|
||||
"name": "Test App 1",
|
||||
"description": "Test description 1",
|
||||
"category": "productivity",
|
||||
},
|
||||
{
|
||||
"id": "app-2",
|
||||
"name": "Test App 2",
|
||||
"description": "Test description 2",
|
||||
"category": "communication",
|
||||
},
|
||||
]
|
||||
if categories is None:
|
||||
categories = ["productivity", "communication", "utilities"]
|
||||
|
||||
return {
|
||||
"recommended_apps": recommended_apps,
|
||||
"categories": categories,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def create_app_detail_response(
|
||||
app_id: str = "app-123",
|
||||
name: str = "Test App",
|
||||
description: str = "Test description",
|
||||
**kwargs,
|
||||
) -> dict:
|
||||
"""
|
||||
Create a mock response for app detail.
|
||||
|
||||
Args:
|
||||
app_id: App identifier
|
||||
name: App name
|
||||
description: App description
|
||||
**kwargs: Additional fields
|
||||
|
||||
Returns:
|
||||
Dictionary with app details
|
||||
"""
|
||||
detail = {
|
||||
"id": app_id,
|
||||
"name": name,
|
||||
"description": description,
|
||||
"category": kwargs.get("category", "productivity"),
|
||||
"icon": kwargs.get("icon", "🚀"),
|
||||
"model_config": kwargs.get("model_config", {}),
|
||||
}
|
||||
detail.update(kwargs)
|
||||
return detail
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def factory():
|
||||
"""Provide the test data factory to all tests."""
|
||||
return RecommendedAppServiceTestDataFactory
|
||||
|
||||
|
||||
class TestRecommendedAppServiceGetApps:
|
||||
"""Test get_recommended_apps_and_categories operations."""
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommended_apps_success_with_apps(self, mock_config, mock_factory_class, factory):
|
||||
"""Test successful retrieval of recommended apps when apps are returned."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
|
||||
expected_response = factory.create_recommended_apps_response()
|
||||
|
||||
# Mock factory and retrieval instance
|
||||
mock_retrieval_instance = MagicMock()
|
||||
mock_retrieval_instance.get_recommended_apps_and_categories.return_value = expected_response
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_retrieval_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
# Assert
|
||||
assert result == expected_response
|
||||
assert len(result["recommended_apps"]) == 2
|
||||
assert len(result["categories"]) == 3
|
||||
mock_factory_class.get_recommend_app_factory.assert_called_once_with("remote")
|
||||
mock_retrieval_instance.get_recommended_apps_and_categories.assert_called_once_with("en-US")
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommended_apps_fallback_to_builtin_when_empty(self, mock_config, mock_factory_class, factory):
|
||||
"""Test fallback to builtin when no recommended apps are returned."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
|
||||
# Remote returns empty recommended_apps
|
||||
empty_response = {"recommended_apps": [], "categories": []}
|
||||
|
||||
# Builtin fallback response
|
||||
builtin_response = factory.create_recommended_apps_response(
|
||||
recommended_apps=[{"id": "builtin-1", "name": "Builtin App", "category": "default"}]
|
||||
)
|
||||
|
||||
# Mock remote retrieval instance (returns empty)
|
||||
mock_remote_instance = MagicMock()
|
||||
mock_remote_instance.get_recommended_apps_and_categories.return_value = empty_response
|
||||
|
||||
mock_remote_factory = MagicMock()
|
||||
mock_remote_factory.return_value = mock_remote_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_remote_factory
|
||||
|
||||
# Mock builtin retrieval instance
|
||||
mock_builtin_instance = MagicMock()
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.return_value = builtin_response
|
||||
mock_factory_class.get_buildin_recommend_app_retrieval.return_value = mock_builtin_instance
|
||||
|
||||
# Act
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("zh-CN")
|
||||
|
||||
# Assert
|
||||
assert result == builtin_response
|
||||
assert len(result["recommended_apps"]) == 1
|
||||
assert result["recommended_apps"][0]["id"] == "builtin-1"
|
||||
# Verify fallback was called with en-US (hardcoded)
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.assert_called_once_with("en-US")
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommended_apps_fallback_when_none_recommended_apps(self, mock_config, mock_factory_class, factory):
|
||||
"""Test fallback when recommended_apps key is None."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "db"
|
||||
|
||||
# Response with None recommended_apps
|
||||
none_response = {"recommended_apps": None, "categories": ["test"]}
|
||||
|
||||
# Builtin fallback response
|
||||
builtin_response = factory.create_recommended_apps_response()
|
||||
|
||||
# Mock db retrieval instance (returns None)
|
||||
mock_db_instance = MagicMock()
|
||||
mock_db_instance.get_recommended_apps_and_categories.return_value = none_response
|
||||
|
||||
mock_db_factory = MagicMock()
|
||||
mock_db_factory.return_value = mock_db_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_db_factory
|
||||
|
||||
# Mock builtin retrieval instance
|
||||
mock_builtin_instance = MagicMock()
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.return_value = builtin_response
|
||||
mock_factory_class.get_buildin_recommend_app_retrieval.return_value = mock_builtin_instance
|
||||
|
||||
# Act
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
# Assert
|
||||
assert result == builtin_response
|
||||
mock_builtin_instance.fetch_recommended_apps_from_builtin.assert_called_once()
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommended_apps_with_different_languages(self, mock_config, mock_factory_class, factory):
|
||||
"""Test retrieval with different language codes."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "builtin"
|
||||
|
||||
languages = ["en-US", "zh-CN", "ja-JP", "fr-FR"]
|
||||
|
||||
for language in languages:
|
||||
# Create language-specific response
|
||||
lang_response = factory.create_recommended_apps_response(
|
||||
recommended_apps=[{"id": f"app-{language}", "name": f"App {language}", "category": "test"}]
|
||||
)
|
||||
|
||||
# Mock retrieval instance
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommended_apps_and_categories.return_value = lang_response
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories(language)
|
||||
|
||||
# Assert
|
||||
assert result["recommended_apps"][0]["id"] == f"app-{language}"
|
||||
mock_instance.get_recommended_apps_and_categories.assert_called_with(language)
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommended_apps_uses_correct_factory_mode(self, mock_config, mock_factory_class, factory):
|
||||
"""Test that correct factory is selected based on mode."""
|
||||
# Arrange
|
||||
modes = ["remote", "builtin", "db"]
|
||||
|
||||
for mode in modes:
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = mode
|
||||
|
||||
response = factory.create_recommended_apps_response()
|
||||
|
||||
# Mock retrieval instance
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommended_apps_and_categories.return_value = response
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
# Assert
|
||||
mock_factory_class.get_recommend_app_factory.assert_called_with(mode)
|
||||
|
||||
|
||||
class TestRecommendedAppServiceGetDetail:
|
||||
"""Test get_recommend_app_detail operations."""
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommend_app_detail_success(self, mock_config, mock_factory_class, factory):
|
||||
"""Test successful retrieval of app detail."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
app_id = "app-123"
|
||||
|
||||
expected_detail = factory.create_app_detail_response(
|
||||
app_id=app_id,
|
||||
name="Productivity App",
|
||||
description="A great productivity app",
|
||||
category="productivity",
|
||||
)
|
||||
|
||||
# Mock retrieval instance
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = expected_detail
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail(app_id))
|
||||
|
||||
# Assert
|
||||
assert result == expected_detail
|
||||
assert result["id"] == app_id
|
||||
assert result["name"] == "Productivity App"
|
||||
mock_instance.get_recommend_app_detail.assert_called_once_with(app_id)
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommend_app_detail_with_different_modes(self, mock_config, mock_factory_class, factory):
|
||||
"""Test app detail retrieval with different factory modes."""
|
||||
# Arrange
|
||||
modes = ["remote", "builtin", "db"]
|
||||
app_id = "test-app"
|
||||
|
||||
for mode in modes:
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = mode
|
||||
|
||||
detail = factory.create_app_detail_response(app_id=app_id, name=f"App from {mode}")
|
||||
|
||||
# Mock retrieval instance
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = detail
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail(app_id))
|
||||
|
||||
# Assert
|
||||
assert result["name"] == f"App from {mode}"
|
||||
mock_factory_class.get_recommend_app_factory.assert_called_with(mode)
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommend_app_detail_returns_none_when_not_found(self, mock_config, mock_factory_class, factory):
|
||||
"""Test that None is returned when app is not found."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
app_id = "nonexistent-app"
|
||||
|
||||
# Mock retrieval instance returning None
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = None
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail(app_id))
|
||||
|
||||
# Assert
|
||||
assert result is None
|
||||
mock_instance.get_recommend_app_detail.assert_called_once_with(app_id)
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommend_app_detail_returns_empty_dict(self, mock_config, mock_factory_class, factory):
|
||||
"""Test handling of empty dict response."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "builtin"
|
||||
app_id = "app-empty"
|
||||
|
||||
# Mock retrieval instance returning empty dict
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = {}
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail(app_id))
|
||||
|
||||
# Assert
|
||||
assert result == {}
|
||||
|
||||
@patch("services.recommended_app_service.RecommendAppRetrievalFactory", autospec=True)
|
||||
@patch("services.recommended_app_service.dify_config", autospec=True)
|
||||
def test_get_recommend_app_detail_with_complex_model_config(self, mock_config, mock_factory_class, factory):
|
||||
"""Test app detail with complex model configuration."""
|
||||
# Arrange
|
||||
mock_config.HOSTED_FETCH_APP_TEMPLATES_MODE = "remote"
|
||||
app_id = "complex-app"
|
||||
|
||||
complex_model_config = {
|
||||
"provider": "openai",
|
||||
"model": "gpt-4",
|
||||
"parameters": {
|
||||
"temperature": 0.7,
|
||||
"max_tokens": 2000,
|
||||
"top_p": 1.0,
|
||||
},
|
||||
}
|
||||
|
||||
expected_detail = factory.create_app_detail_response(
|
||||
app_id=app_id,
|
||||
name="Complex App",
|
||||
model_config=complex_model_config,
|
||||
workflows=["workflow-1", "workflow-2"],
|
||||
tools=["tool-1", "tool-2", "tool-3"],
|
||||
)
|
||||
|
||||
# Mock retrieval instance
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_recommend_app_detail.return_value = expected_detail
|
||||
|
||||
mock_factory = MagicMock()
|
||||
mock_factory.return_value = mock_instance
|
||||
mock_factory_class.get_recommend_app_factory.return_value = mock_factory
|
||||
|
||||
# Act
|
||||
result = _recommendation_detail(RecommendedAppService.get_recommend_app_detail(app_id))
|
||||
|
||||
# Assert
|
||||
assert result["model_config"] == complex_model_config
|
||||
assert len(result["workflows"]) == 2
|
||||
assert len(result["tools"]) == 3
|
||||
|
||||
|
||||
# === Merged from test_recommended_app_service_additional.py ===
|
||||
|
||||
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from services import recommended_app_service as service_module
|
||||
from services.recommended_app_service import RecommendedAppService
|
||||
|
||||
|
||||
def _recommendation_detail(result: dict[str, Any] | None) -> dict[str, Any]:
|
||||
return cast(dict[str, Any], result)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mocked_db_session(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
|
||||
# Arrange
|
||||
session = MagicMock()
|
||||
monkeypatch.setattr(service_module, "db", SimpleNamespace(session=session))
|
||||
|
||||
# Assert
|
||||
return session
|
||||
|
||||
|
||||
def _mock_factory_for_apps(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
*,
|
||||
mode: str,
|
||||
result: dict[str, Any],
|
||||
fallback_result: dict[str, Any] | None = None,
|
||||
) -> tuple[MagicMock, MagicMock]:
|
||||
retrieval_instance = MagicMock()
|
||||
retrieval_instance.get_recommended_apps_and_categories.return_value = result
|
||||
retrieval_factory = MagicMock(return_value=retrieval_instance)
|
||||
monkeypatch.setattr(service_module.dify_config, "HOSTED_FETCH_APP_TEMPLATES_MODE", mode, raising=False)
|
||||
monkeypatch.setattr(
|
||||
service_module.RecommendAppRetrievalFactory,
|
||||
"get_recommend_app_factory",
|
||||
MagicMock(return_value=retrieval_factory),
|
||||
)
|
||||
|
||||
builtin_instance = MagicMock()
|
||||
if fallback_result is not None:
|
||||
builtin_instance.fetch_recommended_apps_from_builtin.return_value = fallback_result
|
||||
monkeypatch.setattr(
|
||||
service_module.RecommendAppRetrievalFactory,
|
||||
"get_buildin_recommend_app_retrieval",
|
||||
MagicMock(return_value=builtin_instance),
|
||||
)
|
||||
return retrieval_instance, builtin_instance
|
||||
|
||||
|
||||
def test_get_recommended_apps_and_categories_should_not_query_trial_table_when_trial_feature_disabled(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
mocked_db_session: MagicMock,
|
||||
) -> None:
|
||||
# Arrange
|
||||
expected = {"recommended_apps": [{"app_id": "app-1"}], "categories": ["all"]}
|
||||
retrieval_instance, builtin_instance = _mock_factory_for_apps(
|
||||
monkeypatch,
|
||||
mode="remote",
|
||||
result=expected,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
MagicMock(return_value=SimpleNamespace(enable_trial_app=False)),
|
||||
)
|
||||
|
||||
# Act
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("en-US")
|
||||
|
||||
# Assert
|
||||
assert result == expected
|
||||
retrieval_instance.get_recommended_apps_and_categories.assert_called_once_with("en-US")
|
||||
builtin_instance.fetch_recommended_apps_from_builtin.assert_not_called()
|
||||
mocked_db_session.scalar.assert_not_called()
|
||||
|
||||
|
||||
def test_get_recommended_apps_and_categories_should_fallback_and_enrich_can_trial_when_trial_feature_enabled(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
mocked_db_session: MagicMock,
|
||||
) -> None:
|
||||
# Arrange
|
||||
remote_result = {"recommended_apps": [], "categories": []}
|
||||
fallback_result = {"recommended_apps": [{"app_id": "app-1"}, {"app_id": "app-2"}], "categories": ["all"]}
|
||||
_, builtin_instance = _mock_factory_for_apps(
|
||||
monkeypatch,
|
||||
mode="remote",
|
||||
result=remote_result,
|
||||
fallback_result=fallback_result,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
MagicMock(return_value=SimpleNamespace(enable_trial_app=True)),
|
||||
)
|
||||
mocked_db_session.scalar.side_effect = [SimpleNamespace(id="trial-app"), None]
|
||||
|
||||
# Act
|
||||
result = RecommendedAppService.get_recommended_apps_and_categories("ja-JP")
|
||||
|
||||
# Assert
|
||||
builtin_instance.fetch_recommended_apps_from_builtin.assert_called_once_with("en-US")
|
||||
assert result["recommended_apps"][0]["can_trial"] is True
|
||||
assert result["recommended_apps"][1]["can_trial"] is False
|
||||
assert mocked_db_session.scalar.call_count == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("trial_query_result", "expected_can_trial"),
|
||||
[
|
||||
(SimpleNamespace(id="trial"), True),
|
||||
(None, False),
|
||||
],
|
||||
)
|
||||
def test_get_recommend_app_detail_should_set_can_trial_when_trial_feature_enabled(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
mocked_db_session: MagicMock,
|
||||
trial_query_result: Any,
|
||||
expected_can_trial: bool,
|
||||
) -> None:
|
||||
# Arrange
|
||||
detail = {"id": "app-1", "name": "Test App"}
|
||||
retrieval_instance = MagicMock()
|
||||
retrieval_instance.get_recommend_app_detail.return_value = detail
|
||||
retrieval_factory = MagicMock(return_value=retrieval_instance)
|
||||
monkeypatch.setattr(service_module.dify_config, "HOSTED_FETCH_APP_TEMPLATES_MODE", "remote", raising=False)
|
||||
monkeypatch.setattr(
|
||||
service_module.RecommendAppRetrievalFactory,
|
||||
"get_recommend_app_factory",
|
||||
MagicMock(return_value=retrieval_factory),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
MagicMock(return_value=SimpleNamespace(enable_trial_app=True)),
|
||||
)
|
||||
mocked_db_session.scalar.return_value = trial_query_result
|
||||
|
||||
# Act
|
||||
result = cast(dict[str, Any], RecommendedAppService.get_recommend_app_detail("app-1"))
|
||||
|
||||
# Assert
|
||||
assert result["id"] == "app-1"
|
||||
assert result["can_trial"] is expected_can_trial
|
||||
mocked_db_session.scalar.assert_called_once()
|
||||
|
||||
|
||||
def test_add_trial_app_record_should_increment_count_when_existing_record_found(
|
||||
mocked_db_session: MagicMock,
|
||||
) -> None:
|
||||
# Arrange
|
||||
existing_record = SimpleNamespace(count=3)
|
||||
mocked_db_session.scalar.return_value = existing_record
|
||||
|
||||
# Act
|
||||
RecommendedAppService.add_trial_app_record("app-1", "account-1")
|
||||
|
||||
# Assert
|
||||
assert existing_record.count == 4
|
||||
mocked_db_session.scalar.assert_called_once()
|
||||
mocked_db_session.commit.assert_called_once()
|
||||
mocked_db_session.add.assert_not_called()
|
||||
|
||||
|
||||
def test_add_trial_app_record_should_create_new_record_when_no_existing_record(
|
||||
mocked_db_session: MagicMock,
|
||||
) -> None:
|
||||
# Arrange
|
||||
mocked_db_session.scalar.return_value = None
|
||||
|
||||
# Act
|
||||
RecommendedAppService.add_trial_app_record("app-2", "account-2")
|
||||
|
||||
# Assert
|
||||
mocked_db_session.scalar.assert_called_once()
|
||||
mocked_db_session.add.assert_called_once()
|
||||
added = mocked_db_session.add.call_args.args[0]
|
||||
assert added.app_id == "app-2"
|
||||
assert added.account_id == "account-2"
|
||||
assert added.count == 1
|
||||
mocked_db_session.commit.assert_called_once()
|
||||
Loading…
Reference in New Issue
Block a user