diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py index 8fc8959cd23..43dcab8c241 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_generator.py +++ b/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