opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 10:38:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use sqlite3 session in test_remote_retrieval (#38748)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-13 16:25:38 +09:00 committed by GitHub
parent 57f325df27
commit 0cea67e72d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 36 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
api/tests/unit_tests/services/recommend_app/test_remote_retrieval.py
View File

@ -1,12 +1,21 @@
from collections.abc import Iterator
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from sqlalchemy import Engine
from sqlalchemy.orm import Session
from services.recommend_app.recommend_app_type import RecommendAppType
from services.recommend_app.remote.remote_retrieval import RemoteRecommendAppRetrieval
@pytest.fixture
def empty_sqlite_session(sqlite_engine: Engine) -> Iterator[Session]:
with Session(sqlite_engine) as session:
yield session
class TestRemoteRecommendAppRetrieval:
def test_get_type(self):
assert RemoteRecommendAppRetrieval().get_type() == RecommendAppType.REMOTE
@ -16,10 +25,11 @@ class TestRemoteRecommendAppRetrieval:
"fetch_recommended_app_detail_from_dify_official",
return_value={"id": "app-1"},
)
def test_get_recommend_app_detail_success(self, mock_fetch):
result = RemoteRecommendAppRetrieval().get_recommend_app_detail("app-1", session=MagicMock())
def test_get_recommend_app_detail_success(self, mock_fetch, empty_sqlite_session: Session):
result = RemoteRecommendAppRetrieval().get_recommend_app_detail("app-1", session=empty_sqlite_session)
assert result == {"id": "app-1"}
mock_fetch.assert_called_once_with("app-1")
assert not empty_sqlite_session.in_transaction()
@patch(
"services.recommend_app.remote.remote_retrieval"
@ -31,19 +41,25 @@ class TestRemoteRecommendAppRetrieval:
"fetch_recommended_app_detail_from_dify_official",
side_effect=ConnectionError("timeout"),
)
def test_get_recommend_app_detail_falls_back_on_error(self, mock_fetch, mock_builtin):
result = RemoteRecommendAppRetrieval().get_recommend_app_detail("app-1", session=MagicMock())
def test_get_recommend_app_detail_falls_back_on_error(
self, mock_fetch, mock_builtin, empty_sqlite_session: Session
):
result = RemoteRecommendAppRetrieval().get_recommend_app_detail("app-1", session=empty_sqlite_session)
assert result == {"id": "fallback"}
mock_builtin.assert_called_once_with("app-1")
assert not empty_sqlite_session.in_transaction()
@patch.object(
RemoteRecommendAppRetrieval,
"fetch_recommended_apps_from_dify_official",
return_value={"recommended_apps": [], "categories": []},
)
def test_get_recommended_apps_success(self, mock_fetch):
result = RemoteRecommendAppRetrieval().get_recommended_apps_and_categories("en-US", session=MagicMock())
def test_get_recommended_apps_success(self, mock_fetch, empty_sqlite_session: Session):
result = RemoteRecommendAppRetrieval().get_recommended_apps_and_categories(
"en-US", session=empty_sqlite_session
)
assert result == {"recommended_apps": [], "categories": []}
assert not empty_sqlite_session.in_transaction()
@patch(
"services.recommend_app.remote.remote_retrieval"
@ -55,20 +71,24 @@ class TestRemoteRecommendAppRetrieval:
"fetch_recommended_apps_from_dify_official",
side_effect=ValueError("server error"),
)
def test_get_recommended_apps_falls_back_on_error(self, mock_fetch, mock_builtin):
result = RemoteRecommendAppRetrieval().get_recommended_apps_and_categories("en-US", session=MagicMock())
def test_get_recommended_apps_falls_back_on_error(self, mock_fetch, mock_builtin, empty_sqlite_session: Session):
result = RemoteRecommendAppRetrieval().get_recommended_apps_and_categories(
"en-US", session=empty_sqlite_session
)
assert result == {"recommended_apps": [{"id": "builtin"}]}
assert not empty_sqlite_session.in_transaction()
@patch.object(
RemoteRecommendAppRetrieval,
"fetch_learn_dify_apps_from_dify_official",
return_value={"recommended_apps": [{"id": "learn-dify-app"}]},
)
def test_get_learn_dify_apps_success(self, mock_fetch):
result = RemoteRecommendAppRetrieval().get_learn_dify_apps("en-US", session=MagicMock())
def test_get_learn_dify_apps_success(self, mock_fetch, empty_sqlite_session: Session):
result = RemoteRecommendAppRetrieval().get_learn_dify_apps("en-US", session=empty_sqlite_session)
assert result == {"recommended_apps": [{"id": "learn-dify-app"}]}
mock_fetch.assert_called_once_with("en-US")
assert not empty_sqlite_session.in_transaction()
@patch(
"services.recommend_app.remote.remote_retrieval.DatabaseRecommendAppRetrieval.fetch_learn_dify_apps_from_db",
@ -79,13 +99,14 @@ class TestRemoteRecommendAppRetrieval:
"fetch_learn_dify_apps_from_dify_official",
side_effect=ValueError("server error"),
)
def test_get_learn_dify_apps_falls_back_to_database_on_error(self, mock_fetch, mock_database):
session = MagicMock()
result = RemoteRecommendAppRetrieval().get_learn_dify_apps("en-US", session=session)
def test_get_learn_dify_apps_falls_back_to_database_on_error(
self, mock_fetch, mock_database, empty_sqlite_session: Session
):
result = RemoteRecommendAppRetrieval().get_learn_dify_apps("en-US", session=empty_sqlite_session)
assert result == {"recommended_apps": [{"id": "db-fallback"}]}
mock_database.assert_called_once_with("en-US", session=session)
mock_database.assert_called_once_with("en-US", session=empty_sqlite_session)
assert not empty_sqlite_session.in_transaction()
class TestFetchFromDifyOfficial: