From ca1cec601136f69a9fb2f7a8d1fdbf23b8803333 Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Tue, 4 Aug 2026 00:44:43 +0900 Subject: [PATCH] test: use SQLite sessions in controllers console explore (#39064) --- .../console/explore/test_installed_app.py | 311 +++++++++++++++++- 1 file changed, 303 insertions(+), 8 deletions(-) diff --git a/api/tests/unit_tests/controllers/console/explore/test_installed_app.py b/api/tests/unit_tests/controllers/console/explore/test_installed_app.py index 4d038a66b7b..a589b2b2dd3 100644 --- a/api/tests/unit_tests/controllers/console/explore/test_installed_app.py +++ b/api/tests/unit_tests/controllers/console/explore/test_installed_app.py @@ -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"