test: use SQLite sessions in core app apps (#39083)

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

api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py

@@ -1,10 +1,133 @@
+"""SQLite-backed tests for workflow app generation and worker reload behavior."""
+
 import contextlib
+import json
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
-from pytest_mock import MockerFixture
+import pytest
+from flask import Flask
+from sqlalchemy.orm import Session, sessionmaker
 
+import core.app.apps.workflow.app_generator as app_generator_module
+from core.app.app_config.entities import WorkflowUIBasedAppConfig
+from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY, WorkflowAppGenerator
+from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
+from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
+from core.ops.ops_trace_manager import TraceQueueManager
+from core.repositories import SQLAlchemyWorkflowExecutionRepository, SQLAlchemyWorkflowNodeExecutionRepository
+from graphon.enums import WorkflowExecutionStatus
+from graphon.runtime import GraphRuntimeState, VariablePool
+from models.enums import CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom
+from models.model import App, AppMode, EndUser
+from models.snippet import CustomizedSnippet
+from models.workflow import Workflow, WorkflowKind, WorkflowNodeExecutionTriggeredFrom, WorkflowRun, WorkflowType
+
+
+def _workflow(
+    *,
+    workflow_id: str = "workflow",
+    app_id: str = "app",
+    tenant_id: str = "tenant",
+    kind: WorkflowKind = WorkflowKind.STANDARD,
+) -> Workflow:
+    return Workflow(
+        id=workflow_id,
+        tenant_id=tenant_id,
+        app_id=app_id,
+        type=WorkflowType.WORKFLOW,
+        kind=kind,
+        version="1",
+        graph=json.dumps({"nodes": [], "edges": []}),
+        features="{}",
+        created_by="creator",
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+
+
+def _persist_generator_rows(sqlite_session: Session) -> tuple[App, Workflow, EndUser]:
+    workflow = _workflow()
+    app = App(
+        id="app",
+        tenant_id="tenant",
+        name="Workflow app",
+        description="",
+        mode=AppMode.WORKFLOW,
+        workflow_id=workflow.id,
+        enable_site=True,
+        enable_api=True,
+        max_active_requests=0,
+    )
+    end_user = EndUser(
+        id="user",
+        tenant_id="tenant",
+        app_id=app.id,
+        type=EndUserType.SERVICE_API,
+        name="End user",
+        session_id="end-user-session",
+    )
+    sqlite_session.add_all([workflow, app, end_user])
+    sqlite_session.commit()
+    return app, workflow, end_user
+
+
+def _app_config(app: App, workflow: Workflow) -> WorkflowUIBasedAppConfig:
+    return WorkflowUIBasedAppConfig(
+        tenant_id=app.tenant_id,
+        app_id=app.id,
+        app_mode=app.mode,
+        workflow_id=workflow.id,
+    )
+
+
+def _generate_entity(
+    app: App,
+    workflow: Workflow,
+    end_user: EndUser,
+    *,
+    invoke_from: InvokeFrom = InvokeFrom.SERVICE_API,
+    stream: bool = True,
+) -> WorkflowAppGenerateEntity:
+    return WorkflowAppGenerateEntity(
+        task_id="task",
+        app_config=_app_config(app, workflow),
+        inputs={},
+        files=[],
+        user_id=end_user.id,
+        stream=stream,
+        invoke_from=invoke_from,
+        trace_manager=MagicMock(spec=TraceQueueManager),
+        workflow_execution_id="run",
+    )
+
+
+def _runtime_state() -> GraphRuntimeState:
+    return GraphRuntimeState(variable_pool=VariablePool(), start_at=0.0)
+
+
+def _repositories(
+    sqlite_session: Session, app: App, end_user: EndUser
+) -> tuple[SQLAlchemyWorkflowExecutionRepository, SQLAlchemyWorkflowNodeExecutionRepository]:
+    session_factory = sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False)
+    return (
+        SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            tenant_id=app.tenant_id,
+            user=end_user,
+            app_id=app.id,
+            triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        ),
+        SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=session_factory,
+            tenant_id=app.tenant_id,
+            user=end_user,
+            app_id=app.id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        ),
+    )
 
 
 def test_should_prepare_user_inputs_defaults_to_true():
@@ -25,47 +148,67 @@ def test_should_prepare_user_inputs_keeps_validation_when_flag_false():
     assert WorkflowAppGenerator()._should_prepare_user_inputs(args)
 
 
-def test_ensure_snippet_start_node_in_worker_returns_standard_workflow_without_lookup():
-    session = MagicMock()
-    workflow = SimpleNamespace(kind_or_standard="standard")
+def test_ensure_snippet_start_node_in_worker_returns_standard_workflow_without_lookup(
+    sqlite_session: Session,
+) -> None:
+    workflow = _workflow()
+    sqlite_session.add(workflow)
+    sqlite_session.commit()
 
-    result = WorkflowAppGenerator._ensure_snippet_start_node_in_worker(session=session, workflow=workflow)
+    result = WorkflowAppGenerator._ensure_snippet_start_node_in_worker(session=sqlite_session, workflow=workflow)
 
     assert result is workflow
-    session.scalar.assert_not_called()
 
 
-def test_ensure_snippet_start_node_in_worker_returns_snippet_workflow_when_snippet_missing():
-    session = MagicMock()
-    session.scalar.return_value = None
-    workflow = SimpleNamespace(kind_or_standard="snippet", app_id="snippet-1", tenant_id="tenant-1")
+def test_ensure_snippet_start_node_in_worker_returns_snippet_workflow_when_snippet_missing(
+    sqlite_session: Session,
+) -> None:
+    workflow = _workflow(app_id="snippet-1", tenant_id="tenant-1", kind=WorkflowKind.SNIPPET)
+    other_tenant_snippet = CustomizedSnippet(
+        id="snippet-1",
+        tenant_id="tenant-2",
+        name="Other tenant snippet",
+        description="",
+        type="node",
+    )
+    sqlite_session.add_all([workflow, other_tenant_snippet])
+    sqlite_session.commit()
 
-    result = WorkflowAppGenerator._ensure_snippet_start_node_in_worker(session=session, workflow=workflow)
+    result = WorkflowAppGenerator._ensure_snippet_start_node_in_worker(session=sqlite_session, workflow=workflow)
 
     assert result is workflow
-    session.scalar.assert_called_once()
 
 
-def test_ensure_snippet_start_node_in_worker_applies_snippet_start_injection(mocker: MockerFixture):
-    session = MagicMock()
-    snippet = SimpleNamespace(id="snippet-1", tenant_id="tenant-1")
-    session.scalar.return_value = snippet
-    workflow = SimpleNamespace(kind_or_standard="snippet", app_id="snippet-1", tenant_id="tenant-1")
-    updated_workflow = SimpleNamespace(name="updated-workflow")
-    ensure_start_node = mocker.patch(
+def test_ensure_snippet_start_node_in_worker_applies_snippet_start_injection(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
+    workflow = _workflow(app_id="snippet-1", tenant_id="tenant-1", kind=WorkflowKind.SNIPPET)
+    snippet = CustomizedSnippet(
+        id="snippet-1",
+        tenant_id="tenant-1",
+        name="Matching snippet",
+        description="",
+        type="node",
+    )
+    sqlite_session.add_all([workflow, snippet])
+    sqlite_session.commit()
+    ensure_start_node = MagicMock(return_value=workflow)
+    monkeypatch.setattr(
         "services.snippet_generate_service.SnippetGenerateService.ensure_start_node_for_worker",
-        return_value=updated_workflow,
+        ensure_start_node,
     )
 
-    result = WorkflowAppGenerator._ensure_snippet_start_node_in_worker(session=session, workflow=workflow)
+    result = WorkflowAppGenerator._ensure_snippet_start_node_in_worker(session=sqlite_session, workflow=workflow)
 
-    assert result is updated_workflow
-    session.scalar.assert_called_once()
+    assert result is workflow
     ensure_start_node.assert_called_once_with(workflow, snippet)
 
 
-def test_generate_includes_parent_trace_context_in_extras(monkeypatch):
+def test_generate_includes_parent_trace_context_in_extras(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
     generator = WorkflowAppGenerator()
+    app, workflow, end_user = _persist_generator_rows(sqlite_session)
 
     monkeypatch.setattr(
         "core.app.apps.workflow.app_generator.WorkflowAppGenerator._bind_file_access_scope",
@@ -73,46 +216,52 @@ def test_generate_includes_parent_trace_context_in_extras(monkeypatch):
     )
     monkeypatch.setattr(
         "core.app.apps.workflow.app_generator.WorkflowAppConfigManager.get_app_config",
-        lambda *args, **kwargs: SimpleNamespace(
-            app_id="app-1", tenant_id="tenant-1", workflow_id="workflow-1", variables=[]
-        ),
+        lambda *args, **kwargs: _app_config(app, workflow),
     )
     monkeypatch.setattr(
         "core.app.apps.workflow.app_generator.file_factory.build_from_mappings", lambda *args, **kwargs: []
     )
-    monkeypatch.setattr("core.app.apps.workflow.app_generator.TraceQueueManager", MagicMock())
-    workflow_execution_factory = MagicMock(return_value=MagicMock())
-    workflow_node_execution_factory = MagicMock(return_value=MagicMock())
+    monkeypatch.setattr(
+        "core.app.apps.workflow.app_generator.TraceQueueManager",
+        MagicMock(return_value=MagicMock(spec=TraceQueueManager)),
+    )
+    repository_tenant_ids: dict[str, str] = {}
+    workflow_execution_factory = app_generator_module.DifyCoreRepositoryFactory.create_workflow_execution_repository
+    workflow_node_execution_factory = (
+        app_generator_module.DifyCoreRepositoryFactory.create_workflow_node_execution_repository
+    )
+
+    def create_workflow_execution_repository(**kwargs):
+        repository_tenant_ids["workflow"] = kwargs["tenant_id"]
+        return workflow_execution_factory(**kwargs)
+
+    def create_workflow_node_execution_repository(**kwargs):
+        repository_tenant_ids["node"] = kwargs["tenant_id"]
+        return workflow_node_execution_factory(**kwargs)
+
     monkeypatch.setattr(
         "core.app.apps.workflow.app_generator.DifyCoreRepositoryFactory.create_workflow_execution_repository",
-        workflow_execution_factory,
+        create_workflow_execution_repository,
     )
     monkeypatch.setattr(
         "core.app.apps.workflow.app_generator.DifyCoreRepositoryFactory.create_workflow_node_execution_repository",
-        workflow_node_execution_factory,
+        create_workflow_node_execution_repository,
     )
-    monkeypatch.setattr("core.app.apps.workflow.app_generator.db", SimpleNamespace(engine=MagicMock()))
+    monkeypatch.setattr("core.app.apps.workflow.app_generator.db", SimpleNamespace(engine=sqlite_session.get_bind()))
     monkeypatch.setattr(generator, "_prepare_user_inputs", lambda *, user_inputs, **kwargs: user_inputs)
 
     captured = {}
 
-    def fake_workflow_app_generate_entity(**kwargs):
-        captured["workflow_app_generate_entity_kwargs"] = kwargs
-        return SimpleNamespace(**kwargs)
-
     def fake_generate(**kwargs):
-        captured["application_generate_entity"] = kwargs["application_generate_entity"]
+        captured.update(kwargs)
         return {"data": {}}
 
-    monkeypatch.setattr(
-        "core.app.apps.workflow.app_generator.WorkflowAppGenerateEntity", fake_workflow_app_generate_entity
-    )
     monkeypatch.setattr(generator, "_generate", fake_generate)
 
     result = generator.generate(
-        app_model=SimpleNamespace(tenant_id="tenant-1", id="app-1"),
-        workflow=SimpleNamespace(features_dict={}),
-        user=SimpleNamespace(id="user-1", session_id="session-1"),
+        app_model=app,
+        workflow=workflow,
+        user=end_user,
         args={
             "inputs": {"query": "hello"},
             "files": [],
@@ -123,42 +272,56 @@ def test_generate_includes_parent_trace_context_in_extras(monkeypatch):
             },
             "trace_session_id": "session-1",
         },
-        invoke_from="service-api",
+        invoke_from=InvokeFrom.SERVICE_API,
         streaming=False,
         call_depth=0,
     )
 
     assert result == {"data": {}}
-    extras = captured["workflow_app_generate_entity_kwargs"]["extras"]
+    application_generate_entity = captured["application_generate_entity"]
+    assert isinstance(application_generate_entity, WorkflowAppGenerateEntity)
+    extras = application_generate_entity.extras
     assert extras["external_trace_id"] == "trace-1"
     assert extras["parent_trace_context"].model_dump() == {
         "parent_workflow_run_id": "outer-workflow-run-1",
         "parent_node_execution_id": "outer-node-execution-1",
     }
     assert extras["trace_session_id"] == "session-1"
-    assert workflow_execution_factory.call_args.kwargs["tenant_id"] == "tenant-1"
-    assert workflow_node_execution_factory.call_args.kwargs["tenant_id"] == "tenant-1"
+    assert isinstance(captured["workflow_execution_repository"], SQLAlchemyWorkflowExecutionRepository)
+    assert isinstance(captured["workflow_node_execution_repository"], SQLAlchemyWorkflowNodeExecutionRepository)
+    assert repository_tenant_ids == {"workflow": app.tenant_id, "node": app.tenant_id}
 
 
-def test_resume_delegates_to_generate(mocker: MockerFixture):
+def test_resume_delegates_to_generate(monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None:
     generator = WorkflowAppGenerator()
-    mock_generate = mocker.patch.object(generator, "_generate", return_value="ok")
+    app, workflow, end_user = _persist_generator_rows(sqlite_session)
+    mock_generate = MagicMock(return_value="ok")
+    monkeypatch.setattr(generator, "_generate", mock_generate)
 
-    application_generate_entity = SimpleNamespace(stream=False, invoke_from="debugger", trace_manager=MagicMock())
-    runtime_state = MagicMock(name="runtime-state")
-    pause_config = MagicMock(name="pause-config")
+    application_generate_entity = _generate_entity(
+        app,
+        workflow,
+        end_user,
+        invoke_from=InvokeFrom.DEBUGGER,
+        stream=False,
+    )
+    runtime_state = _runtime_state()
+    workflow_execution_repository, workflow_node_execution_repository = _repositories(sqlite_session, app, end_user)
+    pause_config = PauseStateLayerConfig(
+        session_factory=sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False),
+        state_owner_user_id="owner",
+    )
 
     result = generator.resume(
-        app_model=MagicMock(),
-        workflow=MagicMock(),
-        user=MagicMock(),
+        app_model=app,
+        workflow=workflow,
+        user=end_user,
         application_generate_entity=application_generate_entity,
         graph_runtime_state=runtime_state,
-        workflow_execution_repository=MagicMock(),
-        workflow_node_execution_repository=MagicMock(),
+        workflow_execution_repository=workflow_execution_repository,
+        workflow_node_execution_repository=workflow_node_execution_repository,
         graph_engine_layers=("layer",),
         pause_state_config=pause_config,
-        variable_loader=MagicMock(),
     )
 
     assert result == "ok"
@@ -167,39 +330,45 @@ def test_resume_delegates_to_generate(mocker: MockerFixture):
     assert kwargs["graph_runtime_state"] is runtime_state
     assert kwargs["pause_state_config"] is pause_config
     assert kwargs["streaming"] is False
-    assert kwargs["invoke_from"] == "debugger"
+    assert kwargs["invoke_from"] == InvokeFrom.DEBUGGER
 
 
-def test_generate_appends_pause_layer_and_forwards_state(mocker: MockerFixture):
+def test_generate_appends_pause_layer_and_forwards_state(
+    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+) -> None:
     generator = WorkflowAppGenerator()
+    app, workflow, end_user = _persist_generator_rows(sqlite_session)
 
-    mock_queue_manager = MagicMock()
-    mocker.patch("core.app.apps.workflow.app_generator.WorkflowAppQueueManager", return_value=mock_queue_manager)
-
-    fake_current_app = MagicMock()
-    fake_current_app._get_current_object.return_value = MagicMock()
-    mocker.patch("core.app.apps.workflow.app_generator.current_app", fake_current_app)
-
-    mocker.patch(
-        "core.app.apps.workflow.app_generator.WorkflowAppGenerateResponseConverter.convert",
-        return_value="converted",
+    queue_manager = MagicMock()
+    monkeypatch.setattr(
+        "core.app.apps.workflow.app_generator.WorkflowAppQueueManager",
+        MagicMock(return_value=queue_manager),
     )
-    mocker.patch.object(WorkflowAppGenerator, "_handle_response", return_value="response")
-    draft_saver_factory = mocker.patch.object(
+
+    monkeypatch.setattr(
+        "core.app.apps.workflow.app_generator.WorkflowAppGenerateResponseConverter.convert",
+        MagicMock(return_value="converted"),
+    )
+    monkeypatch.setattr(WorkflowAppGenerator, "_handle_response", MagicMock(return_value="response"))
+
+    draft_factory_tenant_ids: list[str] = []
+    get_draft_var_saver_factory = WorkflowAppGenerator._get_draft_var_saver_factory
+
+    def get_recording_draft_var_saver_factory(
+        invoke_from: InvokeFrom, account: EndUser, *, tenant_id: str
+    ) -> DraftVariableSaverFactory:
+        draft_factory_tenant_ids.append(tenant_id)
+        return get_draft_var_saver_factory(invoke_from, account, tenant_id=tenant_id)
+
+    monkeypatch.setattr(
         WorkflowAppGenerator,
         "_get_draft_var_saver_factory",
-        return_value=MagicMock(),
+        staticmethod(get_recording_draft_var_saver_factory),
     )
 
-    pause_layer = MagicMock(name="pause-layer")
-    mocker.patch(
-        "core.app.apps.workflow.app_generator.PauseStatePersistenceLayer",
-        return_value=pause_layer,
-    )
-
-    dummy_session = MagicMock()
-    dummy_session.close = MagicMock()
-    mocker.patch("core.app.apps.workflow.app_generator.db.session", dummy_session)
+    engine = sqlite_session.get_bind()
+    scoped_session = Session(engine, expire_on_commit=False)
+    monkeypatch.setattr(app_generator_module.db, "session", scoped_session)
 
     worker_kwargs: dict[str, object] = {}
 
@@ -218,84 +387,97 @@ def test_generate_appends_pause_layer_and_forwards_state(mocker: MockerFixture):
         def is_alive(self):
             return False
 
-    mocker.patch("core.app.apps.workflow.app_generator.threading.Thread", DummyThread)
+    monkeypatch.setattr("core.app.apps.workflow.app_generator.threading.Thread", DummyThread)
 
-    app_model = SimpleNamespace(mode="workflow", tenant_id="tenant")
-    app_config = SimpleNamespace(app_id="app", tenant_id="tenant", workflow_id="wf")
-    application_generate_entity = SimpleNamespace(
-        task_id="task",
-        user_id="user",
-        invoke_from="service-api",
-        app_config=app_config,
-        files=[],
-        stream=True,
-        workflow_execution_id="run",
-    )
+    application_generate_entity = _generate_entity(app, workflow, end_user)
+    graph_runtime_state = _runtime_state()
+    workflow_execution_repository, workflow_node_execution_repository = _repositories(sqlite_session, app, end_user)
 
-    graph_runtime_state = MagicMock()
-
-    result = generator._generate(
-        app_model=app_model,
-        workflow=MagicMock(),
-        user=MagicMock(),
-        application_generate_entity=application_generate_entity,
-        invoke_from="service-api",
-        workflow_execution_repository=MagicMock(),
-        workflow_node_execution_repository=MagicMock(),
-        streaming=True,
-        graph_engine_layers=("base-layer",),
-        graph_runtime_state=graph_runtime_state,
-        pause_state_config=SimpleNamespace(session_factory=MagicMock(), state_owner_user_id="owner"),
-    )
+    flask_app = Flask(__name__)
+    with flask_app.app_context():
+        result = generator._generate(
+            app_model=app,
+            workflow=workflow,
+            user=end_user,
+            application_generate_entity=application_generate_entity,
+            invoke_from=InvokeFrom.SERVICE_API,
+            workflow_execution_repository=workflow_execution_repository,
+            workflow_node_execution_repository=workflow_node_execution_repository,
+            streaming=True,
+            graph_engine_layers=("base-layer",),
+            graph_runtime_state=graph_runtime_state,
+            pause_state_config=PauseStateLayerConfig(
+                session_factory=sessionmaker(bind=engine, expire_on_commit=False),
+                state_owner_user_id="owner",
+            ),
+        )
 
     assert result == "converted"
-    assert worker_kwargs["kwargs"]["graph_engine_layers"] == ("base-layer", pause_layer)
+    graph_engine_layers = worker_kwargs["kwargs"]["graph_engine_layers"]
+    assert graph_engine_layers[0] == "base-layer"
+    assert isinstance(graph_engine_layers[1], PauseStatePersistenceLayer)
     assert worker_kwargs["kwargs"]["graph_runtime_state"] is graph_runtime_state
     assert worker_kwargs["joined"] is True
     assert worker_kwargs["join_timeout"] == 300
-    assert draft_saver_factory.call_args.kwargs["tenant_id"] == app_model.tenant_id
+    assert draft_factory_tenant_ids == [app.tenant_id]
 
 
-def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
+def test_resume_path_runs_worker_with_runtime_state(monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None:
     generator = WorkflowAppGenerator()
-    runtime_state = MagicMock(name="runtime-state")
-
-    pause_layer = MagicMock(name="pause-layer")
-    mocker.patch("core.app.apps.workflow.app_generator.PauseStatePersistenceLayer", return_value=pause_layer)
+    app, workflow, end_user = _persist_generator_rows(sqlite_session)
+    workflow_run = WorkflowRun(
+        id="run",
+        tenant_id=workflow.tenant_id,
+        app_id=app.id,
+        workflow_id=workflow.id,
+        type=workflow.type,
+        triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        version=workflow.version,
+        graph=workflow.graph,
+        inputs="{}",
+        status=WorkflowExecutionStatus.RUNNING,
+        created_by_role=CreatorUserRole.END_USER,
+        created_by=end_user.id,
+    )
+    sqlite_session.add(workflow_run)
+    sqlite_session.commit()
+    runtime_state = _runtime_state()
 
     queue_manager = MagicMock()
-    mocker.patch("core.app.apps.workflow.app_generator.WorkflowAppQueueManager", return_value=queue_manager)
+    monkeypatch.setattr(
+        "core.app.apps.workflow.app_generator.WorkflowAppQueueManager",
+        MagicMock(return_value=queue_manager),
+    )
 
-    mocker.patch.object(generator, "_handle_response", return_value="raw-response")
-    mocker.patch(
+    monkeypatch.setattr(generator, "_handle_response", MagicMock(return_value="raw-response"))
+    monkeypatch.setattr(
         "core.app.apps.workflow.app_generator.WorkflowAppGenerateResponseConverter.convert",
-        side_effect=lambda response, invoke_from: response,
+        MagicMock(side_effect=lambda response, invoke_from: response),
     )
 
-    fake_db = SimpleNamespace(session=MagicMock(), engine=MagicMock())
-    mocker.patch("core.app.apps.workflow.app_generator.db", fake_db)
-
-    workflow = SimpleNamespace(
-        id="workflow", tenant_id="tenant", app_id="app", graph_dict={}, type="workflow", version="1"
+    engine = sqlite_session.get_bind()
+    monkeypatch.setattr(app_generator_module.db, "session", Session(engine, expire_on_commit=False))
+    monkeypatch.setattr(
+        app_generator_module.session_factory,
+        "create_session",
+        lambda: Session(engine, expire_on_commit=False),
     )
-    end_user = SimpleNamespace(session_id="end-user-session")
-    app_record = SimpleNamespace(id="app")
-
-    session = MagicMock()
-    session.__enter__.return_value = session
-    session.__exit__.return_value = False
-    session.scalar.side_effect = [workflow, end_user, app_record]
-    mocker.patch("core.app.apps.workflow.app_generator.session_factory", return_value=session)
 
     runner_instance = MagicMock()
 
     def runner_ctor(**kwargs):
         assert kwargs["graph_runtime_state"] is runtime_state
+        assert kwargs["workflow"].id == workflow.id
+        assert kwargs["system_user_id"] == end_user.session_id
+        assert kwargs["queue_manager"] is queue_manager
         return runner_instance
 
-    mocker.patch("core.app.apps.workflow.app_generator.WorkflowAppRunner", side_effect=runner_ctor)
+    monkeypatch.setattr(
+        "core.app.apps.workflow.app_generator.WorkflowAppRunner",
+        MagicMock(side_effect=runner_ctor),
+    )
 
-    worker_lifecycle: dict[str, bool] = {}
+    worker_lifecycle: dict[str, object] = {}
 
     class ImmediateThread:
         def __init__(self, target, kwargs):
@@ -311,40 +493,24 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
         def is_alive(self):
             return False
 
-    mocker.patch("core.app.apps.workflow.app_generator.threading.Thread", ImmediateThread)
+    monkeypatch.setattr("core.app.apps.workflow.app_generator.threading.Thread", ImmediateThread)
 
-    mocker.patch(
-        "core.app.apps.workflow.app_generator.DifyCoreRepositoryFactory.create_workflow_execution_repository",
-        return_value=MagicMock(),
-    )
-    mocker.patch(
-        "core.app.apps.workflow.app_generator.DifyCoreRepositoryFactory.create_workflow_node_execution_repository",
-        return_value=MagicMock(),
+    pause_config = PauseStateLayerConfig(
+        session_factory=sessionmaker(bind=engine, expire_on_commit=False),
+        state_owner_user_id="owner",
     )
 
-    pause_config = SimpleNamespace(session_factory=MagicMock(), state_owner_user_id="owner")
-
-    app_model = SimpleNamespace(mode="workflow", tenant_id="tenant")
-    app_config = SimpleNamespace(app_id="app", tenant_id="tenant", workflow_id="workflow")
-    application_generate_entity = SimpleNamespace(
-        task_id="task",
-        user_id="user",
-        invoke_from="service-api",
-        app_config=app_config,
-        files=[],
-        stream=True,
-        workflow_execution_id="run",
-        trace_manager=MagicMock(),
-    )
+    application_generate_entity = _generate_entity(app, workflow, end_user)
+    workflow_execution_repository, workflow_node_execution_repository = _repositories(sqlite_session, app, end_user)
 
     result = generator.resume(
-        app_model=app_model,
+        app_model=app,
         workflow=workflow,
-        user=MagicMock(),
+        user=end_user,
         application_generate_entity=application_generate_entity,
         graph_runtime_state=runtime_state,
-        workflow_execution_repository=MagicMock(),
-        workflow_node_execution_repository=MagicMock(),
+        workflow_execution_repository=workflow_execution_repository,
+        workflow_node_execution_repository=workflow_node_execution_repository,
         pause_state_config=pause_config,
     )
 
@@ -352,4 +518,3 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
     assert worker_lifecycle["joined"] is True
     assert worker_lifecycle["join_timeout"] == 300
     runner_instance.run.assert_called_once()
-    queue_manager.graph_runtime_state = runtime_state