mirror of
https://github.com/langgenius/dify.git
synced 2026-09-08 11:04:27 +08:00
test: use SQLite sessions in controllers console explore (#39064)
This commit is contained in:
Asuka Minato
GitHub
parent
0a05328eb2
commit
ca1cec6011
@ -1,16 +1,20 @@
|
||||
from collections.abc import Callable
|
||||
from contextlib import AbstractContextManager
|
||||
from collections.abc import Callable, Generator
|
||||
from contextlib import AbstractContextManager, contextmanager
|
||||
from datetime import datetime
|
||||
from inspect import unwrap
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, PropertyMock, patch
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
from werkzeug.exceptions import BadRequest, Forbidden, NotFound
|
||||
|
||||
import controllers.console.explore.installed_app as module
|
||||
import services.installed_app_service as service_module
|
||||
from models.model import AppMode, IconType
|
||||
from models.model import App, AppMode, AppModelConfig, IconType, InstalledApp, RecommendedApp
|
||||
from models.workflow import Workflow, WorkflowKind, WorkflowType
|
||||
|
||||
type Payload = dict[str, object]
|
||||
type PayloadPatch = Callable[[Payload], AbstractContextManager[object]]
|
||||
@ -132,7 +136,7 @@ class TestInstalledAppsListApi:
|
||||
method = unwrap(api.get)
|
||||
|
||||
session = MagicMock()
|
||||
session.execute.return_value.all.return_value = []
|
||||
session.execute.return_value.all.return_value = list[tuple[InstalledApp, App]]()
|
||||
|
||||
with (
|
||||
app.test_request_context("/?app_id=a1"),
|
||||
@ -152,7 +156,7 @@ class TestInstalledAppsListApi:
|
||||
api = module.InstalledAppsListApi()
|
||||
method = unwrap(api.get)
|
||||
session = MagicMock()
|
||||
session.execute.return_value.all.return_value = []
|
||||
session.execute.return_value.all.return_value = list[tuple[InstalledApp, App]]()
|
||||
|
||||
with (
|
||||
app.test_request_context("/?name=Sales%25_Q3"),
|
||||
@ -439,7 +443,7 @@ class TestInstalledAppsListApi:
|
||||
method = unwrap(api.get)
|
||||
|
||||
session = MagicMock()
|
||||
session.execute.return_value.all.return_value = []
|
||||
session.execute.return_value.all.return_value = list[tuple[InstalledApp, App]]()
|
||||
|
||||
with (
|
||||
app.test_request_context("/"),
|
||||
@ -462,7 +466,7 @@ class TestInstalledAppsListApi:
|
||||
method = unwrap(api.get)
|
||||
|
||||
session = MagicMock()
|
||||
session.execute.return_value.all.return_value = []
|
||||
session.execute.return_value.all.return_value = list[tuple[InstalledApp, App]]()
|
||||
|
||||
with (
|
||||
app.test_request_context("/"),
|
||||
@ -485,7 +489,7 @@ class TestInstalledAppsListApi:
|
||||
method = unwrap(api.get)
|
||||
|
||||
session = MagicMock()
|
||||
session.execute.return_value.all.return_value = []
|
||||
session.execute.return_value.all.return_value = list[tuple[InstalledApp, App]]()
|
||||
|
||||
with (
|
||||
app.test_request_context("/"),
|
||||
@ -674,3 +678,294 @@ class TestInstalledAppApi:
|
||||
result = method(installed_app)
|
||||
|
||||
assert result["result"] == "success"
|
||||
|
||||
|
||||
def _persist_app(
|
||||
session: Session,
|
||||
*,
|
||||
app_id: str = "app-1",
|
||||
tenant_id: str = "owner-tenant",
|
||||
mode: AppMode = AppMode.CHAT,
|
||||
public: bool = True,
|
||||
published: bool = True,
|
||||
) -> App:
|
||||
app = App(
|
||||
id=app_id,
|
||||
tenant_id=tenant_id,
|
||||
name=f"App {app_id}",
|
||||
description="description",
|
||||
mode=mode,
|
||||
icon_type=None,
|
||||
icon=None,
|
||||
icon_background=None,
|
||||
enable_site=True,
|
||||
enable_api=True,
|
||||
is_public=public,
|
||||
max_active_requests=None,
|
||||
)
|
||||
session.add(app)
|
||||
session.flush()
|
||||
if published and mode in {AppMode.WORKFLOW, AppMode.ADVANCED_CHAT}:
|
||||
workflow = Workflow(
|
||||
id=f"workflow-{app_id}",
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
type=WorkflowType.WORKFLOW,
|
||||
kind=WorkflowKind.STANDARD,
|
||||
version="1",
|
||||
graph='{"nodes":[],"edges":[]}',
|
||||
features="{}",
|
||||
created_by="user-1",
|
||||
environment_variables=[],
|
||||
conversation_variables=[],
|
||||
rag_pipeline_variables=[],
|
||||
)
|
||||
session.add(workflow)
|
||||
app.workflow_id = workflow.id
|
||||
elif published:
|
||||
model_config = AppModelConfig(app_id=app_id)
|
||||
session.add(model_config)
|
||||
session.flush()
|
||||
app.app_model_config_id = model_config.id
|
||||
session.commit()
|
||||
return app
|
||||
|
||||
|
||||
def _persist_installed_app(
|
||||
session: Session,
|
||||
app: App,
|
||||
*,
|
||||
tenant_id: str,
|
||||
pinned: bool = False,
|
||||
) -> InstalledApp:
|
||||
installed = InstalledApp(
|
||||
app_id=app.id,
|
||||
tenant_id=tenant_id,
|
||||
app_owner_tenant_id=app.tenant_id,
|
||||
is_pinned=pinned,
|
||||
last_used_at=datetime(2024, 1, 1),
|
||||
)
|
||||
session.add(installed)
|
||||
session.commit()
|
||||
return installed
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _sqlite_controller_context(
|
||||
database: Session, *, role: str = "owner", auth_enabled: bool = False
|
||||
) -> Generator[None]:
|
||||
session_proxy = MagicMock(wraps=database)
|
||||
session_proxy.return_value = database
|
||||
with (
|
||||
patch.object(module.db, "session", session_proxy),
|
||||
patch.object(module.TenantService, "get_user_role", return_value=role),
|
||||
patch.object(
|
||||
service_module.FeatureService,
|
||||
"get_system_features",
|
||||
return_value=MagicMock(webapp_auth=MagicMock(enabled=auth_enabled)),
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
def test_sqlite_get_installed_apps_filters_tenant_publication_mode_and_app_id(
|
||||
app: Flask,
|
||||
current_user: MagicMock,
|
||||
tenant_id: str,
|
||||
sqlite_session: Session,
|
||||
) -> None:
|
||||
database = sqlite_session
|
||||
chat = _persist_app(database, app_id="chat")
|
||||
workflow = _persist_app(database, app_id="workflow", mode=AppMode.WORKFLOW)
|
||||
unpublished = _persist_app(database, app_id="unpublished", published=False)
|
||||
agent = _persist_app(database, app_id="agent", mode=AppMode.AGENT)
|
||||
foreign = _persist_app(database, app_id="foreign")
|
||||
for model, installed_tenant in (
|
||||
(chat, tenant_id),
|
||||
(workflow, tenant_id),
|
||||
(unpublished, tenant_id),
|
||||
(agent, tenant_id),
|
||||
(foreign, "other-tenant"),
|
||||
):
|
||||
_persist_installed_app(database, model, tenant_id=installed_tenant)
|
||||
|
||||
api = module.InstalledAppsListApi()
|
||||
method = unwrap(api.get)
|
||||
with app.test_request_context("/"), _sqlite_controller_context(database):
|
||||
result = method(api, tenant_id, current_user)
|
||||
|
||||
assert {item["app"]["id"] for item in result["installed_apps"]} == {"chat", "workflow"}
|
||||
assert all(item["editable"] is True for item in result["installed_apps"])
|
||||
assert all(item["uninstallable"] is False for item in result["installed_apps"])
|
||||
|
||||
with app.test_request_context("/?app_id=workflow"), _sqlite_controller_context(database, role="member"):
|
||||
filtered = method(api, tenant_id, current_user)
|
||||
assert [item["app"]["id"] for item in filtered["installed_apps"]] == ["workflow"]
|
||||
assert filtered["installed_apps"][0]["editable"] is False
|
||||
|
||||
|
||||
def test_sqlite_get_installed_apps_applies_web_auth_permission_state(
|
||||
app: Flask,
|
||||
current_user: MagicMock,
|
||||
tenant_id: str,
|
||||
sqlite_session: Session,
|
||||
) -> None:
|
||||
database = sqlite_session
|
||||
allowed = _persist_app(database, app_id="allowed")
|
||||
denied = _persist_app(database, app_id="denied")
|
||||
sso = _persist_app(database, app_id="sso")
|
||||
for model in (allowed, denied, sso):
|
||||
_persist_installed_app(database, model, tenant_id=tenant_id)
|
||||
settings = {
|
||||
"allowed": SimpleNamespace(access_mode="restricted"),
|
||||
"denied": SimpleNamespace(access_mode="restricted"),
|
||||
"sso": SimpleNamespace(access_mode="sso_verified"),
|
||||
}
|
||||
|
||||
api = module.InstalledAppsListApi()
|
||||
method = unwrap(api.get)
|
||||
with (
|
||||
app.test_request_context("/"),
|
||||
_sqlite_controller_context(database, auth_enabled=True),
|
||||
patch.object(
|
||||
service_module.EnterpriseService.WebAppAuth, "batch_get_app_access_mode_by_id", return_value=settings
|
||||
),
|
||||
patch.object(
|
||||
service_module.EnterpriseService.WebAppAuth,
|
||||
"batch_is_user_allowed_to_access_webapps",
|
||||
return_value={"allowed": True, "denied": False},
|
||||
),
|
||||
):
|
||||
result = method(api, tenant_id, current_user)
|
||||
|
||||
assert [item["app"]["id"] for item in result["installed_apps"]] == ["allowed"]
|
||||
|
||||
|
||||
def test_sqlite_post_installs_public_recommended_app_and_is_idempotent(
|
||||
app: Flask,
|
||||
tenant_id: str,
|
||||
payload_patch: PayloadPatch,
|
||||
sqlite_session: Session,
|
||||
) -> None:
|
||||
database = sqlite_session
|
||||
app_model = _persist_app(database, public=True)
|
||||
recommended = RecommendedApp(
|
||||
app_id=app_model.id,
|
||||
description={"en-US": "recommended"},
|
||||
copyright="copyright",
|
||||
privacy_policy="https://example.com/privacy",
|
||||
category="productivity",
|
||||
)
|
||||
database.add(recommended)
|
||||
database.commit()
|
||||
recommended_id = recommended.id
|
||||
app_owner_tenant_id = app_model.tenant_id
|
||||
api = module.InstalledAppsListApi()
|
||||
method = unwrap(api.post)
|
||||
|
||||
for _ in range(2):
|
||||
with (
|
||||
app.test_request_context("/", json={"app_id": app_model.id}),
|
||||
payload_patch({"app_id": app_model.id}),
|
||||
patch.object(module.db, "session", database),
|
||||
):
|
||||
assert method(api, tenant_id) == {"message": "App installed successfully"}
|
||||
|
||||
# End the request-scoped session so this assertion only observes committed data.
|
||||
bind = database.get_bind()
|
||||
database.close()
|
||||
with Session(bind) as verification_session:
|
||||
installed = verification_session.scalars(select(InstalledApp)).all()
|
||||
assert len(installed) == 1
|
||||
assert installed[0].tenant_id == tenant_id
|
||||
assert installed[0].app_owner_tenant_id == app_owner_tenant_id
|
||||
persisted_recommendation = verification_session.get(RecommendedApp, recommended_id)
|
||||
assert persisted_recommendation is not None
|
||||
assert persisted_recommendation.install_count == 1
|
||||
|
||||
|
||||
def test_sqlite_post_enforces_recommendation_and_public_state(
|
||||
app: Flask,
|
||||
tenant_id: str,
|
||||
payload_patch: PayloadPatch,
|
||||
sqlite_session: Session,
|
||||
) -> None:
|
||||
database = sqlite_session
|
||||
api = module.InstalledAppsListApi()
|
||||
method = unwrap(api.post)
|
||||
with (
|
||||
app.test_request_context("/", json={"app_id": "missing"}),
|
||||
payload_patch({"app_id": "missing"}),
|
||||
patch.object(module.db, "session", database),
|
||||
pytest.raises(NotFound),
|
||||
):
|
||||
method(api, tenant_id)
|
||||
|
||||
private_app = _persist_app(database, app_id="private", public=False)
|
||||
database.add(
|
||||
RecommendedApp(
|
||||
app_id=private_app.id,
|
||||
description={},
|
||||
copyright="copyright",
|
||||
privacy_policy="privacy",
|
||||
category="category",
|
||||
)
|
||||
)
|
||||
database.commit()
|
||||
with (
|
||||
app.test_request_context("/", json={"app_id": private_app.id}),
|
||||
payload_patch({"app_id": private_app.id}),
|
||||
patch.object(module.db, "session", database),
|
||||
pytest.raises(Forbidden),
|
||||
):
|
||||
method(api, tenant_id)
|
||||
|
||||
|
||||
def test_sqlite_delete_removes_foreign_installed_app_and_rejects_owned_app(
|
||||
tenant_id: str,
|
||||
sqlite_session: Session,
|
||||
) -> None:
|
||||
database = sqlite_session
|
||||
foreign_app = _persist_app(database, app_id="foreign")
|
||||
installed = _persist_installed_app(database, foreign_app, tenant_id=tenant_id)
|
||||
installed_id = installed.id
|
||||
api = module.InstalledAppApi()
|
||||
with patch.object(module.db, "session", database):
|
||||
response, status = unwrap(api.delete)(api, tenant_id, installed)
|
||||
assert (response, status) == ("", 204)
|
||||
assert database.get(InstalledApp, installed_id) is None
|
||||
|
||||
owned_app = _persist_app(database, app_id="owned", tenant_id=tenant_id)
|
||||
owned_install = _persist_installed_app(database, owned_app, tenant_id=tenant_id)
|
||||
with pytest.raises(BadRequest):
|
||||
unwrap(api.delete)(api, tenant_id, owned_install)
|
||||
assert database.get(InstalledApp, owned_install.id) is not None
|
||||
|
||||
|
||||
def test_sqlite_patch_persists_pin_and_noop_payload(
|
||||
app: Flask,
|
||||
tenant_id: str,
|
||||
payload_patch: PayloadPatch,
|
||||
sqlite_session: Session,
|
||||
) -> None:
|
||||
database = sqlite_session
|
||||
app_model = _persist_app(database)
|
||||
installed = _persist_installed_app(database, app_model, tenant_id=tenant_id)
|
||||
api = module.InstalledAppApi()
|
||||
with (
|
||||
app.test_request_context("/", json={"is_pinned": True}),
|
||||
payload_patch({"is_pinned": True}),
|
||||
patch.object(module.db, "session", database),
|
||||
):
|
||||
assert unwrap(api.patch)(installed)["result"] == "success"
|
||||
database.expire_all()
|
||||
persisted_installed_app = database.get(InstalledApp, installed.id)
|
||||
assert persisted_installed_app is not None
|
||||
assert persisted_installed_app.is_pinned is True
|
||||
|
||||
with (
|
||||
app.test_request_context("/", json={}),
|
||||
payload_patch({}),
|
||||
patch.object(module.db, "session", database),
|
||||
):
|
||||
assert unwrap(api.patch)(installed)["result"] == "success"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user