test: migrate console workflow sessions and ORM models to SQLite (#40563)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/controllers/console/app/workflow.py

@@ -586,14 +586,14 @@ class DraftWorkflowApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()))
+    @with_session(write=False)
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
+    def get(self, session: Session, app_model: App):
         """
         Get draft workflow
         """
         # fetch draft workflow by app_model
         workflow_service = WorkflowService()
-        session = db.session()
         workflow = workflow_service.get_draft_workflow(app_model=app_model, session=session)
 
         if not workflow:
@@ -1489,11 +1489,12 @@ class PublishedAllWorkflowApi(Resource):
     @login_required
     @account_initialization_required
     @rbac_permission_required(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()))
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
     @with_current_user
     @edit_permission_required
+    @with_session(write=False)
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
     @model_validate(WorkflowListQuery)
-    def get(self, args: WorkflowListQuery, current_user: Account, app_model: App):
+    def get(self, args: WorkflowListQuery, session: Session, current_user: Account, app_model: App):
         """
         Get published workflows
         """
@@ -1508,23 +1509,22 @@ class PublishedAllWorkflowApi(Resource):
                 raise Forbidden()
 
         workflow_service = WorkflowService()
-        with sessionmaker(db.engine).begin() as session:
-            workflows, has_more = workflow_service.get_all_published_workflow(
-                session=session,
-                app_model=app_model,
-                page=page,
-                limit=limit,
-                user_id=user_id,
-                named_only=named_only,
-            )
-            return WorkflowPaginationResponse.model_validate(
-                {
-                    "items": [WorkflowResponseSource(workflow, session=session) for workflow in workflows],
-                    "page": page,
-                    "limit": limit,
-                    "has_more": has_more,
-                }
-            ).model_dump(mode="json")
+        workflows, has_more = workflow_service.get_all_published_workflow(
+            session=session,
+            app_model=app_model,
+            page=page,
+            limit=limit,
+            user_id=user_id,
+            named_only=named_only,
+        )
+        return WorkflowPaginationResponse.model_validate(
+            {
+                "items": [WorkflowResponseSource(workflow, session=session) for workflow in workflows],
+                "page": page,
+                "limit": limit,
+                "has_more": has_more,
+            }
+        ).model_dump(mode="json")
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>/restore")

api/controllers/console/app/workflow_trigger.py

@@ -4,13 +4,13 @@ from datetime import datetime
 from flask_restx import Resource
 from pydantic import BaseModel, field_validator
 from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound
 
 from configs import dify_config
 from controllers.common.rbac import PlainApp, RBACCheck
 from controllers.common.schema import query_params_from_model, register_schema_models
-from extensions.ext_database import db
+from controllers.common.session import with_session
 from fields.base import ResponseModel
 from libs.helper import dump_response
 from libs.login import login_required
@@ -101,28 +101,28 @@ class WebhookTriggerApi(Resource):
     @account_initialization_required
     @console_ns.response(200, "Success", console_ns.models[WebhookTriggerResponse.__name__])
     @rbac_permission_required(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()))
+    @with_session(write=False)
     @get_app_model(mode=AppMode.WORKFLOW)
     @model_validate(Parser)
-    def get(self, req_data: Parser, app_model: App):
+    def get(self, req_data: Parser, session: Session, app_model: App):
         """Get webhook trigger for a node"""
 
         node_id = req_data.node_id
 
-        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            # Get webhook trigger for this app and node
-            webhook_trigger = session.scalar(
-                select(WorkflowWebhookTrigger)
-                .where(
-                    WorkflowWebhookTrigger.app_id == app_model.id,
-                    WorkflowWebhookTrigger.node_id == node_id,
-                )
-                .limit(1)
+        # Get webhook trigger for this app and node
+        webhook_trigger = session.scalar(
+            select(WorkflowWebhookTrigger)
+            .where(
+                WorkflowWebhookTrigger.app_id == app_model.id,
+                WorkflowWebhookTrigger.node_id == node_id,
             )
+            .limit(1)
+        )
 
-            if not webhook_trigger:
-                raise NotFound("Webhook trigger not found for this node")
+        if not webhook_trigger:
+            raise NotFound("Webhook trigger not found for this node")
 
-            return dump_response(WebhookTriggerResponse, webhook_trigger)
+        return dump_response(WebhookTriggerResponse, webhook_trigger)
 
 
 @console_ns.route("/apps/<uuid:app_id>/triggers")
@@ -135,23 +135,23 @@ class AppTriggersApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[WorkflowTriggerListResponse.__name__])
     @with_current_tenant_id
     @rbac_permission_required(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp()))
+    @with_session(write=False)
     @get_app_model(mode=AppMode.WORKFLOW)
-    def get(self, current_tenant_id: str, app_model: App):
+    def get(self, session: Session, current_tenant_id: str, app_model: App):
         """Get app triggers list"""
-        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            # Get all triggers for this app using select API
-            triggers = (
-                session.execute(
-                    select(AppTrigger)
-                    .where(
-                        AppTrigger.tenant_id == current_tenant_id,
-                        AppTrigger.app_id == app_model.id,
-                    )
-                    .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc())
+        # Get all triggers for this app using select API
+        triggers = (
+            session.execute(
+                select(AppTrigger)
+                .where(
+                    AppTrigger.tenant_id == current_tenant_id,
+                    AppTrigger.app_id == app_model.id,
                 )
-                .scalars()
-                .all()
+                .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc())
             )
+            .scalars()
+            .all()
+        )
 
         # Add computed icon field for each trigger
         url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
@@ -174,27 +174,27 @@ class AppTriggerEnableApi(Resource):
     @rbac_permission_required(RBACCheck(RBACPermission.APP_EDIT, PlainApp()))
     @console_ns.response(200, "Success", console_ns.models[WorkflowTriggerResponse.__name__])
     @with_current_tenant_id
+    @with_session
     @get_app_model(mode=AppMode.WORKFLOW)
     @model_validate(ParserEnable)
-    def post(self, req_data: ParserEnable, current_tenant_id: str, app_model: App):
+    def post(self, req_data: ParserEnable, session: Session, current_tenant_id: str, app_model: App):
         """Update app trigger (enable/disable)"""
 
         trigger_id = req_data.trigger_id
-        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            # Find the trigger using select
-            trigger = session.execute(
-                select(AppTrigger).where(
-                    AppTrigger.id == trigger_id,
-                    AppTrigger.tenant_id == current_tenant_id,
-                    AppTrigger.app_id == app_model.id,
-                )
-            ).scalar_one_or_none()
+        # Find the trigger using select
+        trigger = session.execute(
+            select(AppTrigger).where(
+                AppTrigger.id == trigger_id,
+                AppTrigger.tenant_id == current_tenant_id,
+                AppTrigger.app_id == app_model.id,
+            )
+        ).scalar_one_or_none()
 
-            if not trigger:
-                raise NotFound("Trigger not found")
+        if not trigger:
+            raise NotFound("Trigger not found")
 
-            # Update status based on enable_trigger boolean
-            trigger.status = AppTriggerStatus.ENABLED if req_data.enable_trigger else AppTriggerStatus.DISABLED
+        # Update status based on enable_trigger boolean
+        trigger.status = AppTriggerStatus.ENABLED if req_data.enable_trigger else AppTriggerStatus.DISABLED
 
         # Add computed icon field
         url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"

api/tests/unit_tests/controllers/console/app/test_app_apis.py

@@ -920,6 +920,7 @@ class TestWorkflowTriggerEndpoints:
     def test_webhook_trigger_api_get(
         self,
         database_app: Flask,
+        sqlite_session: Session,
     ) -> None:
         api = workflow_trigger_module.WebhookTriggerApi()
         method = unwrap(api.get)
@@ -930,11 +931,11 @@ class TestWorkflowTriggerEndpoints:
             webhook_id="webhook-1",
             created_by=USER_ID,
         )
-        db.session.add(trigger)
-        db.session.commit()
+        sqlite_session.add(trigger)
+        sqlite_session.commit()
 
         with database_app.test_request_context("/?node_id=node-1"):
-            result = method(api, Parser(node_id="node-1"), app_model=_make_app())
+            result = method(api, Parser(node_id="node-1"), sqlite_session, app_model=_make_app())
 
         assert isinstance(result, dict)
         assert {"id", "webhook_id", "webhook_url", "webhook_debug_url", "node_id", "created_at"} <= set(result.keys())

api/tests/unit_tests/controllers/console/app/test_workflow.py

@@ -5,13 +5,11 @@ import json
 from contextlib import contextmanager, nullcontext
 from datetime import datetime
 from types import SimpleNamespace
-from typing import cast
 from unittest.mock import Mock
 
 import pytest
 from flask import Flask
 from pydantic import ValidationError
-from sqlalchemy import Engine
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import HTTPException, NotFound
 
@@ -22,65 +20,117 @@ from core.workflow.llm_environment_variable import LLMEnvironmentVariable
 from graphon.file import File, FileTransferMethod, FileType
 from graphon.variables import SecretVariable, StringVariable
 from graphon.variables.variables import RAGPipelineVariable
+from models.account import Account
+from models.model import App, AppMode
+from models.workflow import Workflow, WorkflowType
 from tests.unit_tests.config_override import apply_config_overrides
 
 
-def _make_workflow(**overrides):
-    workflow = SimpleNamespace(
-        id="workflow-1",
-        graph_dict={"nodes": [], "edges": []},
-        features_dict={"file_upload": {"enabled": False}},
-        unique_hash="hash-1",
-        version="1",
-        marked_name="Release 1",
-        marked_comment="Initial release",
-        created_by_account=SimpleNamespace(id="user-1", name="Alice", email="alice@example.com"),
-        created_at=datetime(2024, 1, 1, 12, 0, 0),
-        updated_by_account=None,
-        updated_at=datetime(2024, 1, 1, 12, 1, 0),
-        tool_published=False,
-        environment_variables=[
-            {
-                "id": "env-1",
-                "name": "API_KEY",
-                "value": "[__HIDDEN__]",
-                "value_type": "secret",
-                "description": "API key",
-            }
-        ],
-        conversation_variables=[
-            {
-                "id": "conv-1",
-                "name": "topic",
-                "value": "hello",
-                "value_type": "string",
-                "description": "Topic",
-            }
-        ],
-        rag_pipeline_variables=[
-            {
-                "variable": "query",
-                "type": "text-input",
-                "label": "Query",
-                "belong_to_node_id": "shared",
-                "max_length": 0,
-                "required": False,
-                "unit": "",
-                "default_value": "",
-                "options": [],
-                "placeholder": "",
-                "tooltips": "",
-                "allowed_file_types": ["custom"],
-                "allowed_file_extensions": [".pdf"],
-                "allowed_file_upload_methods": ["local_file"],
-            }
+@pytest.fixture(autouse=True)
+def _identity_workflow_encryption(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep variable serialization focused on ORM behavior, not the external key provider."""
+
+    monkeypatch.setattr(workflow_module.encrypter, "encrypt_token", lambda *, tenant_id, token: token)
+    monkeypatch.setattr(workflow_module.encrypter, "decrypt_token", lambda *, tenant_id, token: token)
+
+
+def _account() -> Account:
+    account = Account(name="Alice", email="alice@example.com")
+    account.id = "user-1"
+    return account
+
+
+def _app(*, app_id: str = "app", tenant_id: str = "t1") -> App:
+    return App(
+        id=app_id,
+        tenant_id=tenant_id,
+        name="Workflow App",
+        description="",
+        mode=AppMode.WORKFLOW,
+        enable_site=True,
+        enable_api=True,
+        max_active_requests=0,
+    )
+
+
+def _make_workflow(**overrides) -> Workflow:
+    graph = overrides.pop("graph_dict", {"nodes": [], "edges": []})
+    features = overrides.pop("features_dict", {"file_upload": {"enabled": False}})
+    environment_variables = overrides.pop(
+        "environment_variables",
+        [
+            SecretVariable(
+                id="env-1",
+                name="API_KEY",
+                value="plain-token",
+                selector=["env", "API_KEY"],
+                description="API key",
+            )
         ],
     )
+    conversation_variables = overrides.pop(
+        "conversation_variables",
+        [
+            StringVariable(
+                id="conv-1",
+                name="topic",
+                value="hello",
+                selector=["conversation", "topic"],
+                description="Topic",
+            )
+        ],
+    )
+    rag_pipeline_variables = overrides.pop(
+        "rag_pipeline_variables",
+        [
+            RAGPipelineVariable.model_validate(
+                {
+                    "variable": "query",
+                    "type": "text-input",
+                    "label": "Query",
+                    "belong_to_node_id": "shared",
+                    "max_length": 0,
+                    "required": False,
+                    "unit": "",
+                    "default_value": "",
+                    "options": [],
+                    "placeholder": "",
+                    "tooltips": "",
+                    "allowed_file_types": ["custom"],
+                    "allowed_file_extensions": [".pdf"],
+                    "allowed_file_upload_methods": ["local_file"],
+                }
+            )
+        ],
+    )
+    workflow = Workflow(
+        id="workflow-1",
+        tenant_id="t1",
+        app_id="app",
+        type=WorkflowType.WORKFLOW,
+        version="1",
+        graph=json.dumps(graph),
+        features=json.dumps(features),
+        marked_name="Release 1",
+        marked_comment="Initial release",
+        created_by="user-1",
+        created_at=datetime(2024, 1, 1, 12, 0, 0),
+        updated_by=None,
+        updated_at=datetime(2024, 1, 1, 12, 1, 0),
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    if environment_variables and isinstance(environment_variables[0], dict):
+        workflow._environment_variables = json.dumps(
+            {str(index): value for index, value in enumerate(environment_variables)}
+        )
+    else:
+        workflow.environment_variables = environment_variables
+    workflow.conversation_variables = conversation_variables
+    workflow.rag_pipeline_variables = rag_pipeline_variables
     for key, value in overrides.items():
         setattr(workflow, key, value)
-    workflow.get_created_by_account = Mock(return_value=workflow.created_by_account)
-    workflow.get_updated_by_account = Mock(return_value=workflow.updated_by_account)
-    workflow.get_tool_published = Mock(return_value=workflow.tool_published)
     return workflow
 
 
@@ -180,9 +230,9 @@ def test_delete_workflow_retires_candidates_only_after_transaction_exit(
 
 def test_parse_file_no_config(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(workflow_module.FileUploadConfigManager, "convert", lambda *_args, **_kwargs: None)
-    workflow = SimpleNamespace(features_dict={}, tenant_id="t1")
+    workflow = _make_workflow(features_dict={})
 
-    assert workflow_module._parse_file(cast(workflow_module.Workflow, workflow), files=[{"id": "f"}]) == []
+    assert workflow_module._parse_file(workflow, files=[{"id": "f"}]) == []
 
 
 def test_parse_file_with_config(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -199,8 +249,8 @@ def test_parse_file_with_config(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(workflow_module.FileUploadConfigManager, "convert", lambda *_args, **_kwargs: config)
     monkeypatch.setattr(workflow_module.file_factory, "build_from_mappings", build_mock)
 
-    workflow = SimpleNamespace(features_dict={}, tenant_id="t1")
-    result = workflow_module._parse_file(cast(workflow_module.Workflow, workflow), files=[{"id": "f"}])
+    workflow = _make_workflow(features_dict={})
+    result = workflow_module._parse_file(workflow, files=[{"id": "f"}])
 
     assert result == file_list
     build_mock.assert_called_once()
@@ -212,7 +262,7 @@ def test_sync_draft_workflow_invalid_content_type(app: Flask, monkeypatch: pytes
 
     with app.test_request_context("/apps/app/workflows/draft", method="POST", data="x", content_type="text/html"):
         with pytest.raises(HTTPException) as exc:
-            handler(api, "t1", app_model=SimpleNamespace(id="app"))
+            handler(api, _account(), app_model=_app())
 
     assert exc.value.code == 415
 
@@ -227,18 +277,14 @@ def test_sync_draft_workflow_invalid_json(app: Flask, monkeypatch: pytest.Monkey
         data="[]",
         content_type="application/json",
     ):
-        response, status = handler(api, "t1", app_model=SimpleNamespace(id="app"))
+        response, status = handler(api, _account(), app_model=_app())
 
     assert status == 400
     assert response["message"] == "Invalid JSON data"
 
 
 def test_sync_draft_workflow_success(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
-    workflow = SimpleNamespace(
-        unique_hash="h",
-        updated_at=None,
-        created_at=datetime(2024, 1, 1),
-    )
+    workflow = _make_workflow(updated_at=None, created_at=datetime(2024, 1, 1))
 
     monkeypatch.setattr(
         workflow_module.variable_factory, "build_environment_variable_from_mapping", lambda *_args: "env"
@@ -259,7 +305,7 @@ def test_sync_draft_workflow_success(app: Flask, monkeypatch: pytest.MonkeyPatch
         method="POST",
         json={"graph": {}, "features": {}, "hash": "h"},
     ):
-        response = handler(api, "t1", app_model=SimpleNamespace(id="app"))
+        response = handler(api, _account(), app_model=_app())
 
     assert response["result"] == "success"
     assert sync_draft_workflow.call_args.kwargs["environment_variables"] == []
@@ -267,11 +313,7 @@ def test_sync_draft_workflow_success(app: Flask, monkeypatch: pytest.MonkeyPatch
 
 
 def test_sync_draft_workflow_passes_environment_patch(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
-    workflow = SimpleNamespace(
-        unique_hash="next-hash",
-        updated_at=None,
-        created_at=datetime(2024, 1, 1),
-    )
+    workflow = _make_workflow(updated_at=None, created_at=datetime(2024, 1, 1))
     patched_variable = StringVariable(
         id="env-model",
         name="shared_model",
@@ -313,7 +355,7 @@ def test_sync_draft_workflow_passes_environment_patch(app: Flask, monkeypatch: p
             },
         },
     ):
-        response = handler(api, "t1", app_model=SimpleNamespace(id="app"))
+        response = handler(api, _account(), app_model=_app())
 
     assert response["result"] == "success"
     assert sync_draft_workflow.call_args.kwargs["preserve_environment_variables"] is True
@@ -364,7 +406,7 @@ def test_sync_draft_workflow_hash_mismatch(app: Flask, monkeypatch: pytest.Monke
         json={"graph": {}, "features": {}, "hash": "h"},
     ):
         with pytest.raises(DraftWorkflowNotSync):
-            handler(api, "t1", app_model=SimpleNamespace(id="app"))
+            handler(api, _account(), app_model=_app())
 
 
 def test_sync_draft_workflow_variable_validation_error(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -385,18 +427,13 @@ def test_sync_draft_workflow_variable_validation_error(app: Flask, monkeypatch:
         json={"graph": {}, "features": {}, "hash": "h", "conversation_variables": [{"name": "topic"}]},
     ):
         with pytest.raises(InvalidArgumentError) as exc:
-            handler(api, "t1", app_model=SimpleNamespace(id="app"))
+            handler(api, _account(), app_model=_app())
 
     assert exc.value.description == "description too long"
 
 
 def test_restore_published_workflow_to_draft_success(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
-    workflow = SimpleNamespace(
-        unique_hash="restored-hash",
-        updated_at=None,
-        created_at=datetime(2024, 1, 1),
-    )
-    user = SimpleNamespace(id="account-1")
+    workflow = _make_workflow(updated_at=None, created_at=datetime(2024, 1, 1))
 
     monkeypatch.setattr(
         workflow_module,
@@ -413,13 +450,13 @@ def test_restore_published_workflow_to_draft_success(app: Flask, monkeypatch: py
     ):
         response = handler(
             api,
-            "t1",
-            app_model=SimpleNamespace(id="app", tenant_id="tenant-1"),
+            _account(),
+            app_model=_app(tenant_id="tenant-1"),
             workflow_id="published-workflow",
         )
 
     assert response["result"] == "success"
-    assert response["hash"] == "restored-hash"
+    assert response["hash"] == workflow.unique_hash
 
 
 def test_restore_published_workflow_to_draft_not_found(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -443,8 +480,8 @@ def test_restore_published_workflow_to_draft_not_found(app: Flask, monkeypatch:
         with pytest.raises(NotFound):
             handler(
                 api,
-                "t1",
-                app_model=SimpleNamespace(id="app", tenant_id="tenant-1"),
+                _account(),
+                app_model=_app(tenant_id="tenant-1"),
                 workflow_id="published-workflow",
             )
 
@@ -475,8 +512,8 @@ def test_restore_published_workflow_to_draft_returns_400_for_draft_source(
         with pytest.raises(HTTPException) as exc:
             handler(
                 api,
-                "t1",
-                app_model=SimpleNamespace(id="app", tenant_id="tenant-1"),
+                _account(),
+                app_model=_app(tenant_id="tenant-1"),
                 workflow_id="draft-workflow",
             )
 
@@ -507,8 +544,8 @@ def test_restore_published_workflow_to_draft_returns_400_for_invalid_structure(
         with pytest.raises(HTTPException) as exc:
             handler(
                 api,
-                "t1",
-                app_model=SimpleNamespace(id="app", tenant_id="tenant-1"),
+                _account(),
+                app_model=_app(tenant_id="tenant-1"),
                 workflow_id="published-workflow",
             )
 
@@ -516,31 +553,19 @@ def test_restore_published_workflow_to_draft_returns_400_for_invalid_structure(
     assert exc.value.description == "invalid workflow graph"
 
 
-def test_get_published_workflows_serializes_items_before_session_closes(
-    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine
+def test_get_published_workflows_uses_the_request_session(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
 ) -> None:
     api = workflow_module.PublishedAllWorkflowApi()
     handler = inspect.unwrap(api.get)
-
-    base_workflow = _make_workflow()
-
-    class _Workflow:
-        def __init__(self, session: Session) -> None:
-            self._session = session
-
-        def __getattr__(self, name):
-            return getattr(base_workflow, name)
-
-        @property
-        def id(self):
-            assert self._session.in_transaction()
-            return "w1"
+    workflow = _make_workflow()
+    sqlite_session.add_all([_account(), workflow])
+    sqlite_session.commit()
 
     def get_all_published_workflow(*, session: Session, **_kwargs):
-        assert isinstance(session, Session)
-        return [_Workflow(session)], False
+        assert session is sqlite_session
+        return [workflow], False
 
-    monkeypatch.setattr(workflow_module, "db", SimpleNamespace(engine=sqlite_engine))
     monkeypatch.setattr(
         workflow_module,
         "WorkflowService",
@@ -555,16 +580,18 @@ def test_get_published_workflows_serializes_items_before_session_closes(
         query = workflow_module.WorkflowListQuery.model_validate(
             {"page": "1", "limit": "10", "user_id": "", "named_only": "false"}
         )
-        response = handler(api, query, "t1", app_model=SimpleNamespace(id="app", workflow_id="wf-1"))
+        response = handler(api, query, sqlite_session, _account(), app_model=_app())
 
-    assert response["items"][0]["id"] == "w1"
+    assert response["items"][0]["id"] == "workflow-1"
     assert response["page"] == 1
     assert response["limit"] == 10
     assert response["has_more"] is False
 
 
-def test_draft_workflow_get_serializes_response_model(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_draft_workflow_get_serializes_response_model(monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None:
     workflow = _make_workflow()
+    sqlite_session.add_all([_account(), workflow])
+    sqlite_session.commit()
     monkeypatch.setattr(
         workflow_module, "WorkflowService", lambda: SimpleNamespace(get_draft_workflow=lambda **_kwargs: workflow)
     )
@@ -572,12 +599,12 @@ def test_draft_workflow_get_serializes_response_model(monkeypatch: pytest.Monkey
     api = workflow_module.DraftWorkflowApi()
     handler = inspect.unwrap(api.get)
 
-    response = handler(api, app_model=SimpleNamespace(id="app"))
+    response = handler(api, sqlite_session, app_model=_app())
 
     assert response["id"] == "workflow-1"
     assert response["graph"] == {"nodes": [], "edges": []}
     assert response["features"] == {"file_upload": {"enabled": False}}
-    assert response["hash"] == "hash-1"
+    assert response["hash"] == workflow.unique_hash
     assert response["created_by"] == {"id": "user-1", "name": "Alice", "email": "alice@example.com"}
     assert response["updated_by"] is None
     assert response["created_at"] == int(datetime(2024, 1, 1, 12, 0, 0).timestamp())
@@ -586,7 +613,7 @@ def test_draft_workflow_get_serializes_response_model(monkeypatch: pytest.Monkey
         {
             "id": "env-1",
             "name": "API_KEY",
-            "value": "[__HIDDEN__]",
+            "value": workflow_module.encrypter.full_mask_token(),
             "value_type": "secret",
             "description": "API key",
         }
@@ -620,10 +647,13 @@ def test_draft_workflow_get_serializes_response_model(monkeypatch: pytest.Monkey
     ]
 
 
-def test_published_workflow_get_uses_session_aware_response_source(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_published_workflow_get_uses_session_aware_response_source(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
+    sqlite_session.add(_account())
+    sqlite_session.commit()
     workflow = _make_workflow()
-    session = Mock(spec=Session)
-    monkeypatch.setattr(workflow_module, "db", SimpleNamespace(session=Mock(return_value=session)))
+    monkeypatch.setattr(workflow_module.db, "session", lambda: sqlite_session)
     monkeypatch.setattr(
         workflow_module, "WorkflowService", lambda: SimpleNamespace(get_published_workflow=lambda **_kwargs: workflow)
     )
@@ -634,9 +664,9 @@ def test_published_workflow_get_uses_session_aware_response_source(monkeypatch:
     response = handler(api, app_model=SimpleNamespace(id="app"))
 
     assert response["id"] == "workflow-1"
-    workflow.get_created_by_account.assert_called_once_with(session=session)
-    workflow.get_updated_by_account.assert_called_once_with(session=session)
-    workflow.get_tool_published.assert_called_once_with(session=session)
+    assert response["created_by"] == {"id": "user-1", "name": "Alice", "email": "alice@example.com"}
+    assert response["updated_by"] is None
+    assert response["tool_published"] is False
 
 
 def test_pipeline_variable_response_accepts_legacy_file_field_names() -> None:
@@ -690,15 +720,20 @@ def test_pipeline_variable_response_accepts_explicit_null_optional_fields() -> N
     assert response["allowed_file_upload_methods"] is None
 
 
-def test_workflow_response_masks_secret_environment_variables() -> None:
+def test_workflow_response_masks_secret_environment_variables(sqlite_session: Session) -> None:
     workflow = _make_workflow(
         environment_variables=[
             SecretVariable(id="env-secret", name="API_KEY", value="plain-token", selector=["env", "API_KEY"]),
             StringVariable(id="env-string", name="REGION", value="us-east-1", selector=["env", "REGION"]),
         ]
     )
+    sqlite_session.add_all([_account(), workflow])
+    sqlite_session.commit()
 
-    response = workflow_module.WorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
+    response = workflow_module.WorkflowResponse.model_validate(
+        workflow_module.WorkflowResponseSource(workflow, session=sqlite_session),
+        from_attributes=True,
+    ).model_dump(mode="json")
 
     assert response["environment_variables"] == [
         {
@@ -718,7 +753,7 @@ def test_workflow_response_masks_secret_environment_variables() -> None:
     ]
 
 
-def test_workflow_response_preserves_llm_environment_variable_type() -> None:
+def test_workflow_response_preserves_llm_environment_variable_type(sqlite_session: Session) -> None:
     workflow = _make_workflow(
         environment_variables=[
             LLMEnvironmentVariable(
@@ -729,8 +764,13 @@ def test_workflow_response_preserves_llm_environment_variable_type() -> None:
             )
         ]
     )
+    sqlite_session.add_all([_account(), workflow])
+    sqlite_session.commit()
 
-    response = workflow_module.WorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
+    response = workflow_module.WorkflowResponse.model_validate(
+        workflow_module.WorkflowResponseSource(workflow, session=sqlite_session),
+        from_attributes=True,
+    ).model_dump(mode="json")
 
     assert response["environment_variables"] == [
         {
@@ -743,14 +783,19 @@ def test_workflow_response_preserves_llm_environment_variable_type() -> None:
     ]
 
 
-def test_workflow_response_rejects_invalid_environment_variable_dict() -> None:
+def test_workflow_response_rejects_invalid_environment_variable_dict(sqlite_session: Session) -> None:
     workflow = _make_workflow(environment_variables=[{"value_type": "not-a-segment-type"}])
+    sqlite_session.add_all([_account(), workflow])
+    sqlite_session.commit()
 
     with pytest.raises(ValidationError):
-        workflow_module.WorkflowResponse.model_validate(workflow, from_attributes=True)
+        workflow_module.WorkflowResponse.model_validate(
+            workflow_module.WorkflowResponseSource(workflow, session=sqlite_session),
+            from_attributes=True,
+        )
 
 
-def test_draft_workflow_get_not_found(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_draft_workflow_get_not_found(monkeypatch: pytest.MonkeyPatch, unbound_session: Session) -> None:
     monkeypatch.setattr(
         workflow_module, "WorkflowService", lambda: SimpleNamespace(get_draft_workflow=lambda **_k: None)
     )
@@ -759,10 +804,12 @@ def test_draft_workflow_get_not_found(monkeypatch: pytest.MonkeyPatch) -> None:
     handler = inspect.unwrap(api.get)
 
     with pytest.raises(DraftWorkflowNotExist):
-        handler(api, app_model=SimpleNamespace(id="app"))
+        handler(api, unbound_session, app_model=_app())
 
 
-def test_draft_workflow_get_projects_agent_node_job_to_graph(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_draft_workflow_get_projects_agent_node_job_to_graph(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
     workflow = _make_workflow(
         graph_dict={
             "nodes": [
@@ -778,6 +825,8 @@ def test_draft_workflow_get_projects_agent_node_job_to_graph(monkeypatch: pytest
             "edges": [],
         }
     )
+    sqlite_session.add_all([_account(), workflow])
+    sqlite_session.commit()
     projected_graph = {
         "nodes": [
             {
@@ -811,12 +860,14 @@ def test_draft_workflow_get_projects_agent_node_job_to_graph(monkeypatch: pytest
     api = workflow_module.DraftWorkflowApi()
     handler = inspect.unwrap(api.get)
 
-    response = handler(api, app_model=SimpleNamespace(id="app"))
+    response = handler(api, sqlite_session, app_model=_app())
 
     assert response["graph"] == projected_graph
 
 
-def test_advanced_chat_run_conversation_not_exists(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_advanced_chat_run_conversation_not_exists(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, unbound_session: Session
+) -> None:
     monkeypatch.setattr(
         workflow_module.AppGenerateService,
         "generate",
@@ -835,7 +886,7 @@ def test_advanced_chat_run_conversation_not_exists(app: Flask, monkeypatch: pyte
     ):
         payload = workflow_module.AdvancedChatWorkflowRunPayload.model_validate({"inputs": {}})
         with pytest.raises(NotFound):
-            handler(api, payload, Mock(), "t1", app_model=SimpleNamespace(id="app"))
+            handler(api, payload, unbound_session, _account(), app_model=_app())
 
 
 @pytest.mark.parametrize(
@@ -868,7 +919,7 @@ def test_trigger_run_loads_draft_with_request_session(
         lambda: SimpleNamespace(get_draft_workflow=get_draft_workflow),
     )
     session = unbound_session
-    app_model = SimpleNamespace(id="app-1")
+    app_model = _app(app_id="app-1")
     handler = inspect.unwrap(resource.post)
 
     with app.test_request_context("/", method="POST", json=payload):
@@ -877,7 +928,7 @@ def test_trigger_run_loads_draft_with_request_session(
                 resource(),
                 payload_model.model_validate(payload),
                 session,
-                SimpleNamespace(id="account-1"),
+                _account(),
                 app_model,
             )
 

api/tests/unit_tests/controllers/console/app/test_workflow_trigger_api.py

@@ -3,32 +3,18 @@ from __future__ import annotations
 import inspect
 import uuid
 from datetime import UTC, datetime
-from types import SimpleNamespace
 from unittest.mock import PropertyMock, patch
 
-import pytest
 from flask import Flask
-from sqlalchemy import Engine
 from sqlalchemy.orm import Session
 
 from controllers.console import console_ns
 from controllers.console.app import workflow_trigger as workflow_trigger_module
-from models.base import TypeBase
 from models.enums import AppTriggerStatus, AppTriggerType
 from models.model import App, AppMode, IconType
 from models.trigger import AppTrigger
 
 
-@pytest.fixture
-def database_session(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch):
-    models = (App, AppTrigger)
-    tables = [model.metadata.tables[model.__tablename__] for model in models]
-    TypeBase.metadata.create_all(sqlite_engine, tables=tables)
-    monkeypatch.setattr(workflow_trigger_module, "db", SimpleNamespace(engine=sqlite_engine))
-    with Session(sqlite_engine, expire_on_commit=False) as session:
-        yield session
-
-
 def _persist_app_trigger(
     session: Session,
     *,
@@ -107,8 +93,8 @@ def test_webhook_trigger_response_serializes_datetime():
     assert payload["created_at"] == "2026-01-02T03:04:05Z"
 
 
-def test_app_triggers_get_uses_injected_tenant_id(app: Flask, database_session: Session) -> None:
-    app_model, trigger = _persist_app_trigger(database_session)
+def test_app_triggers_get_uses_injected_tenant_id(app: Flask, sqlite_session: Session) -> None:
+    app_model, trigger = _persist_app_trigger(sqlite_session)
     other_tenant_trigger = AppTrigger(
         tenant_id=str(uuid.uuid4()),
         app_id=app_model.id,
@@ -118,21 +104,21 @@ def test_app_triggers_get_uses_injected_tenant_id(app: Flask, database_session:
         provider_name="provider",
         status=AppTriggerStatus.ENABLED,
     )
-    database_session.add(other_tenant_trigger)
-    database_session.commit()
+    sqlite_session.add(other_tenant_trigger)
+    sqlite_session.commit()
 
     api = workflow_trigger_module.AppTriggersApi()
     method = inspect.unwrap(api.get)
 
     with app.test_request_context("/"):
-        response = method(api, app_model.tenant_id, app_model)
+        response = method(api, sqlite_session, app_model.tenant_id, app_model)
 
     assert [item["id"] for item in response["data"]] == [trigger.id]
     assert response["data"][0]["icon"].endswith("/provider/icon")
 
 
-def test_app_trigger_enable_uses_injected_tenant_id(app: Flask, database_session: Session) -> None:
-    app_model, trigger = _persist_app_trigger(database_session, status=AppTriggerStatus.DISABLED)
+def test_app_trigger_enable_uses_injected_tenant_id(app: Flask, sqlite_session: Session) -> None:
+    app_model, trigger = _persist_app_trigger(sqlite_session, status=AppTriggerStatus.DISABLED)
     payload = {"trigger_id": trigger.id, "enable_trigger": True}
 
     api = workflow_trigger_module.AppTriggerEnableApi()
@@ -145,13 +131,15 @@ def test_app_trigger_enable_uses_injected_tenant_id(app: Flask, database_session
         response = method(
             api,
             workflow_trigger_module.ParserEnable(trigger_id=trigger.id, enable_trigger=True),
+            sqlite_session,
             app_model.tenant_id,
             app_model,
         )
 
     assert response["id"] == trigger.id
     assert response["status"] == "enabled"
-    database_session.expire_all()
-    persisted_trigger = database_session.get(AppTrigger, trigger.id)
+    sqlite_session.flush()
+    sqlite_session.expire_all()
+    persisted_trigger = sqlite_session.get(AppTrigger, trigger.id)
     assert persisted_trigger is not None
     assert persisted_trigger.status == AppTriggerStatus.ENABLED