kudzues/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-04-05 06:06:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: migrate database retrieval tests to testcontainers (#34087)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
YBoy
2026-03-25 16:06:46 +02:00
committed by GitHub
parent 22dd0aa20c
commit baf7d2c7c0
3 changed files with 184 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
api/tests/test_containers_integration_tests/services/recommend_app/__init__.py Normal file
View File

184
api/tests/test_containers_integration_tests/services/recommend_app/test_database_retrieval.py Normal file
View File

@@ -0,0 +1,184 @@
from __future__ import annotations
from unittest.mock import patch
from uuid import uuid4
from models.model import App, RecommendedApp, Site
from services.recommend_app.database.database_retrieval import DatabaseRecommendAppRetrieval
from services.recommend_app.recommend_app_type import RecommendAppType
def _create_app(db_session, *, tenant_id: str, is_public: bool = True) -> App:
app = App(
tenant_id=tenant_id,
name=f"app-{uuid4()}",
mode="chat",
enable_site=True,
enable_api=True,
is_public=is_public,
)
app.id = str(uuid4())
db_session.add(app)
db_session.commit()
return app
def _create_site(db_session, *, app_id: str) -> Site:
site = Site(
app_id=app_id,
title=f"site-{uuid4()}",
default_language="en-US",
customize_token_strategy="not_allow",
description="desc",
copyright="copy",
privacy_policy="pp",
custom_disclaimer="cd",
)
site.id = str(uuid4())
db_session.add(site)
db_session.commit()
return site
def _create_recommended_app(
db_session,
*,
app_id: str,
category: str = "chat",
language: str = "en-US",
is_listed: bool = True,
position: int = 1,
) -> RecommendedApp:
rec = RecommendedApp(
app_id=app_id,
description={"en-US": "test"},
copyright="copy",
privacy_policy="pp",
category=category,
language=language,
is_listed=is_listed,
position=position,
)
rec.id = str(uuid4())
db_session.add(rec)
db_session.commit()
return rec
class TestDatabaseRecommendAppRetrieval:
def test_get_type(self):
assert DatabaseRecommendAppRetrieval().get_type() == RecommendAppType.DATABASE
def test_get_recommended_apps_delegates(self):
with patch.object(
DatabaseRecommendAppRetrieval,
"fetch_recommended_apps_from_db",
return_value={"recommended_apps": [], "categories": []},
) as mock_fetch:
result = DatabaseRecommendAppRetrieval().get_recommended_apps_and_categories("en-US")
mock_fetch.assert_called_once_with("en-US")
assert result == {"recommended_apps": [], "categories": []}
def test_get_recommend_app_detail_delegates(self):
with patch.object(
DatabaseRecommendAppRetrieval,
"fetch_recommended_app_detail_from_db",
return_value={"id": "app-1"},
) as mock_fetch:
result = DatabaseRecommendAppRetrieval().get_recommend_app_detail("app-1")
mock_fetch.assert_called_once_with("app-1")
assert result == {"id": "app-1"}
class TestFetchRecommendedAppsFromDb:
def test_returns_apps_and_sorted_categories(self, flask_app_with_containers, db_session_with_containers):
tenant_id = str(uuid4())
app1 = _create_app(db_session_with_containers, tenant_id=tenant_id)
_create_site(db_session_with_containers, app_id=app1.id)
_create_recommended_app(db_session_with_containers, app_id=app1.id, category="writing")
app2 = _create_app(db_session_with_containers, tenant_id=tenant_id)
_create_site(db_session_with_containers, app_id=app2.id)
_create_recommended_app(db_session_with_containers, app_id=app2.id, category="assistant")
db_session_with_containers.expire_all()
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("en-US")
app_ids = {r["app_id"] for r in result["recommended_apps"]}
assert app1.id in app_ids
assert app2.id in app_ids
assert "assistant" in result["categories"]
assert "writing" in result["categories"]
def test_falls_back_to_default_language_when_empty(self, flask_app_with_containers, db_session_with_containers):
tenant_id = str(uuid4())
app1 = _create_app(db_session_with_containers, tenant_id=tenant_id)
_create_site(db_session_with_containers, app_id=app1.id)
_create_recommended_app(db_session_with_containers, app_id=app1.id, language="en-US")
db_session_with_containers.expire_all()
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("fr-FR")
app_ids = {r["app_id"] for r in result["recommended_apps"]}
assert app1.id in app_ids
def test_skips_non_public_apps(self, flask_app_with_containers, db_session_with_containers):
tenant_id = str(uuid4())
app1 = _create_app(db_session_with_containers, tenant_id=tenant_id, is_public=False)
_create_site(db_session_with_containers, app_id=app1.id)
_create_recommended_app(db_session_with_containers, app_id=app1.id)
db_session_with_containers.expire_all()
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("en-US")
app_ids = {r["app_id"] for r in result["recommended_apps"]}
assert app1.id not in app_ids
def test_skips_apps_without_site(self, flask_app_with_containers, db_session_with_containers):
tenant_id = str(uuid4())
app1 = _create_app(db_session_with_containers, tenant_id=tenant_id)
_create_recommended_app(db_session_with_containers, app_id=app1.id)
db_session_with_containers.expire_all()
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("en-US")
app_ids = {r["app_id"] for r in result["recommended_apps"]}
assert app1.id not in app_ids
class TestFetchRecommendedAppDetailFromDb:
def test_returns_none_when_not_listed(self, flask_app_with_containers, db_session_with_containers):
result = DatabaseRecommendAppRetrieval.fetch_recommended_app_detail_from_db(str(uuid4()))
assert result is None
def test_returns_none_when_app_not_public(self, flask_app_with_containers, db_session_with_containers):
tenant_id = str(uuid4())
app1 = _create_app(db_session_with_containers, tenant_id=tenant_id, is_public=False)
_create_recommended_app(db_session_with_containers, app_id=app1.id)
db_session_with_containers.expire_all()
result = DatabaseRecommendAppRetrieval.fetch_recommended_app_detail_from_db(app1.id)
assert result is None
@patch("services.recommend_app.database.database_retrieval.AppDslService")
def test_returns_detail_on_success(self, mock_dsl, flask_app_with_containers, db_session_with_containers):
tenant_id = str(uuid4())
app1 = _create_app(db_session_with_containers, tenant_id=tenant_id)
_create_site(db_session_with_containers, app_id=app1.id)
_create_recommended_app(db_session_with_containers, app_id=app1.id)
mock_dsl.export_dsl.return_value = "exported_yaml"
db_session_with_containers.expire_all()
result = DatabaseRecommendAppRetrieval.fetch_recommended_app_detail_from_db(app1.id)
assert result is not None
assert result["id"] == app1.id
assert result["export_data"] == "exported_yaml"

145
api/tests/unit_tests/services/recommend_app/test_database_retrieval.py
View File

@@ -1,145 +0,0 @@
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from services.recommend_app.database.database_retrieval import DatabaseRecommendAppRetrieval
from services.recommend_app.recommend_app_type import RecommendAppType
class TestDatabaseRecommendAppRetrieval:
def test_get_type(self):
assert DatabaseRecommendAppRetrieval().get_type() == RecommendAppType.DATABASE
def test_get_recommended_apps_delegates(self):
with patch.object(
DatabaseRecommendAppRetrieval,
"fetch_recommended_apps_from_db",
return_value={"recommended_apps": [], "categories": []},
) as mock_fetch:
result = DatabaseRecommendAppRetrieval().get_recommended_apps_and_categories("en-US")
mock_fetch.assert_called_once_with("en-US")
assert result == {"recommended_apps": [], "categories": []}
def test_get_recommend_app_detail_delegates(self):
with patch.object(
DatabaseRecommendAppRetrieval,
"fetch_recommended_app_detail_from_db",
return_value={"id": "app-1"},
) as mock_fetch:
result = DatabaseRecommendAppRetrieval().get_recommend_app_detail("app-1")
mock_fetch.assert_called_once_with("app-1")
assert result == {"id": "app-1"}
class TestFetchRecommendedAppsFromDb:
def _make_recommended_app(self, app_id, category, is_public=True, has_site=True):
site = (
SimpleNamespace(
description="desc",
copyright="copy",
privacy_policy="pp",
custom_disclaimer="cd",
)
if has_site
else None
)
app = (
SimpleNamespace(is_public=is_public, site=site)
if is_public
else SimpleNamespace(is_public=False, site=site)
)
return SimpleNamespace(
id=f"rec-{app_id}",
app=app,
app_id=app_id,
category=category,
position=1,
is_listed=True,
)
@patch("services.recommend_app.database.database_retrieval.db")
def test_returns_apps_and_sorted_categories(self, mock_db):
rec1 = self._make_recommended_app("a1", "writing")
rec2 = self._make_recommended_app("a2", "assistant")
mock_db.session.scalars.return_value.all.return_value = [rec1, rec2]
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("en-US")
assert len(result["recommended_apps"]) == 2
assert result["categories"] == ["assistant", "writing"]
@patch("services.recommend_app.database.database_retrieval.db")
def test_falls_back_to_default_language_when_empty(self, mock_db):
mock_db.session.scalars.return_value.all.side_effect = [
[],
[self._make_recommended_app("a1", "chat")],
]
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("fr-FR")
assert len(result["recommended_apps"]) == 1
assert mock_db.session.scalars.call_count == 2
@patch("services.recommend_app.database.database_retrieval.db")
def test_skips_non_public_apps(self, mock_db):
rec = self._make_recommended_app("a1", "chat", is_public=False)
mock_db.session.scalars.return_value.all.return_value = [rec]
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("en-US")
assert result["recommended_apps"] == []
@patch("services.recommend_app.database.database_retrieval.db")
def test_skips_apps_without_site(self, mock_db):
rec = self._make_recommended_app("a1", "chat", has_site=False)
mock_db.session.scalars.return_value.all.return_value = [rec]
result = DatabaseRecommendAppRetrieval.fetch_recommended_apps_from_db("en-US")
assert result["recommended_apps"] == []
class TestFetchRecommendedAppDetailFromDb:
@patch("services.recommend_app.database.database_retrieval.db")
def test_returns_none_when_not_listed(self, mock_db):
mock_db.session.query.return_value.where.return_value.first.return_value = None
result = DatabaseRecommendAppRetrieval.fetch_recommended_app_detail_from_db("app-1")
assert result is None
@patch("services.recommend_app.database.database_retrieval.AppDslService")
@patch("services.recommend_app.database.database_retrieval.db")
def test_returns_none_when_app_not_public(self, mock_db, mock_dsl):
rec_chain = MagicMock()
rec_chain.where.return_value.first.return_value = SimpleNamespace(app_id="app-1")
app_chain = MagicMock()
app_chain.where.return_value.first.return_value = SimpleNamespace(id="app-1", is_public=False)
mock_db.session.query.side_effect = [rec_chain, app_chain]
result = DatabaseRecommendAppRetrieval.fetch_recommended_app_detail_from_db("app-1")
assert result is None
@patch("services.recommend_app.database.database_retrieval.AppDslService")
@patch("services.recommend_app.database.database_retrieval.db")
def test_returns_detail_on_success(self, mock_db, mock_dsl):
app_model = SimpleNamespace(
id="app-1",
name="My App",
icon="icon.png",
icon_background="#fff",
mode="chat",
is_public=True,
)
rec_chain = MagicMock()
rec_chain.where.return_value.first.return_value = SimpleNamespace(app_id="app-1")
app_chain = MagicMock()
app_chain.where.return_value.first.return_value = app_model
mock_db.session.query.side_effect = [rec_chain, app_chain]
mock_dsl.export_dsl.return_value = "exported_yaml"
result = DatabaseRecommendAppRetrieval.fetch_recommended_app_detail_from_db("app-1")
assert result["id"] == "app-1"
assert result["name"] == "My App"
assert result["export_data"] == "exported_yaml"