From f8930aba34b76544561e425f9bb2ac1e3ccb4d32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9D=9E=E6=B3=95=E6=93=8D=E4=BD=9C?= Date: Fri, 4 Sep 2026 08:16:49 +0000 Subject: [PATCH] refactor(api): extract console workflow run application service (#41336) --- api/controllers/console/app/workflow_run.py | 297 +++++++---------- api/extensions/ext_application_services.py | 11 + .../api_workflow_run_repository.py | 3 +- .../sqlalchemy_api_workflow_run_repository.py | 85 ++++- api/services/workflow_run_service.py | 181 ++++++----- .../controllers/web/test_human_input_form.py | 14 +- .../layers/test_pause_state_persist_layer.py | 17 +- ...alchemy_workflow_run_cleanup_repository.py | 12 +- .../services/test_workflow_run_service.py | 264 ++++++--------- .../app/test_workflow_pause_details_api.py | 234 +++++--------- .../console/app/test_workflow_run_api.py | 175 +++++++--- .../test_ext_application_services.py | 18 ++ ..._sqlalchemy_api_workflow_run_repository.py | 137 +++++++- .../services/test_workflow_run_service.py | 305 +++++------------- .../test_workflow_run_service_pause.py | 133 +++++--- 15 files changed, 980 insertions(+), 906 deletions(-) diff --git a/api/controllers/console/app/workflow_run.py b/api/controllers/console/app/workflow_run.py index 26606e1c174..8a4b179439c 100644 --- a/api/controllers/console/app/workflow_run.py +++ b/api/controllers/console/app/workflow_run.py @@ -3,26 +3,19 @@ from uuid import UUID from flask_restx import Resource from pydantic import BaseModel, Field, field_validator -from sqlalchemy.orm import sessionmaker from configs import dify_config from controllers.common.errors import NotFoundError from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models from controllers.console import console_ns from controllers.console.app.wraps import get_app_model +from controllers.console.flask_admission import console_account_admission from controllers.console.wraps import ( RBACPermission, RBACResourceScope, - account_initialization_required, model_validate, - rbac_permission_required, - setup_required, - with_current_tenant_id, - with_current_user, ) -from core.workflow.human_input_forms import load_form_tokens_by_form_id as _load_form_tokens_by_form_id -from core.workflow.nodes.human_input.pause_reason import HumanInputRequired -from extensions.ext_database import db +from extensions.ext_application_services import application_services from fields.base import ResponseModel from fields.workflow_run_fields import ( AdvancedChatWorkflowRunPaginationResponse, @@ -32,14 +25,11 @@ from fields.workflow_run_fields import ( WorkflowRunNodeExecutionResponse, WorkflowRunPaginationResponse, ) -from graphon.enums import WorkflowExecutionStatus from libs.custom_inputs import time_duration -from libs.helper import uuid_value -from libs.login import login_required -from models import Account, App, AppMode, WorkflowRunTriggeredFrom -from models.workflow import WorkflowRun -from repositories.factory import DifyAPIRepositoryFactory -from services.workflow_run_service import WorkflowRunListArgs, WorkflowRunService +from libs.helper import dump_response, uuid_value +from machinery.context import RequestContext +from models import App, AppMode, WorkflowRunTriggeredFrom +from services.workflow_run_service import WorkflowRunListArgs def _build_backstage_input_url(form_token: str | None) -> str | None: @@ -51,10 +41,6 @@ def _build_backstage_input_url(form_token: str | None) -> str | None: return f"{base_url.rstrip('/')}/form/{form_token}" -# Workflow run status choices for filtering -WORKFLOW_RUN_STATUS_CHOICES = ["running", "succeeded", "failed", "stopped", "partial-succeeded"] - - class WorkflowRunListQuery(BaseModel): last_id: str | None = Field(default=None, description="Last run ID for pagination") limit: int = Field(default=20, ge=1, le=100, description="Number of items per page (1-100)") @@ -96,6 +82,19 @@ class WorkflowRunCountQuery(BaseModel): return time_duration(value) +def _workflow_run_list_args(req_data: WorkflowRunListQuery) -> WorkflowRunListArgs: + args: WorkflowRunListArgs = {"limit": req_data.limit} + if req_data.last_id is not None: + args["last_id"] = req_data.last_id + if req_data.status is not None: + args["status"] = req_data.status + return args + + +def _triggered_from(value: str | None) -> WorkflowRunTriggeredFrom: + return WorkflowRunTriggeredFrom(value) if value else WorkflowRunTriggeredFrom.DEBUGGING + + class HumanInputPauseTypeResponse(ResponseModel): type: Literal["human_input"] form_id: str @@ -143,37 +142,24 @@ class AdvancedChatAppWorkflowRunListApi(Resource): "Workflow runs retrieved successfully", console_ns.models[AdvancedChatWorkflowRunPaginationResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_CREATE_AND_MANAGEMENT, + ) @get_app_model(mode=[AppMode.ADVANCED_CHAT]) @model_validate(WorkflowRunListQuery) - def get(self, req_data: WorkflowRunListQuery, app_model: App): + def get(self, req_data: WorkflowRunListQuery, request_context: RequestContext, app_model: App): """ Get advanced chat app workflow run list """ - args: WorkflowRunListArgs = {"limit": req_data.limit} - if req_data.last_id is not None: - args["last_id"] = req_data.last_id - if req_data.status is not None: - args["status"] = req_data.status - - # Default to DEBUGGING if not specified - triggered_from = ( - WorkflowRunTriggeredFrom(req_data.triggered_from) - if req_data.triggered_from - else WorkflowRunTriggeredFrom.DEBUGGING + result = application_services().workflow_runs.get_paginate_advanced_chat_workflow_runs( + request_context, + app_id=app_model.id, + args=_workflow_run_list_args(req_data), + triggered_from=_triggered_from(req_data.triggered_from), ) - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_paginate_advanced_chat_workflow_runs( - app_model=app_model, args=args, triggered_from=triggered_from - ) - - return AdvancedChatWorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump( - mode="json" - ) + return dump_response(AdvancedChatWorkflowRunPaginationResponse, result) @console_ns.route("/apps//advanced-chat/workflow-runs/count") @@ -187,34 +173,25 @@ class AdvancedChatAppWorkflowRunCountApi(Resource): "Workflow runs count retrieved successfully", console_ns.models[WorkflowRunCountResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_CREATE_AND_MANAGEMENT, + ) @get_app_model(mode=[AppMode.ADVANCED_CHAT]) @model_validate(WorkflowRunCountQuery) - def get(self, req_data: WorkflowRunCountQuery, app_model: App): + def get(self, req_data: WorkflowRunCountQuery, request_context: RequestContext, app_model: App): """ Get advanced chat workflow runs count statistics """ - args = req_data.model_dump(exclude_none=True) - - # Default to DEBUGGING if not specified - triggered_from = ( - WorkflowRunTriggeredFrom(req_data.triggered_from) - if req_data.triggered_from - else WorkflowRunTriggeredFrom.DEBUGGING + result = application_services().workflow_runs.get_workflow_runs_count( + request_context, + app_id=app_model.id, + status=req_data.status, + time_range=req_data.time_range, + triggered_from=_triggered_from(req_data.triggered_from), ) - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_workflow_runs_count( - app_model=app_model, - status=args.get("status"), - time_range=args.get("time_range"), - triggered_from=triggered_from, - ) - - return WorkflowRunCountResponse.model_validate(result).model_dump(mode="json") + return dump_response(WorkflowRunCountResponse, result) @console_ns.route("/apps//workflow-runs") @@ -228,35 +205,24 @@ class WorkflowRunListApi(Resource): "Workflow runs retrieved successfully", console_ns.models[WorkflowRunPaginationResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_CREATE_AND_MANAGEMENT, + ) @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW]) @model_validate(WorkflowRunListQuery) - def get(self, req_data: WorkflowRunListQuery, app_model: App): + def get(self, req_data: WorkflowRunListQuery, request_context: RequestContext, app_model: App): """ Get workflow run list """ - args: WorkflowRunListArgs = {"limit": req_data.limit} - if req_data.last_id is not None: - args["last_id"] = req_data.last_id - if req_data.status is not None: - args["status"] = req_data.status - - # Default to DEBUGGING for workflow if not specified (backward compatibility) - triggered_from = ( - WorkflowRunTriggeredFrom(req_data.triggered_from) - if req_data.triggered_from - else WorkflowRunTriggeredFrom.DEBUGGING + result = application_services().workflow_runs.get_paginate_workflow_runs( + request_context, + app_id=app_model.id, + args=_workflow_run_list_args(req_data), + triggered_from=_triggered_from(req_data.triggered_from), ) - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_paginate_workflow_runs( - app_model=app_model, args=args, triggered_from=triggered_from - ) - - return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json") + return dump_response(WorkflowRunPaginationResponse, result) @console_ns.route("/apps//workflow-runs/count") @@ -270,34 +236,25 @@ class WorkflowRunCountApi(Resource): "Workflow runs count retrieved successfully", console_ns.models[WorkflowRunCountResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_CREATE_AND_MANAGEMENT, + ) @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW]) @model_validate(WorkflowRunCountQuery) - def get(self, req_data: WorkflowRunCountQuery, app_model: App): + def get(self, req_data: WorkflowRunCountQuery, request_context: RequestContext, app_model: App): """ Get workflow runs count statistics """ - args = req_data.model_dump(exclude_none=True) - - # Default to DEBUGGING for workflow if not specified (backward compatibility) - triggered_from = ( - WorkflowRunTriggeredFrom(req_data.triggered_from) - if req_data.triggered_from - else WorkflowRunTriggeredFrom.DEBUGGING + result = application_services().workflow_runs.get_workflow_runs_count( + request_context, + app_id=app_model.id, + status=req_data.status, + time_range=req_data.time_range, + triggered_from=_triggered_from(req_data.triggered_from), ) - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_workflow_runs_count( - app_model=app_model, - status=args.get("status"), - time_range=args.get("time_range"), - triggered_from=triggered_from, - ) - - return WorkflowRunCountResponse.model_validate(result).model_dump(mode="json") + return dump_response(WorkflowRunCountResponse, result) @console_ns.route("/apps//workflow-runs/") @@ -311,23 +268,24 @@ class WorkflowRunDetailApi(Resource): console_ns.models[WorkflowRunDetailResponse.__name__], ) @console_ns.response(404, "Workflow run not found") - @setup_required - @login_required - @account_initialization_required - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_CREATE_AND_MANAGEMENT, + ) @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW]) - def get(self, app_model: App, run_id: UUID): + def get(self, request_context: RequestContext, app_model: App, run_id: UUID): """ Get workflow run detail """ - run_id_str = str(run_id) - - workflow_run_service = WorkflowRunService() - workflow_run = workflow_run_service.get_workflow_run(app_model=app_model, run_id=run_id_str) + workflow_run = application_services().workflow_runs.get_workflow_run( + request_context, + app_id=app_model.id, + run_id=str(run_id), + ) if workflow_run is None: raise NotFoundError("Workflow run not found") - return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json") + return dump_response(WorkflowRunDetailResponse, workflow_run) @console_ns.route("/apps//workflow-runs//node-executions") @@ -341,28 +299,22 @@ class WorkflowRunNodeExecutionListApi(Resource): console_ns.models[WorkflowRunNodeExecutionListResponse.__name__], ) @console_ns.response(404, "Workflow run not found") - @setup_required - @login_required - @account_initialization_required - @with_current_user - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_CREATE_AND_MANAGEMENT, + ) @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW]) - def get(self, current_user: Account, app_model: App, run_id: UUID): + def get(self, request_context: RequestContext, app_model: App, run_id: UUID): """ Get workflow run node execution list """ - run_id_str = str(run_id) - - workflow_run_service = WorkflowRunService() - node_executions = workflow_run_service.get_workflow_run_node_executions( - app_model=app_model, - run_id=run_id_str, - user=current_user, + node_executions = application_services().workflow_runs.get_workflow_run_node_executions( + request_context, + app_id=app_model.id, + run_id=str(run_id), ) - return WorkflowRunNodeExecutionListResponse.model_validate( - {"data": node_executions}, from_attributes=True - ).model_dump(mode="json") + return dump_response(WorkflowRunNodeExecutionListResponse, {"data": node_executions}) @console_ns.route("/workflow//pause-details") @@ -378,11 +330,8 @@ class ConsoleWorkflowPauseDetailsApi(Resource): console_ns.models[WorkflowPauseDetailsResponse.__name__], ) @console_ns.response(404, "Workflow run not found") - @setup_required - @login_required - @account_initialization_required - @with_current_tenant_id - def get(self, current_tenant_id: str, workflow_run_id: str): + @console_account_admission() + def get(self, request_context: RequestContext, workflow_run_id: str): """ Get workflow pause details. @@ -391,51 +340,31 @@ class ConsoleWorkflowPauseDetailsApi(Resource): Returns information about why and where the workflow is paused. """ - # Query WorkflowRun to determine if workflow is suspended - session_maker = sessionmaker(bind=db.engine) - workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker=session_maker) - - workflow_run = db.session.get(WorkflowRun, workflow_run_id) - if not workflow_run: + details = application_services().workflow_runs.get_pause_details( + request_context, + workflow_run_id=workflow_run_id, + ) + if details is None: raise NotFoundError("Workflow run not found") - if workflow_run.tenant_id != current_tenant_id: - raise NotFoundError("Workflow run not found") - - # Check if workflow is suspended - is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED - if not is_paused: - empty_response = WorkflowPauseDetailsResponse(paused_at=None, paused_nodes=[]) - return empty_response.model_dump(mode="json"), 200 - - pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id) - pause_reasons = pause_entity.get_pause_reasons() if pause_entity else [] - form_tokens_by_form_id = _load_form_tokens_by_form_id( - [reason.form_id for reason in pause_reasons if isinstance(reason, HumanInputRequired)] + return ( + dump_response( + WorkflowPauseDetailsResponse, + { + "paused_at": details.paused_at.isoformat() + "Z" if details.paused_at else None, + "paused_nodes": [ + { + "node_id": node.node_id, + "node_title": node.node_title, + "pause_type": { + "type": "human_input", + "form_id": node.form_id, + "backstage_input_url": _build_backstage_input_url(node.form_token), + }, + } + for node in details.paused_nodes + ], + }, + ), + 200, ) - - # Build response - paused_at = pause_entity.paused_at if pause_entity else None - paused_nodes: list[PausedNodeResponse] = [] - - for reason in pause_reasons: - if isinstance(reason, HumanInputRequired): - paused_nodes.append( - PausedNodeResponse( - node_id=reason.node_id, - node_title=reason.node_title, - pause_type=HumanInputPauseTypeResponse( - type="human_input", - form_id=reason.form_id, - backstage_input_url=_build_backstage_input_url(form_tokens_by_form_id.get(reason.form_id)), - ), - ) - ) - else: - raise AssertionError("unimplemented.") - - response = WorkflowPauseDetailsResponse( - paused_at=paused_at.isoformat() + "Z" if paused_at else None, - paused_nodes=paused_nodes, - ) - return response.model_dump(mode="json"), 200 diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py index f2328234f19..c726edae8f5 100644 --- a/api/extensions/ext_application_services.py +++ b/api/extensions/ext_application_services.py @@ -49,6 +49,7 @@ from repositories.installation_state_repository import InstallationStateReposito from repositories.oauth_access_token_repository import SQLAlchemyOAuthAccessTokenRepository from repositories.oauth_server_repository import RedisOAuthServerTokenRepository, SQLAlchemyOAuthServerRepository from repositories.recommended_app_catalog_repository import DatabaseRecommendedAppCatalogRepository +from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from repositories.step_by_step_tour_repository import SQLAlchemyStepByStepTourStateRepository from repositories.tag_repository import TagRepository from repositories.trial_app_query_repository import TrialAppQueryRepository @@ -185,6 +186,7 @@ from services.webapp_access_query_service import ( WebAppAccessQueryService, WebAppAccessUnavailableError, ) +from services.workflow_run_service import WorkflowRunService from services.workflow_statistic_query_service import WorkflowStatisticQueryService from services.workspace_member_query_service import WorkspaceMemberQueryService from services.workspace_member_role_resolver import DeploymentWorkspaceMemberRoleResolver @@ -258,6 +260,7 @@ class ApplicationServices: remote_files: RemoteFileService trial_app_usage: TrialAppUsageRecorder workflow_run_archives: WorkflowRunArchiveService + workflow_runs: WorkflowRunService workspace_queries: WorkspaceQueryService workspace_member_queries: WorkspaceMemberQueryService inner_mail: InnerMailService @@ -414,6 +417,10 @@ def build_application_services( invitation_tokens = RedisInvitationTokenStore(redis=redis) activation_accounts = SQLAlchemyAccountActivationRepository(session_factory=database_client) account_provisioning = SQLAlchemyConsoleAuthProvisioningGateway(session_factory=database_client) + workflow_run_repository = DifyAPISQLAlchemyWorkflowRunRepository(session_maker=database_client) + workflow_node_execution_repository = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository( + session_maker=database_client + ) return ApplicationServices( accounts=AccountServices( access=AccountAccessService( @@ -660,6 +667,10 @@ def build_application_services( dispatcher=dispatch_workflow_run_archive_download_task, sign_download_url=sign_workflow_run_archive_download_url, ), + workflow_runs=WorkflowRunService( + workflow_runs=workflow_run_repository, + node_executions=workflow_node_execution_repository, + ), workspace_queries=WorkspaceQueryService( workspaces=workspace_query_repository, plans=DeploymentWorkspacePlanGateway(), diff --git a/api/repositories/api_workflow_run_repository.py b/api/repositories/api_workflow_run_repository.py index b3225794db9..f08782153b0 100644 --- a/api/repositories/api_workflow_run_repository.py +++ b/api/repositories/api_workflow_run_repository.py @@ -41,7 +41,6 @@ from typing import Protocol, TypedDict from sqlalchemy.orm import Session -from core.repositories.factory import WorkflowExecutionRepository from core.workflow.nodes.human_input.pause_reason import PauseReason as DifyPauseReason from graphon.entities.pause_reason import PauseReason as GraphonPauseReason from graphon.enums import WorkflowType @@ -82,7 +81,7 @@ class WorkflowRunCleanupRef: created_at: datetime -class APIWorkflowRunRepository(WorkflowExecutionRepository, Protocol): +class APIWorkflowRunRepository(Protocol): """ Protocol for service-layer WorkflowRun repository operations. diff --git a/api/repositories/sqlalchemy_api_workflow_run_repository.py b/api/repositories/sqlalchemy_api_workflow_run_repository.py index bc0a494384d..aa858282df1 100644 --- a/api/repositories/sqlalchemy_api_workflow_run_repository.py +++ b/api/repositories/sqlalchemy_api_workflow_run_repository.py @@ -22,10 +22,10 @@ Implementation Notes: import json import logging import uuid -from collections.abc import Callable, Sequence +from collections.abc import Callable, Mapping, Sequence from datetime import datetime from decimal import Decimal -from typing import Any, cast, override +from typing import Any, NamedTuple, cast, override import sqlalchemy as sa from pydantic import ValidationError @@ -33,6 +33,7 @@ from sqlalchemy import and_, delete, func, null, or_, select, tuple_ from sqlalchemy.engine import CursorResult from sqlalchemy.orm import Session, selectinload, sessionmaker +from core.workflow.human_input_forms import load_form_tokens_by_form_id from core.workflow.nodes.human_input.entities import FormDefinition from core.workflow.nodes.human_input.pause_reason import ( HumanInputRequired, @@ -55,6 +56,7 @@ from libs.datetime_utils import naive_utc_now from libs.helper import convert_datetime_to_date from libs.infinite_scroll_pagination import InfiniteScrollPagination from libs.time_parser import get_time_threshold +from models import Message from models.enums import WorkflowRunTriggeredFrom from models.human_input import HumanInputForm, HumanInputFormRecipient from models.workflow import WorkflowAppLog, WorkflowArchiveLog, WorkflowPause, WorkflowPauseReason, WorkflowRun @@ -76,6 +78,18 @@ logger = logging.getLogger(__name__) _HITL_REASON_TYPES = frozenset({PauseReasonType.LEGACY_HUMAN_INPUT_REQUIRED, PauseReasonType.HITL_REQUIRED}) +class WorkflowRunMessageRef(NamedTuple): + message_id: str + conversation_id: str + + +class WorkflowRunPauseRecord(NamedTuple): + status: WorkflowExecutionStatus + paused_at: datetime | None + reasons: tuple[DifyPauseReason, ...] + form_tokens: Mapping[str, str] + + class _WorkflowRunError(Exception): pass @@ -184,6 +198,31 @@ class DifyAPISQLAlchemyWorkflowRunRepository(APIWorkflowRunRepository): """ self._session_maker = session_maker + def get_message_refs( + self, + *, + app_id: str, + workflow_run_ids: Sequence[str], + ) -> dict[str, WorkflowRunMessageRef]: + if not workflow_run_ids: + return {} + + stmt = select(Message.workflow_run_id, Message.id, Message.conversation_id).where( + Message.app_id == app_id, + Message.workflow_run_id.in_(workflow_run_ids), + ) + with self._session_maker() as session: + rows = session.execute(stmt).all() + + messages_by_run_id: dict[str, WorkflowRunMessageRef] = {} + for workflow_run_id, message_id, conversation_id in rows: + if workflow_run_id is not None: + messages_by_run_id.setdefault( + workflow_run_id, + WorkflowRunMessageRef(message_id=message_id, conversation_id=conversation_id), + ) + return messages_by_run_id + @override def get_paginated_workflow_runs( self, @@ -1152,6 +1191,48 @@ class DifyAPISQLAlchemyWorkflowRunRepository(APIWorkflowRunRepository): pause_reasons=pause_reasons, ) + def get_pause_record( + self, + *, + workspace_id: str, + workflow_run_id: str, + ) -> WorkflowRunPauseRecord | None: + stmt = ( + select(WorkflowRun) + .options(selectinload(WorkflowRun.pause)) + .where( + WorkflowRun.tenant_id == workspace_id, + WorkflowRun.id == workflow_run_id, + ) + ) + with self._session_maker() as session: + workflow_run = session.scalar(stmt) + if workflow_run is None: + return None + if workflow_run.status != WorkflowExecutionStatus.PAUSED: + return WorkflowRunPauseRecord( + status=workflow_run.status, + paused_at=None, + reasons=(), + form_tokens={}, + ) + + pause_model = workflow_run.pause + if pause_model is None: + reasons: tuple[DifyPauseReason, ...] = () + else: + reason_models = self._get_reasons_by_pause_id(session, pause_model.id) + reasons = tuple(self._hydrate_pause_reasons(session, reason_models)) + form_ids = [reason.form_id for reason in reasons if isinstance(reason, HumanInputRequired)] + form_tokens = load_form_tokens_by_form_id(form_ids, session=session) + + return WorkflowRunPauseRecord( + status=workflow_run.status, + paused_at=pause_model.created_at if pause_model is not None else None, + reasons=reasons, + form_tokens=form_tokens, + ) + @override def resume_workflow_pause( self, diff --git a/api/services/workflow_run_service.py b/api/services/workflow_run_service.py index 8618b343ae6..dac000065cd 100644 --- a/api/services/workflow_run_service.py +++ b/api/services/workflow_run_service.py @@ -1,22 +1,19 @@ import threading +from dataclasses import dataclass +from datetime import datetime from typing import TypedDict -from sqlalchemy import Engine, select -from sqlalchemy.orm import sessionmaker - import contexts -from extensions.ext_database import db +from core.workflow.nodes.human_input.pause_reason import HumanInputRequired +from graphon.enums import WorkflowExecutionStatus from libs.infinite_scroll_pagination import InfiniteScrollPagination +from machinery.context import RequestContext from models import ( - Account, - App, - EndUser, - Message, WorkflowRun, WorkflowRunTriggeredFrom, ) -from repositories.api_workflow_run_repository import APIWorkflowRunRepository -from repositories.factory import DifyAPIRepositoryFactory +from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowNodeExecutionRepository +from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from services.workflow_node_execution_trace_service import ( WorkflowNodeExecutionTrace, assemble_workflow_node_execution_traces, @@ -31,34 +28,43 @@ class WorkflowRunListArgs(TypedDict, total=False): status: str +@dataclass(frozen=True, slots=True) +class WorkflowRunPausedNode: + node_id: str + node_title: str + form_id: str + form_token: str | None + + +@dataclass(frozen=True, slots=True) +class WorkflowRunPauseDetails: + paused_at: datetime | None + paused_nodes: tuple[WorkflowRunPausedNode, ...] + + class WorkflowRunService: - _session_factory: sessionmaker - _workflow_run_repo: APIWorkflowRunRepository - - def __init__(self, session_factory: Engine | sessionmaker | None = None): - """Initialize WorkflowRunService with repository dependencies.""" - match session_factory: - case None: - session_factory = sessionmaker(bind=db.engine, expire_on_commit=False) - case Engine(): - session_factory = sessionmaker(bind=session_factory, expire_on_commit=False) - - self._session_factory = session_factory - self._node_execution_service_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository( - self._session_factory - ) - self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(self._session_factory) + def __init__( + self, + *, + workflow_runs: DifyAPISQLAlchemyWorkflowRunRepository, + node_executions: DifyAPIWorkflowNodeExecutionRepository, + ) -> None: + self._workflow_runs = workflow_runs + self._node_executions = node_executions def get_paginate_advanced_chat_workflow_runs( self, - app_model: App, + context: RequestContext, + *, + app_id: str, args: WorkflowRunListArgs, triggered_from: WorkflowRunTriggeredFrom = WorkflowRunTriggeredFrom.DEBUGGING, ) -> InfiniteScrollPagination: """ Get advanced chat app workflow run list - :param app_model: app model + :param context: admitted Console request context + :param app_id: app id :param args: request args :param triggered_from: workflow run triggered from (default: DEBUGGING for preview runs) """ @@ -73,35 +79,29 @@ class WorkflowRunService: def __getattr__(self, item): return getattr(self._workflow_run, item) - pagination = self.get_paginate_workflow_runs(app_model, args, triggered_from) + pagination = self.get_paginate_workflow_runs( + context, + app_id=app_id, + args=args, + triggered_from=triggered_from, + ) # Batch-load the associated Message for every run in a single query to avoid # an N+1 pattern: the deprecated WorkflowRun.message property issues one query # per run. The filter matches that property exactly (app_id + workflow_run_id). workflow_runs = pagination.data run_ids = [workflow_run.id for workflow_run in workflow_runs] - messages_by_run_id: dict[str, Message] = {} - if run_ids: - with self._session_factory() as session: - messages = session.scalars( - select(Message).where( - Message.app_id == app_model.id, - Message.workflow_run_id.in_(run_ids), - ) - ).all() - for loaded_message in messages: - run_id = loaded_message.workflow_run_id - if run_id is None: - continue - # setdefault mirrors scalar()'s single-row-per-run semantics. - messages_by_run_id.setdefault(run_id, loaded_message) + messages_by_run_id = self._workflow_runs.get_message_refs( + app_id=app_id, + workflow_run_ids=run_ids, + ) with_message_workflow_runs = [] for workflow_run in workflow_runs: message = messages_by_run_id.get(workflow_run.id) with_message_workflow_run = WorkflowWithMessage(workflow_run=workflow_run) if message: - with_message_workflow_run.message_id = message.id + with_message_workflow_run.message_id = message.message_id with_message_workflow_run.conversation_id = message.conversation_id with_message_workflow_runs.append(with_message_workflow_run) @@ -111,14 +111,17 @@ class WorkflowRunService: def get_paginate_workflow_runs( self, - app_model: App, + context: RequestContext, + *, + app_id: str, args: WorkflowRunListArgs, triggered_from: WorkflowRunTriggeredFrom = WorkflowRunTriggeredFrom.DEBUGGING, ) -> InfiniteScrollPagination: """ Get workflow run list - :param app_model: app model + :param context: admitted Console request context + :param app_id: app id :param args: request args :param triggered_from: workflow run triggered from (default: DEBUGGING) """ @@ -126,31 +129,34 @@ class WorkflowRunService: last_id = args.get("last_id") status = args.get("status") - return self._workflow_run_repo.get_paginated_workflow_runs( - tenant_id=app_model.tenant_id, - app_id=app_model.id, + return self._workflow_runs.get_paginated_workflow_runs( + tenant_id=context.active_workspace_id, + app_id=app_id, triggered_from=triggered_from, limit=limit, last_id=last_id, status=status, ) - def get_workflow_run(self, app_model: App, run_id: str) -> WorkflowRun | None: + def get_workflow_run(self, context: RequestContext, *, app_id: str, run_id: str) -> WorkflowRun | None: """ Get workflow run detail - :param app_model: app model + :param context: admitted Console request context + :param app_id: app id :param run_id: workflow run id """ - return self._workflow_run_repo.get_workflow_run_by_id( - tenant_id=app_model.tenant_id, - app_id=app_model.id, + return self._workflow_runs.get_workflow_run_by_id( + tenant_id=context.active_workspace_id, + app_id=app_id, run_id=run_id, ) def get_workflow_runs_count( self, - app_model: App, + context: RequestContext, + *, + app_id: str, status: str | None = None, time_range: str | None = None, triggered_from: WorkflowRunTriggeredFrom = WorkflowRunTriggeredFrom.DEBUGGING, @@ -158,15 +164,16 @@ class WorkflowRunService: """ Get workflow runs count statistics - :param app_model: app model + :param context: admitted Console request context + :param app_id: app id :param status: optional status filter :param time_range: optional time range filter (e.g., "7d", "4h", "30m", "30s") :param triggered_from: workflow run triggered from (default: DEBUGGING) :return: dict with total and status counts """ - return self._workflow_run_repo.get_workflow_runs_count( - tenant_id=app_model.tenant_id, - app_id=app_model.id, + return self._workflow_runs.get_workflow_runs_count( + tenant_id=context.active_workspace_id, + app_id=app_id, triggered_from=triggered_from, status=status, time_range=time_range, @@ -174,14 +181,15 @@ class WorkflowRunService: def get_workflow_run_node_executions( self, - app_model: App, + context: RequestContext, + *, + app_id: str, run_id: str, - user: Account | EndUser, ) -> list[WorkflowNodeExecutionTrace]: """ Get workflow run node execution list """ - workflow_run = self.get_workflow_run(app_model, run_id) + workflow_run = self.get_workflow_run(context, app_id=app_id, run_id=run_id) contexts.plugin_tool_providers.set({}) contexts.plugin_tool_providers_lock.set(threading.Lock()) @@ -189,14 +197,43 @@ class WorkflowRunService: if not workflow_run: return [] - # Get tenant_id from user - tenant_id = user.tenant_id if isinstance(user, EndUser) else user.current_tenant_id - if tenant_id is None: - raise ValueError("User tenant_id cannot be None") - - node_executions = self._node_execution_service_repo.get_executions_by_workflow_run( - tenant_id=tenant_id, - app_id=app_model.id, + node_executions = self._node_executions.get_executions_by_workflow_run( + tenant_id=context.active_workspace_id, + app_id=app_id, workflow_run_id=run_id, ) - return assemble_workflow_node_execution_traces(node_executions, self._node_execution_service_repo) + return assemble_workflow_node_execution_traces(node_executions, self._node_executions) + + def get_pause_details( + self, + context: RequestContext, + *, + workflow_run_id: str, + ) -> WorkflowRunPauseDetails | None: + pause_record = self._workflow_runs.get_pause_record( + workspace_id=context.active_workspace_id, + workflow_run_id=workflow_run_id, + ) + if pause_record is None: + return None + if pause_record.status != WorkflowExecutionStatus.PAUSED: + return WorkflowRunPauseDetails(paused_at=None, paused_nodes=()) + + human_input_reasons: list[HumanInputRequired] = [] + for reason in pause_record.reasons: + if not isinstance(reason, HumanInputRequired): + raise NotImplementedError(f"Pause details do not support {type(reason).__name__}") + human_input_reasons.append(reason) + + return WorkflowRunPauseDetails( + paused_at=pause_record.paused_at, + paused_nodes=tuple( + WorkflowRunPausedNode( + node_id=reason.node_id, + node_title=reason.node_title, + form_id=reason.form_id, + form_token=pause_record.form_tokens.get(reason.form_id), + ) + for reason in human_input_reasons + ), + ) diff --git a/api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py b/api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py index 363197e6c74..d80e3fe31ff 100644 --- a/api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py +++ b/api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py @@ -2,7 +2,6 @@ from __future__ import annotations import json from datetime import UTC, datetime, timedelta -from typing import override from uuid import uuid4 import pytest @@ -22,7 +21,6 @@ from core.workflow.nodes.human_input.entities import ( ) from core.workflow.nodes.human_input.enums import HumanInputFormKind, HumanInputFormStatus, ValueSourceType from core.workflow.nodes.human_input.pause_reason import HumanInputRequired -from graphon.entities import WorkflowExecution from graphon.enums import WorkflowExecutionStatus from graphon.runtime import GraphRuntimeState, VariablePool from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole @@ -40,14 +38,6 @@ from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchem from services.entities.feature_entities import FeatureModel -class _TestWorkflowRunRepository(DifyAPISQLAlchemyWorkflowRunRepository): - """Concrete repository for tests where save() is not under test.""" - - @override - def save(self, execution: WorkflowExecution) -> None: - return None - - def _create_app_with_site(session: Session) -> tuple[App, Account]: tenant = Tenant(name="Test Tenant") account = Account(name="Tester", email=f"tester-{uuid4()}@example.com") @@ -218,7 +208,9 @@ def test_get_human_input_form_resolves_runtime_select_options( ) engine = db_session_with_containers.get_bind() assert isinstance(engine, Engine) - workflow_run_repo = _TestWorkflowRunRepository(session_maker=sessionmaker(bind=engine, expire_on_commit=False)) + workflow_run_repo = DifyAPISQLAlchemyWorkflowRunRepository( + session_maker=sessionmaker(bind=engine, expire_on_commit=False) + ) workflow_run_repo.create_workflow_pause( workflow_run_id=workflow_run.id, state_owner_user_id=account.id, diff --git a/api/tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py b/api/tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py index 2d5e3ebcca9..56da1c948d1 100644 --- a/api/tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py +++ b/api/tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py @@ -23,7 +23,7 @@ from unittest.mock import Mock import pytest from sqlalchemy import Engine, delete, select -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, sessionmaker from core.app.app_config.entities import WorkflowUIBasedAppConfig from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity @@ -46,6 +46,8 @@ from models import Account from models import WorkflowPause as WorkflowPauseModel from models.model import AppMode, UploadFile from models.workflow import Workflow, WorkflowRun +from repositories.factory import DifyAPIRepositoryFactory +from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from services.file_service import FileService from services.workflow_run_service import WorkflowRunService @@ -99,7 +101,14 @@ class TestPauseStatePersistenceLayerTestContainers: @pytest.fixture def workflow_run_service(self, engine: Engine, file_service: FileService): """Create WorkflowRunService instance with TestContainers engine and FileService.""" - return WorkflowRunService(engine) + session_factory = sessionmaker(bind=engine, expire_on_commit=False) + workflow_runs = DifyAPISQLAlchemyWorkflowRunRepository(session_maker=session_factory) + return WorkflowRunService( + workflow_runs=workflow_runs, + node_executions=DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository( + session_maker=session_factory + ), + ) @pytest.fixture(autouse=True) def setup_test_data(self, db_session_with_containers: Session, file_service, workflow_run_service): @@ -403,7 +412,7 @@ class TestPauseStatePersistenceLayerTestContainers: layer.on_event(event) # Assert - Retrieve and verify - pause_entity = self.workflow_run_service._workflow_run_repo.get_workflow_pause(self.test_workflow_run_id) + pause_entity = self.workflow_run_service._workflow_runs.get_workflow_pause(self.test_workflow_run_id) assert pause_entity is not None assert pause_entity.workflow_execution_id == self.test_workflow_run_id assert pause_entity.get_pause_reasons() == event.reasons @@ -542,7 +551,7 @@ class TestPauseStatePersistenceLayerTestContainers: assert pause_model is not None # Verify the state owner is the workflow creator - pause_entity = self.workflow_run_service._workflow_run_repo.get_workflow_pause(different_workflow_run.id) + pause_entity = self.workflow_run_service._workflow_runs.get_workflow_pause(different_workflow_run.id) assert pause_entity is not None resumption_context = WorkflowResumptionContext.loads(pause_entity.get_state().decode()) assert resumption_context.get_generate_entity().workflow_execution_id == different_workflow_run.id diff --git a/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_workflow_run_cleanup_repository.py b/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_workflow_run_cleanup_repository.py index be659fac184..7fc557f9b11 100644 --- a/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_workflow_run_cleanup_repository.py +++ b/api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_workflow_run_cleanup_repository.py @@ -4,13 +4,11 @@ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timedelta -from typing import override from uuid import uuid4 from sqlalchemy import Engine, select from sqlalchemy.orm import Session, sessionmaker -from graphon.entities import WorkflowExecution from graphon.entities.pause_reason import PauseReasonType from graphon.enums import WorkflowExecutionStatus, WorkflowType from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom @@ -18,14 +16,6 @@ from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowP from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository -class _TestWorkflowRunRepository(DifyAPISQLAlchemyWorkflowRunRepository): - """Concrete repository for tests where save() is not under test.""" - - @override - def save(self, execution: WorkflowExecution) -> None: - return None - - @dataclass class _TestScope: """Per-test identifiers for rows created by cleanup repository tests.""" @@ -39,7 +29,7 @@ class _TestScope: def _repository(db_session_with_containers: Session) -> DifyAPISQLAlchemyWorkflowRunRepository: engine = db_session_with_containers.get_bind() assert isinstance(engine, Engine) - return _TestWorkflowRunRepository(session_maker=sessionmaker(bind=engine, expire_on_commit=False)) + return DifyAPISQLAlchemyWorkflowRunRepository(session_maker=sessionmaker(bind=engine, expire_on_commit=False)) def _create_workflow_run( diff --git a/api/tests/test_containers_integration_tests/services/test_workflow_run_service.py b/api/tests/test_containers_integration_tests/services/test_workflow_run_service.py index 263f41e5144..8d37aaa4e8c 100644 --- a/api/tests/test_containers_integration_tests/services/test_workflow_run_service.py +++ b/api/tests/test_containers_integration_tests/services/test_workflow_run_service.py @@ -5,13 +5,17 @@ from unittest.mock import patch import pytest from faker import Faker -from sqlalchemy.orm import Session +from sqlalchemy import Engine +from sqlalchemy.orm import Session, sessionmaker -from models.enums import ConversationFromSource, CreatorUserRole, EndUserType +from machinery.context import RequestContext +from models.enums import ConversationFromSource, CreatorUserRole from models.model import ( Message, ) from models.workflow import WorkflowRun +from repositories.factory import DifyAPIRepositoryFactory +from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from services.account_service import AccountService, TenantService from services.app_service import AppService, CreateAppParams from services.workflow_run_service import WorkflowRunService @@ -21,6 +25,19 @@ from tests.test_containers_integration_tests.helpers import generate_valid_passw class TestWorkflowRunService: """Integration tests for WorkflowRunService using testcontainers.""" + @pytest.fixture + def workflow_run_service(self, db_session_with_containers: Session) -> WorkflowRunService: + engine = db_session_with_containers.get_bind() + assert isinstance(engine, Engine) + session_factory = sessionmaker(bind=engine, expire_on_commit=False) + workflow_runs = DifyAPISQLAlchemyWorkflowRunRepository(session_maker=session_factory) + return WorkflowRunService( + workflow_runs=workflow_runs, + node_executions=DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository( + session_maker=session_factory + ), + ) + @pytest.fixture def mock_external_service_dependencies(self): """Mock setup for external service dependencies.""" @@ -50,6 +67,15 @@ class TestWorkflowRunService: "account_feature_service": mock_account_feature_service, } + @staticmethod + def _request_context(app, account) -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id=account.id, + active_workspace_id=app.tenant_id, + ) + def _create_test_app_and_account(self, db_session_with_containers: Session, mock_external_service_dependencies): """ Helper method to create a test app and account for testing. @@ -196,7 +222,10 @@ class TestWorkflowRunService: return message def test_get_paginate_workflow_runs_success( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test successful pagination of workflow runs with debugging trigger. @@ -218,9 +247,12 @@ class TestWorkflowRunService: workflow_runs.append(workflow_run) # Act: Execute the method under test - workflow_run_service = WorkflowRunService() args = {"limit": 3, "last_id": None} - result = workflow_run_service.get_paginate_workflow_runs(app, args) + result = workflow_run_service.get_paginate_workflow_runs( + self._request_context(app, account), + app_id=app.id, + args=args, + ) # Assert: Verify the expected outcomes assert result is not None @@ -238,7 +270,10 @@ class TestWorkflowRunService: assert workflow_run.tenant_id == app.tenant_id def test_get_paginate_workflow_runs_with_last_id( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test pagination of workflow runs with last_id parameter. @@ -261,9 +296,12 @@ class TestWorkflowRunService: workflow_runs.append(workflow_run) # Act: Execute the method under test with last_id - workflow_run_service = WorkflowRunService() args = {"limit": 2, "last_id": workflow_runs[1].id} - result = workflow_run_service.get_paginate_workflow_runs(app, args) + result = workflow_run_service.get_paginate_workflow_runs( + self._request_context(app, account), + app_id=app.id, + args=args, + ) # Assert: Verify the expected outcomes assert result is not None @@ -281,7 +319,10 @@ class TestWorkflowRunService: assert workflow_run.tenant_id == app.tenant_id def test_get_paginate_workflow_runs_default_limit( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test pagination of workflow runs with default limit. @@ -299,9 +340,12 @@ class TestWorkflowRunService: workflow_run = self._create_test_workflow_run(db_session_with_containers, app, account, "debugging") # Act: Execute the method under test without limit - workflow_run_service = WorkflowRunService() args = {} # No limit specified - result = workflow_run_service.get_paginate_workflow_runs(app, args) + result = workflow_run_service.get_paginate_workflow_runs( + self._request_context(app, account), + app_id=app.id, + args=args, + ) # Assert: Verify the expected outcomes assert result is not None @@ -319,7 +363,10 @@ class TestWorkflowRunService: assert workflow_run_result.tenant_id == app.tenant_id def test_get_paginate_advanced_chat_workflow_runs_success( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test successful pagination of advanced chat workflow runs with message information. @@ -344,9 +391,12 @@ class TestWorkflowRunService: workflow_runs.append(workflow_run) # Act: Execute the method under test - workflow_run_service = WorkflowRunService() args = {"limit": 2, "last_id": None} - result = workflow_run_service.get_paginate_advanced_chat_workflow_runs(app, args) + result = workflow_run_service.get_paginate_advanced_chat_workflow_runs( + self._request_context(app, account), + app_id=app.id, + args=args, + ) # Assert: Verify the expected outcomes assert result is not None @@ -364,7 +414,12 @@ class TestWorkflowRunService: assert workflow_run.app_id == app.id assert workflow_run.tenant_id == app.tenant_id - def test_get_workflow_run_success(self, db_session_with_containers: Session, mock_external_service_dependencies): + def test_get_workflow_run_success( + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, + ): """ Test successful retrieval of workflow run by ID. @@ -381,8 +436,11 @@ class TestWorkflowRunService: workflow_run = self._create_test_workflow_run(db_session_with_containers, app, account, "debugging") # Act: Execute the method under test - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_workflow_run(app, workflow_run.id) + result = workflow_run_service.get_workflow_run( + self._request_context(app, account), + app_id=app.id, + run_id=workflow_run.id, + ) # Assert: Verify the expected outcomes assert result is not None @@ -394,7 +452,12 @@ class TestWorkflowRunService: assert result.type == "chat" assert result.version == "1.0.0" - def test_get_workflow_run_not_found(self, db_session_with_containers: Session, mock_external_service_dependencies): + def test_get_workflow_run_not_found( + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, + ): """ Test workflow run retrieval when run ID does not exist. @@ -411,14 +474,20 @@ class TestWorkflowRunService: non_existent_id = str(uuid.uuid4()) # Act: Execute the method under test - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_workflow_run(app, non_existent_id) + result = workflow_run_service.get_workflow_run( + self._request_context(app, account), + app_id=app.id, + run_id=non_existent_id, + ) # Assert: Verify the expected outcomes assert result is None def test_get_workflow_run_node_executions_success( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test successful retrieval of workflow run node executions. @@ -487,8 +556,11 @@ class TestWorkflowRunService: db_session_with_containers.commit() # Act: Execute the method under test - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_workflow_run_node_executions(app, workflow_run.id, account) + result = workflow_run_service.get_workflow_run_node_executions( + self._request_context(app, account), + app_id=app.id, + run_id=workflow_run.id, + ) # Assert: Verify the expected outcomes assert result is not None @@ -507,7 +579,10 @@ class TestWorkflowRunService: assert node_execution.node_id.startswith("node_") def test_get_workflow_run_node_executions_empty( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test getting node executions for a workflow run with no executions. @@ -521,7 +596,6 @@ class TestWorkflowRunService: account_service = AccountService() tenant_service = TenantService() app_service = AppService() - workflow_run_service = WorkflowRunService() # Create account and tenant account = account_service.create_account( @@ -549,9 +623,9 @@ class TestWorkflowRunService: # Act: Get node executions result = workflow_run_service.get_workflow_run_node_executions( - app_model=app, + self._request_context(app, account), + app_id=app.id, run_id=workflow_run.id, - user=account, ) # Assert: Verify empty result @@ -559,7 +633,10 @@ class TestWorkflowRunService: assert len(result) == 0 def test_get_workflow_run_node_executions_invalid_workflow_run_id( - self, db_session_with_containers: Session, mock_external_service_dependencies + self, + db_session_with_containers: Session, + mock_external_service_dependencies, + workflow_run_service: WorkflowRunService, ): """ Test getting node executions with invalid workflow run ID. @@ -573,7 +650,6 @@ class TestWorkflowRunService: account_service = AccountService() tenant_service = TenantService() app_service = AppService() - workflow_run_service = WorkflowRunService() # Create account and tenant account = account_service.create_account( @@ -601,137 +677,11 @@ class TestWorkflowRunService: # Act: Get node executions with invalid ID result = workflow_run_service.get_workflow_run_node_executions( - app_model=app, + self._request_context(app, account), + app_id=app.id, run_id=invalid_workflow_run_id, - user=account, ) # Assert: Verify empty result assert result is not None assert len(result) == 0 - - def test_get_workflow_run_node_executions_database_error( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test getting node executions when database encounters an error. - - This test verifies: - - Proper error handling when database operations fail - - Graceful degradation in error scenarios - - Error propagation to calling code - """ - # Arrange: Setup test data - account_service = AccountService() - tenant_service = TenantService() - app_service = AppService() - workflow_run_service = WorkflowRunService() - - # Create account and tenant - account = account_service.create_account( - email="test@example.com", - name="Test User", - password="password123", - interface_language="en-US", - session=db_session_with_containers, - ) - TenantService.create_owner_tenant_if_not_exist(account, name="test_tenant", session=db_session_with_containers) - tenant = account.current_tenant - - # Create app - app_args = CreateAppParams( - name="Test App", - mode="chat", - icon_type="emoji", - icon="🚀", - icon_background="#4ECDC4", - ) - app = app_service.create_app(tenant.id, app_args, account, session=db_session_with_containers) - - # Create workflow run - workflow_run = self._create_test_workflow_run(db_session_with_containers, app, account, "debugging") - - # Mock database error by closing the session - db_session_with_containers.close() - - # Act & Assert: Verify error handling - with pytest.raises((Exception, RuntimeError)): - workflow_run_service.get_workflow_run_node_executions( - app_model=app, - run_id=workflow_run.id, - user=account, - ) - - def test_get_workflow_run_node_executions_end_user( - self, db_session_with_containers: Session, mock_external_service_dependencies - ): - """ - Test node execution retrieval for end user. - - This test verifies: - - Proper handling of end user vs account user - - Correct tenant ID extraction for end users - - Repository method calls with proper parameters - """ - # Arrange: Create test data - fake = Faker() - app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies) - - # Create workflow run - workflow_run = self._create_test_workflow_run(db_session_with_containers, app, account, "debugging") - - # Create end user - from models.model import EndUser - - end_user = EndUser( - tenant_id=app.tenant_id, - app_id=app.id, - type=EndUserType.BROWSER, - is_anonymous=False, - session_id=str(uuid.uuid4()), - external_user_id=str(uuid.uuid4()), - name=fake.name(), - ) - db_session_with_containers.add(end_user) - db_session_with_containers.commit() - - # Create node execution - from models.workflow import WorkflowNodeExecutionModel - - node_execution = WorkflowNodeExecutionModel( - tenant_id=app.tenant_id, - app_id=app.id, - workflow_id=workflow_run.workflow_id, - triggered_from="workflow-run", - workflow_run_id=workflow_run.id, - index=0, - node_id="node_0", - node_type="llm", - title="Node 0", - inputs=json.dumps({"input": "test_input"}), - process_data=json.dumps({"process": "test_process"}), - status="succeeded", - elapsed_time=0.5, - execution_metadata=json.dumps({"tokens": 50}), - created_by_role=CreatorUserRole.END_USER, - created_by=end_user.id, - created_at=datetime.now(UTC), - ) - db_session_with_containers.add(node_execution) - db_session_with_containers.commit() - - # Act: Execute the method under test - workflow_run_service = WorkflowRunService() - result = workflow_run_service.get_workflow_run_node_executions(app, workflow_run.id, end_user) - - # Assert: Verify the expected outcomes - assert result is not None - assert len(result) == 1 - - # Verify node execution properties - node_exec = result[0] - assert node_exec.tenant_id == app.tenant_id - assert node_exec.app_id == app.id - assert node_exec.workflow_run_id == workflow_run.id - assert node_exec.created_by == end_user.id - assert node_exec.created_by_role == CreatorUserRole.END_USER diff --git a/api/tests/unit_tests/controllers/console/app/test_workflow_pause_details_api.py b/api/tests/unit_tests/controllers/console/app/test_workflow_pause_details_api.py index a1be8254d63..c5872bbf434 100644 --- a/api/tests/unit_tests/controllers/console/app/test_workflow_pause_details_api.py +++ b/api/tests/unit_tests/controllers/console/app/test_workflow_pause_details_api.py @@ -1,193 +1,111 @@ -"""Console workflow pause-detail tests backed by persisted workflow execution state.""" +"""Controller tests for Console workflow pause details.""" from __future__ import annotations -import inspect -from dataclasses import dataclass from datetime import datetime +from inspect import unwrap +from types import SimpleNamespace from unittest.mock import Mock -from uuid import uuid4 import pytest from flask import Flask -from sqlalchemy.engine import Engine -from sqlalchemy.orm import Session from controllers.common.errors import NotFoundError from controllers.console.app import workflow_run as workflow_run_module -from core.workflow.nodes.human_input.entities import ParagraphInputConfig, UserActionConfig -from core.workflow.nodes.human_input.pause_reason import HumanInputRequired -from graphon.enums import WorkflowExecutionStatus -from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom -from models.workflow import WorkflowPause, WorkflowRun, WorkflowType +from machinery.context import RequestContext +from services.workflow_run_service import WorkflowRunPauseDetails, WorkflowRunPausedNode from tests.unit_tests.config_override import apply_config_overrides -@dataclass(frozen=True) -class _Database: - engine: Engine - session: Session - - -def _persist_run( - session: Session, - *, - run_id: str, - tenant_id: str, - status: WorkflowExecutionStatus, - paused: bool = False, -) -> WorkflowRun: - workflow_id = str(uuid4()) - workflow_run = WorkflowRun( - id=run_id, - tenant_id=tenant_id, - app_id=str(uuid4()), - workflow_id=workflow_id, - type=WorkflowType.WORKFLOW, - triggered_from=WorkflowRunTriggeredFrom.DEBUGGING, - version="draft", - graph="{}", - inputs="{}", - status=status, - created_by_role=CreatorUserRole.ACCOUNT, - created_by=str(uuid4()), - created_at=datetime(2024, 1, 1, 12, 0, 0), +def _request_context(*, workspace_id: str = "tenant-1") -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id="account-1", + active_workspace_id=workspace_id, ) - session.add(workflow_run) - if paused: - session.add( - WorkflowPause( - workflow_id=workflow_id, - workflow_run_id=run_id, - state_object_key="workflow-pauses/state.json", - ) - ) - session.commit() - return workflow_run -class _PauseEntity: - def __init__(self, paused_at: datetime, reasons: list[HumanInputRequired]): - self.paused_at = paused_at - self._reasons = reasons - - def get_pause_reasons(self): - return self._reasons - - -def test_pause_details_returns_backstage_input_url( - app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session -) -> None: - apply_config_overrides(monkeypatch, APP_WEB_URL="https://web.example.com") - - tenant_id = str(uuid4()) - run_id = str(uuid4()) - _persist_run( - sqlite_session, - run_id=run_id, - tenant_id=tenant_id, - status=WorkflowExecutionStatus.PAUSED, - paused=True, - ) +def _mock_application_services(monkeypatch: pytest.MonkeyPatch, workflow_runs: Mock) -> None: monkeypatch.setattr( workflow_run_module, - "db", - _Database(engine=sqlite_session.get_bind(), session=sqlite_session), + "application_services", + lambda: SimpleNamespace(workflow_runs=workflow_runs), ) - reason = HumanInputRequired( - form_id="form-1", - form_content="content", - inputs=[ParagraphInputConfig(output_variable_name="name")], - actions=[UserActionConfig(id="approve", title="Approve")], - node_id="node-1", - node_title="Ask Name", - ) - pause_entity = _PauseEntity(paused_at=datetime(2024, 1, 1, 12, 0, 0), reasons=[reason]) - repo = Mock() - repo.get_workflow_pause.return_value = pause_entity - monkeypatch.setattr( - workflow_run_module.DifyAPIRepositoryFactory, - "create_api_workflow_run_repository", - lambda *_, **__: repo, - ) - monkeypatch.setattr( - workflow_run_module, - "_load_form_tokens_by_form_id", - lambda _form_ids: {"form-1": "backstage-token"}, +def test_pause_details_returns_backstage_input_url(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: + apply_config_overrides(monkeypatch, APP_WEB_URL="https://web.example.com/") + workflow_runs = Mock() + workflow_runs.get_pause_details.return_value = WorkflowRunPauseDetails( + paused_at=datetime(2024, 1, 1, 12, 0, 0), + paused_nodes=( + WorkflowRunPausedNode( + node_id="node-1", + node_title="Ask Name", + form_id="form-1", + form_token="backstage-token", + ), + ), ) + _mock_application_services(monkeypatch, workflow_runs) + request_context = _request_context() - with app.test_request_context(f"/console/api/workflow/{run_id}/pause-details", method="GET"): - handler = inspect.unwrap(workflow_run_module.ConsoleWorkflowPauseDetailsApi.get) - response, status = handler( - workflow_run_module.ConsoleWorkflowPauseDetailsApi(), - tenant_id, - workflow_run_id=run_id, - ) + api = workflow_run_module.ConsoleWorkflowPauseDetailsApi() + handler = unwrap(api.get) + with app.test_request_context("/console/api/workflow/run-1/pause-details", method="GET"): + response, status = handler(api, request_context, workflow_run_id="run-1") assert status == 200 - assert response["paused_at"] == "2024-01-01T12:00:00Z" - assert response["paused_nodes"][0]["node_id"] == "node-1" - assert response["paused_nodes"][0]["pause_type"]["type"] == "human_input" - assert ( - response["paused_nodes"][0]["pause_type"]["backstage_input_url"] - == "https://web.example.com/form/backstage-token" - ) - assert "pending_human_inputs" not in response - - -def test_pause_details_tenant_isolation(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None: - apply_config_overrides(monkeypatch, APP_WEB_URL="https://web.example.com") - - run_id = str(uuid4()) - _persist_run( - sqlite_session, - run_id=run_id, - tenant_id=str(uuid4()), - status=WorkflowExecutionStatus.PAUSED, - paused=True, - ) - monkeypatch.setattr( - workflow_run_module, - "db", - _Database(engine=sqlite_session.get_bind(), session=sqlite_session), + assert response == { + "paused_at": "2024-01-01T12:00:00Z", + "paused_nodes": [ + { + "node_id": "node-1", + "node_title": "Ask Name", + "pause_type": { + "type": "human_input", + "form_id": "form-1", + "backstage_input_url": "https://web.example.com/form/backstage-token", + }, + } + ], + } + workflow_runs.get_pause_details.assert_called_once_with( + request_context, + workflow_run_id="run-1", ) - handler = inspect.unwrap(workflow_run_module.ConsoleWorkflowPauseDetailsApi.get) - with app.test_request_context(f"/console/api/workflow/{run_id}/pause-details", method="GET"): - with pytest.raises(NotFoundError): - handler( - workflow_run_module.ConsoleWorkflowPauseDetailsApi(), - str(uuid4()), - workflow_run_id=run_id, - ) - -def test_pause_details_returns_empty_response_for_non_paused_run( - app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session +def test_pause_details_maps_missing_or_inaccessible_run_to_not_found( + app: Flask, monkeypatch: pytest.MonkeyPatch ) -> None: - tenant_id = str(uuid4()) - run_id = str(uuid4()) - _persist_run( - sqlite_session, - run_id=run_id, - tenant_id=tenant_id, - status=WorkflowExecutionStatus.RUNNING, - ) - monkeypatch.setattr( - workflow_run_module, - "db", - _Database(engine=sqlite_session.get_bind(), session=sqlite_session), + workflow_runs = Mock() + workflow_runs.get_pause_details.return_value = None + _mock_application_services(monkeypatch, workflow_runs) + request_context = _request_context(workspace_id="other-tenant") + api = workflow_run_module.ConsoleWorkflowPauseDetailsApi() + handler = unwrap(api.get) + + with app.test_request_context("/console/api/workflow/run-1/pause-details", method="GET"): + with pytest.raises(NotFoundError, match="Workflow run not found"): + handler(api, request_context, workflow_run_id="run-1") + + workflow_runs.get_pause_details.assert_called_once_with( + request_context, + workflow_run_id="run-1", ) - with app.test_request_context(f"/console/api/workflow/{run_id}/pause-details", method="GET"): - handler = inspect.unwrap(workflow_run_module.ConsoleWorkflowPauseDetailsApi.get) - response, status = handler( - workflow_run_module.ConsoleWorkflowPauseDetailsApi(), - tenant_id, - workflow_run_id=run_id, - ) + +def test_pause_details_returns_empty_response_for_non_paused_run(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: + workflow_runs = Mock() + workflow_runs.get_pause_details.return_value = WorkflowRunPauseDetails(paused_at=None, paused_nodes=()) + _mock_application_services(monkeypatch, workflow_runs) + + api = workflow_run_module.ConsoleWorkflowPauseDetailsApi() + handler = unwrap(api.get) + with app.test_request_context("/console/api/workflow/run-1/pause-details", method="GET"): + response, status = handler(api, _request_context(), workflow_run_id="run-1") assert status == 200 assert response == {"paused_at": None, "paused_nodes": []} diff --git a/api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py b/api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py index c1176d4155b..a9686f0cccf 100644 --- a/api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py +++ b/api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py @@ -3,15 +3,20 @@ from __future__ import annotations import json from datetime import UTC, datetime from inspect import unwrap +from types import SimpleNamespace from typing import Any +from unittest.mock import Mock import pytest from flask import Flask from flask_restx import marshal from sqlalchemy.orm import Session +from controllers.common.errors import NotFoundError from controllers.console.app import workflow_run as workflow_run_module +from extensions.ext_database import db from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus +from machinery.context import RequestContext from models import Account, App, AppMode from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom from models.model import IconType @@ -58,6 +63,23 @@ def _app() -> App: ) +def _request_context(*, workspace_id: str = "tenant-1") -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id="account-1", + active_workspace_id=workspace_id, + ) + + +def _mock_application_services(monkeypatch: pytest.MonkeyPatch, workflow_runs: Mock) -> None: + monkeypatch.setattr( + workflow_run_module, + "application_services", + lambda: SimpleNamespace(workflow_runs=workflow_runs), + ) + + def _workflow_run_summary(session: Session, **overrides: object) -> WorkflowRun: created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC) workflow_run = WorkflowRun( @@ -129,17 +151,15 @@ def test_workflow_run_list_returns_frontend_history_contract( ) -> None: _account(sqlite_session) workflow_run = _workflow_run_summary(sqlite_session) - - class WorkflowRunService: - def get_paginate_workflow_runs(self, **_kwargs): - return { - "limit": 10, - "has_more": False, - "data": [workflow_run], - } - - monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService) - monkeypatch.setattr(workflow_run_module.db, "session", sqlite_session) + workflow_runs = Mock() + workflow_runs.get_paginate_workflow_runs.return_value = { + "limit": 10, + "has_more": False, + "data": [workflow_run], + } + _mock_application_services(monkeypatch, workflow_runs) + monkeypatch.setattr(db, "session", sqlite_session) + request_context = _request_context() api = workflow_run_module.WorkflowRunListApi() handler = unwrap(api.get) @@ -148,6 +168,7 @@ def test_workflow_run_list_returns_frontend_history_contract( payload = handler( api, workflow_run_module.WorkflowRunListQuery(limit=10), + request_context, app_model=_app(), ) @@ -168,6 +189,12 @@ def test_workflow_run_list_returns_frontend_history_contract( "exceptions_count": 0, "retry_index": 0, } + workflow_runs.get_paginate_workflow_runs.assert_called_once_with( + request_context, + app_id="app-1", + args={"limit": 10}, + triggered_from=WorkflowRunTriggeredFrom.DEBUGGING, + ) def test_advanced_chat_workflow_run_list_keeps_message_fields( @@ -179,17 +206,15 @@ def test_advanced_chat_workflow_run_list_keeps_message_fields( conversation_id="conversation-1", message_id="message-1", ) - - class WorkflowRunService: - def get_paginate_advanced_chat_workflow_runs(self, **_kwargs): - return { - "limit": 1, - "has_more": True, - "data": [workflow_run], - } - - monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService) - monkeypatch.setattr(workflow_run_module.db, "session", sqlite_session) + workflow_runs = Mock() + workflow_runs.get_paginate_advanced_chat_workflow_runs.return_value = { + "limit": 1, + "has_more": True, + "data": [workflow_run], + } + _mock_application_services(monkeypatch, workflow_runs) + monkeypatch.setattr(db, "session", sqlite_session) + request_context = _request_context() api = workflow_run_module.AdvancedChatAppWorkflowRunListApi() handler = unwrap(api.get) @@ -198,6 +223,7 @@ def test_advanced_chat_workflow_run_list_keeps_message_fields( payload = handler( api, workflow_run_module.WorkflowRunListQuery(limit=1), + request_context, app_model=_app(), ) @@ -205,6 +231,52 @@ def test_advanced_chat_workflow_run_list_keeps_message_fields( assert response["data"][0]["conversation_id"] == "conversation-1" assert response["data"][0]["message_id"] == "message-1" + workflow_runs.get_paginate_advanced_chat_workflow_runs.assert_called_once_with( + request_context, + app_id="app-1", + args={"limit": 1}, + triggered_from=WorkflowRunTriggeredFrom.DEBUGGING, + ) + + +def test_workflow_run_count_passes_filters_to_application_service(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: + workflow_runs = Mock() + workflow_runs.get_workflow_runs_count.return_value = { + "total": 2, + "running": 0, + "succeeded": 2, + "failed": 0, + "stopped": 0, + "partial-succeeded": 0, + } + _mock_application_services(monkeypatch, workflow_runs) + request_context = _request_context() + query = workflow_run_module.WorkflowRunCountQuery( + status="succeeded", + time_range="7d", + triggered_from="app-run", + ) + + api = workflow_run_module.WorkflowRunCountApi() + handler = unwrap(api.get) + with app.test_request_context("/apps/app-1/workflow-runs/count", method="GET"): + payload = handler(api, query, request_context, app_model=_app()) + + assert payload == { + "total": 2, + "running": 0, + "succeeded": 2, + "failed": 0, + "stopped": 0, + "partial_succeeded": 0, + } + workflow_runs.get_workflow_runs_count.assert_called_once_with( + request_context, + app_id="app-1", + status="succeeded", + time_range="7d", + triggered_from=WorkflowRunTriggeredFrom.APP_RUN, + ) def test_workflow_run_detail_returns_frontend_detail_contract( @@ -212,19 +284,17 @@ def test_workflow_run_detail_returns_frontend_detail_contract( ) -> None: _account(sqlite_session) workflow_run = _workflow_run_summary(sqlite_session) - - class WorkflowRunService: - def get_workflow_run(self, **_kwargs): - return workflow_run - - monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService) - monkeypatch.setattr(workflow_run_module.db, "session", sqlite_session) + workflow_runs = Mock() + workflow_runs.get_workflow_run.return_value = workflow_run + _mock_application_services(monkeypatch, workflow_runs) + monkeypatch.setattr(db, "session", sqlite_session) + request_context = _request_context() api = workflow_run_module.WorkflowRunDetailApi() handler = unwrap(api.get) with app.test_request_context("/apps/app-1/workflow-runs/run-1", method="GET"): - payload = handler(api, app_model=_app(), run_id="run-1") + payload = handler(api, request_context, app_model=_app(), run_id="run-1") response = _serialize_200_response(api.get, payload) @@ -246,26 +316,48 @@ def test_workflow_run_detail_returns_frontend_detail_contract( "finished_at": 1767323045, "exceptions_count": 0, } + workflow_runs.get_workflow_run.assert_called_once_with( + request_context, + app_id="app-1", + run_id="run-1", + ) + + +def test_workflow_run_detail_maps_missing_run_to_not_found(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: + workflow_runs = Mock() + workflow_runs.get_workflow_run.return_value = None + _mock_application_services(monkeypatch, workflow_runs) + request_context = _request_context(workspace_id="tenant-2") + api = workflow_run_module.WorkflowRunDetailApi() + handler = unwrap(api.get) + + with app.test_request_context("/apps/app-1/workflow-runs/run-1", method="GET"): + with pytest.raises(NotFoundError, match="Workflow run not found"): + handler(api, request_context, app_model=_app(), run_id="run-1") + + workflow_runs.get_workflow_run.assert_called_once_with( + request_context, + app_id="app-1", + run_id="run-1", + ) def test_workflow_run_node_executions_return_frontend_trace_contract( app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session ) -> None: - account = _account(sqlite_session) + _account(sqlite_session) execution = _workflow_run_node_execution(sqlite_session) - - class WorkflowRunService: - def get_workflow_run_node_executions(self, **_kwargs): - return [execution] - - monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService) - monkeypatch.setattr(workflow_run_module.db, "session", sqlite_session) + workflow_runs = Mock() + workflow_runs.get_workflow_run_node_executions.return_value = [execution] + _mock_application_services(monkeypatch, workflow_runs) + monkeypatch.setattr(db, "session", sqlite_session) + request_context = _request_context() api = workflow_run_module.WorkflowRunNodeExecutionListApi() handler = unwrap(api.get) with app.test_request_context("/apps/app-1/workflow-runs/run-1/node-executions", method="GET"): - payload = handler(api, account, app_model=_app(), run_id="run-1") + payload = handler(api, request_context, app_model=_app(), run_id="run-1") response = _serialize_200_response(api.get, payload) @@ -297,3 +389,8 @@ def test_workflow_run_node_executions_return_frontend_trace_contract( } ] } + workflow_runs.get_workflow_run_node_executions.assert_called_once_with( + request_context, + app_id="app-1", + run_id="run-1", + ) diff --git a/api/tests/unit_tests/extensions/test_ext_application_services.py b/api/tests/unit_tests/extensions/test_ext_application_services.py index 39f8f2c10f6..4eb8a0ee564 100644 --- a/api/tests/unit_tests/extensions/test_ext_application_services.py +++ b/api/tests/unit_tests/extensions/test_ext_application_services.py @@ -29,6 +29,7 @@ from repositories.account_oauth_repository import ( ) from repositories.account_repository import SQLAlchemyAccountRepository from repositories.app_site_command_repository import AppSiteCommandRepository +from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from repositories.workflow_run_archive_repository import WorkflowRunArchiveBundleQueryRepository from services import account_forgot_password_service, recommended_app_catalog_gateway from services.account_adapters import ( @@ -68,6 +69,7 @@ from services.retention.workflow_run.archive_download_task_cache import Workflow from services.retention.workflow_run.archive_log_service import WorkflowRunArchiveService from services.tag_application_service import TagApplicationService from services.webapp_access_query_service import WebAppAccessUnavailableError +from services.workflow_run_service import WorkflowRunService from services.workflow_statistic_query_service import WorkflowStatisticQueryService from tests.unit_tests.config_override import apply_config_overrides @@ -261,6 +263,22 @@ def test_build_application_services_wires_app_site_boundary( assert services.app_sites._sites._session_factory is sqlite_session_factory +def test_build_application_services_wires_workflow_run_service( + sqlite_session_factory: sessionmaker[Session], +) -> None: + services = ext_application_services.build_application_services( + database_client=sqlite_session_factory, + deployment_edition=DeploymentEdition.COMMUNITY, + initialization_password="", + redis=MagicMock(spec=RedisClientWrapper), + ) + + workflow_runs = services.workflow_runs + assert isinstance(workflow_runs, WorkflowRunService) + assert isinstance(workflow_runs._workflow_runs, DifyAPISQLAlchemyWorkflowRunRepository) + assert workflow_runs._workflow_runs._session_maker is sqlite_session_factory + + def test_build_application_services_wires_billing_service( sqlite_session: Session, sqlite_session_factory: sessionmaker[Session], diff --git a/api/tests/unit_tests/repositories/test_sqlalchemy_api_workflow_run_repository.py b/api/tests/unit_tests/repositories/test_sqlalchemy_api_workflow_run_repository.py index 76383d88b9d..0606ccd23c8 100644 --- a/api/tests/unit_tests/repositories/test_sqlalchemy_api_workflow_run_repository.py +++ b/api/tests/unit_tests/repositories/test_sqlalchemy_api_workflow_run_repository.py @@ -2,19 +2,25 @@ from __future__ import annotations import logging from datetime import UTC, datetime +from decimal import Decimal from unittest.mock import Mock, patch import pytest -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, sessionmaker from core.workflow.nodes.human_input.entities import FormDefinition, ParagraphInputConfig, UserActionConfig from core.workflow.nodes.human_input.enums import FormInputType from core.workflow.nodes.human_input.pause_reason import HumanInputRequired from graphon.entities.pause_reason import HitlRequired, PauseReasonType +from graphon.enums import WorkflowExecutionStatus, WorkflowType +from models import Message +from models.enums import ConversationFromSource, CreatorUserRole, WorkflowRunTriggeredFrom from models.human_input import HumanInputForm, HumanInputFormRecipient, RecipientType -from models.workflow import WorkflowPause, WorkflowPauseReason +from models.workflow import WorkflowPause, WorkflowPauseReason, WorkflowRun from repositories.sqlalchemy_api_workflow_run_repository import ( DifyAPISQLAlchemyWorkflowRunRepository, + WorkflowRunMessageRef, + WorkflowRunPauseRecord, _build_human_input_required_reason, _PrivateWorkflowPauseEntity, ) @@ -159,6 +165,133 @@ def test_private_workflow_pause_entity_preserves_list_shaped_pause_reasons() -> assert result == pause_reasons +def _message(*, message_id: str, app_id: str, workflow_run_id: str, conversation_id: str) -> Message: + message = Message( + app_id=app_id, + conversation_id=conversation_id, + query="query", + message={"role": "user", "content": "query"}, + answer="answer", + message_unit_price=Decimal("0.0001"), + answer_unit_price=Decimal("0.0001"), + currency="USD", + from_source=ConversationFromSource.API, + ) + message.id = message_id + message._inputs = {} + message.workflow_run_id = workflow_run_id + return message + + +def _workflow_run(*, run_id: str, tenant_id: str, status: WorkflowExecutionStatus) -> WorkflowRun: + return WorkflowRun( + id=run_id, + tenant_id=tenant_id, + app_id="app-1", + workflow_id="workflow-1", + type=WorkflowType.WORKFLOW, + triggered_from=WorkflowRunTriggeredFrom.DEBUGGING, + version="1", + graph="{}", + inputs="{}", + status=status, + created_by_role=CreatorUserRole.ACCOUNT, + created_by="account-1", + ) + + +def test_get_message_refs_filters_by_app_and_returns_lightweight_records( + sqlite_session: Session, + sqlite_session_factory: sessionmaker[Session], +) -> None: + sqlite_session.add_all( + [ + _message(message_id="msg-1", app_id="app-1", workflow_run_id="run-1", conversation_id="conv-1"), + _message(message_id="msg-2", app_id="app-2", workflow_run_id="run-2", conversation_id="conv-2"), + ] + ) + sqlite_session.commit() + repository = DifyAPISQLAlchemyWorkflowRunRepository(session_maker=sqlite_session_factory) + + result = repository.get_message_refs( + app_id="app-1", + workflow_run_ids=["run-1", "run-2"], + ) + + assert result == { + "run-1": WorkflowRunMessageRef(message_id="msg-1", conversation_id="conv-1"), + } + + +def test_get_pause_record_scopes_the_workflow_run_to_the_workspace( + sqlite_session: Session, + sqlite_session_factory: sessionmaker[Session], +) -> None: + sqlite_session.add( + _workflow_run( + run_id="run-1", + tenant_id="tenant-1", + status=WorkflowExecutionStatus.SUCCEEDED, + ) + ) + sqlite_session.commit() + repository = DifyAPISQLAlchemyWorkflowRunRepository(session_maker=sqlite_session_factory) + + assert repository.get_pause_record(workspace_id="tenant-2", workflow_run_id="run-1") is None + assert repository.get_pause_record( + workspace_id="tenant-1", + workflow_run_id="run-1", + ) == WorkflowRunPauseRecord( + status=WorkflowExecutionStatus.SUCCEEDED, + paused_at=None, + reasons=(), + form_tokens={}, + ) + + +def test_get_pause_record_loads_reasons_and_tokens_in_one_repository_call( + sqlite_session: Session, + sqlite_session_factory: sessionmaker[Session], +) -> None: + workflow_run = _workflow_run( + run_id="run-1", + tenant_id="tenant-1", + status=WorkflowExecutionStatus.PAUSED, + ) + pause = WorkflowPause( + workflow_id=workflow_run.workflow_id, + workflow_run_id=workflow_run.id, + state_object_key="pause-state", + ) + pause.id = "pause-1" + reason = WorkflowPauseReason( + pause_id=pause.id, + type_=PauseReasonType.HITL_REQUIRED, + form_id="form-1", + node_id="node-1", + ) + recipient = HumanInputFormRecipient( + form_id="form-1", + delivery_id="delivery-1", + recipient_type=RecipientType.CONSOLE, + recipient_payload="{}", + access_token="form-token", + ) + sqlite_session.add_all([workflow_run, pause, reason, recipient]) + sqlite_session.commit() + repository = DifyAPISQLAlchemyWorkflowRunRepository(session_maker=sqlite_session_factory) + + result = repository.get_pause_record(workspace_id="tenant-1", workflow_run_id="run-1") + + assert result is not None + assert result.status == WorkflowExecutionStatus.PAUSED + assert result.paused_at == pause.created_at + assert len(result.reasons) == 1 + assert isinstance(result.reasons[0], HumanInputRequired) + assert result.reasons[0].form_id == "form-1" + assert result.form_tokens == {"form-1": "form-token"} + + def test_delete_pause_model_deletes_record_when_state_object_delete_fails( caplog: pytest.LogCaptureFixture, ) -> None: diff --git a/api/tests/unit_tests/services/test_workflow_run_service.py b/api/tests/unit_tests/services/test_workflow_run_service.py index c1aa0fd9cea..b71829dcf7c 100644 --- a/api/tests/unit_tests/services/test_workflow_run_service.py +++ b/api/tests/unit_tests/services/test_workflow_run_service.py @@ -1,62 +1,38 @@ -"""Workflow-run service tests with real SQLite-bound session factories.""" +"""Unit tests for the Console workflow-run application service.""" -from decimal import Decimal from types import SimpleNamespace -from typing import Any from unittest.mock import MagicMock import pytest -from sqlalchemy import Engine, event -from sqlalchemy.orm import Session, sessionmaker from graphon.enums import WorkflowExecutionStatus -from models import Account, App, EndUser, Message, WorkflowRun, WorkflowRunTriggeredFrom, WorkflowType -from models.account import Tenant -from models.enums import ConversationFromSource, CreatorUserRole, EndUserType -from models.model import AppMode +from machinery.context import RequestContext +from models import WorkflowRun, WorkflowRunTriggeredFrom, WorkflowType +from models.enums import CreatorUserRole +from repositories.sqlalchemy_api_workflow_run_repository import WorkflowRunMessageRef from services import workflow_run_service as service_module from services.workflow_run_service import WorkflowRunService @pytest.fixture -def repository_factory_mocks(monkeypatch: pytest.MonkeyPatch) -> tuple[MagicMock, MagicMock, Any]: - node_repo = MagicMock() - workflow_run_repo = MagicMock() - factory = SimpleNamespace( - create_api_workflow_node_execution_repository=MagicMock(return_value=node_repo), - create_api_workflow_run_repository=MagicMock(return_value=workflow_run_repo), - ) - monkeypatch.setattr(service_module, "DifyAPIRepositoryFactory", factory) - return node_repo, workflow_run_repo, factory +def service_dependencies() -> tuple[MagicMock, MagicMock]: + return MagicMock(), MagicMock() -def _app_model(*, app_id: str = "app-1", tenant_id: str = "tenant-1") -> App: - return App( - id=app_id, - tenant_id=tenant_id, - name="Workflow App", - mode=AppMode.ADVANCED_CHAT, - enable_site=False, - enable_api=False, +def _service(dependencies: tuple[MagicMock, MagicMock]) -> WorkflowRunService: + node_executions, workflow_runs = dependencies + return WorkflowRunService( + workflow_runs=workflow_runs, + node_executions=node_executions, ) -def _account(*, account_id: str = "account-1", current_tenant_id: str | None = "tenant-1") -> Account: - account = Account(name="Workflow User", email=f"{account_id}@example.com") - account.id = account_id - if current_tenant_id is not None: - tenant = Tenant(name="Workflow Tenant") - tenant.id = current_tenant_id - account._current_tenant = tenant - return account - - -def _end_user(*, end_user_id: str = "end-user-1", tenant_id: str = "tenant-1") -> EndUser: - return EndUser( - id=end_user_id, - tenant_id=tenant_id, - type=EndUserType.SERVICE_API, - session_id=f"session-{end_user_id}", +def _request_context(*, workspace_id: str = "tenant-1") -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id="account-1", + active_workspace_id=workspace_id, ) @@ -79,87 +55,37 @@ def _workflow_run( ) -def _message(*, message_id: str, workflow_run_id: str, conversation_id: str) -> Message: - message = Message( - app_id="app-1", - conversation_id=conversation_id, - query="query", - message={"role": "user", "content": "query"}, - answer="answer", - message_unit_price=Decimal("0.0001"), - answer_unit_price=Decimal("0.0001"), - currency="USD", - from_source=ConversationFromSource.API, - ) - message.id = message_id - message._inputs = {} - message.workflow_run_id = workflow_run_id - return message +def test_init_keeps_injected_dependencies( + service_dependencies: tuple[MagicMock, MagicMock], +) -> None: + node_executions, workflow_runs = service_dependencies + service = _service(service_dependencies) -class TestWorkflowRunServiceInitialization: - def test___init___should_create_sessionmaker_from_db_engine_when_session_factory_missing( - self, - monkeypatch: pytest.MonkeyPatch, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - sqlite_engine: Engine, - ) -> None: - monkeypatch.setattr(service_module, "db", SimpleNamespace(engine=sqlite_engine)) - - service = WorkflowRunService() - - assert isinstance(service._session_factory, sessionmaker) - assert service._session_factory.kw["bind"] is sqlite_engine - assert service._session_factory.kw["expire_on_commit"] is False - - def test___init___should_create_sessionmaker_when_engine_is_provided( - self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - sqlite_engine: Engine, - ) -> None: - service = WorkflowRunService(session_factory=sqlite_engine) - - assert isinstance(service._session_factory, sessionmaker) - assert service._session_factory.kw["bind"] is sqlite_engine - assert service._session_factory.kw["expire_on_commit"] is False - - def test___init___should_keep_provided_sessionmaker_and_create_repositories( - self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - sqlite_session_factory: sessionmaker[Session], - ) -> None: - node_repo, workflow_run_repo, factory = repository_factory_mocks - - service = WorkflowRunService(session_factory=sqlite_session_factory) - - assert service._session_factory is sqlite_session_factory - assert service._node_execution_service_repo is node_repo - assert service._workflow_run_repo is workflow_run_repo - factory.create_api_workflow_node_execution_repository.assert_called_once_with(sqlite_session_factory) - factory.create_api_workflow_run_repository.assert_called_once_with(sqlite_session_factory) + assert service._workflow_runs is workflow_runs + assert service._node_executions is node_executions class TestWorkflowRunServiceQueries: def test_get_paginate_workflow_runs_should_forward_filters_and_parse_limit( self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - sqlite_session_factory: sessionmaker[Session], + service_dependencies: tuple[MagicMock, MagicMock], ) -> None: - _, workflow_run_repo, _ = repository_factory_mocks - service = WorkflowRunService(session_factory=sqlite_session_factory) - app_model = _app_model(tenant_id="tenant-1", app_id="app-1") + _, workflow_runs = service_dependencies + service = _service(service_dependencies) expected = MagicMock(name="pagination") - workflow_run_repo.get_paginated_workflow_runs.return_value = expected + workflow_runs.get_paginated_workflow_runs.return_value = expected args = {"limit": "7", "last_id": "last-1", "status": "succeeded"} result = service.get_paginate_workflow_runs( - app_model=app_model, + _request_context(workspace_id="tenant-1"), + app_id="app-1", args=args, triggered_from=WorkflowRunTriggeredFrom.APP_RUN, ) assert result is expected - workflow_run_repo.get_paginated_workflow_runs.assert_called_once_with( + workflow_runs.get_paginated_workflow_runs.assert_called_once_with( tenant_id="tenant-1", app_id="app-1", triggered_from=WorkflowRunTriggeredFrom.APP_RUN, @@ -168,25 +94,26 @@ class TestWorkflowRunServiceQueries: status="succeeded", ) - @pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True) def test_get_paginate_advanced_chat_workflow_runs_should_attach_message_fields_when_message_exists( self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], + service_dependencies: tuple[MagicMock, MagicMock], monkeypatch: pytest.MonkeyPatch, - sqlite_session_factory: sessionmaker[Session], - sqlite_session: Session, ) -> None: - service = WorkflowRunService(session_factory=sqlite_session_factory) - app_model = _app_model(tenant_id="tenant-1", app_id="app-1") + _, workflow_runs = service_dependencies + service = _service(service_dependencies) run_with_message = _workflow_run(status=WorkflowExecutionStatus.RUNNING) run_without_message = _workflow_run(run_id="run-2") pagination = SimpleNamespace(data=[run_with_message, run_without_message]) monkeypatch.setattr(service, "get_paginate_workflow_runs", MagicMock(return_value=pagination)) + workflow_runs.get_message_refs.return_value = { + "run-1": WorkflowRunMessageRef(message_id="msg-1", conversation_id="conv-1") + } - sqlite_session.add(_message(message_id="msg-1", conversation_id="conv-1", workflow_run_id="run-1")) - sqlite_session.commit() - - result = service.get_paginate_advanced_chat_workflow_runs(app_model=app_model, args={"limit": "2"}) + result = service.get_paginate_advanced_chat_workflow_runs( + _request_context(), + app_id="app-1", + args={"limit": "2"}, + ) assert result is pagination assert len(result.data) == 2 @@ -195,56 +122,24 @@ class TestWorkflowRunServiceQueries: assert result.data[0].status == "running" assert not hasattr(result.data[1], "message_id") assert result.data[1].id == "run-2" - - @pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True) - def test_get_paginate_advanced_chat_workflow_runs_batch_loads_messages_without_n_plus_one( - self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - monkeypatch: pytest.MonkeyPatch, - sqlite_session_factory: sessionmaker[Session], - sqlite_session: Session, - ) -> None: - """Messages must load with a constant query count regardless of run count. - - Previously the deprecated WorkflowRun.message property issued one query per - run (N+1); they are now batch-loaded in a single query. - """ - service = WorkflowRunService(session_factory=sqlite_session_factory) - app_model = _app_model(tenant_id="tenant-1", app_id="app-1") - runs = [_workflow_run(run_id=f"run-{i}") for i in range(5)] - pagination = SimpleNamespace(data=runs) - monkeypatch.setattr(service, "get_paginate_workflow_runs", MagicMock(return_value=pagination)) - - message_query_count = 0 - - def count_message_query(*_args: object) -> None: - nonlocal message_query_count - message_query_count += 1 - - engine = sqlite_session.get_bind() - event.listen(engine, "before_cursor_execute", count_message_query) - try: - service.get_paginate_advanced_chat_workflow_runs(app_model=app_model, args={}) - finally: - event.remove(engine, "before_cursor_execute", count_message_query) - assert all(not hasattr(run, "message_id") for run in runs) - assert message_query_count == 1 + workflow_runs.get_message_refs.assert_called_once_with( + app_id="app-1", + workflow_run_ids=["run-1", "run-2"], + ) def test_get_workflow_run_should_delegate_to_repository_by_tenant_and_app( self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - sqlite_session_factory: sessionmaker[Session], + service_dependencies: tuple[MagicMock, MagicMock], ) -> None: - _, workflow_run_repo, _ = repository_factory_mocks - service = WorkflowRunService(session_factory=sqlite_session_factory) - app_model = _app_model(tenant_id="tenant-1", app_id="app-1") + _, workflow_runs = service_dependencies + service = _service(service_dependencies) expected = _workflow_run() - workflow_run_repo.get_workflow_run_by_id.return_value = expected + workflow_runs.get_workflow_run_by_id.return_value = expected - result = service.get_workflow_run(app_model=app_model, run_id="run-1") + result = service.get_workflow_run(_request_context(), app_id="app-1", run_id="run-1") assert result is expected - workflow_run_repo.get_workflow_run_by_id.assert_called_once_with( + workflow_runs.get_workflow_run_by_id.assert_called_once_with( tenant_id="tenant-1", app_id="app-1", run_id="run-1", @@ -252,24 +147,23 @@ class TestWorkflowRunServiceQueries: def test_get_workflow_runs_count_should_forward_optional_filters( self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - sqlite_session_factory: sessionmaker[Session], + service_dependencies: tuple[MagicMock, MagicMock], ) -> None: - _, workflow_run_repo, _ = repository_factory_mocks - service = WorkflowRunService(session_factory=sqlite_session_factory) - app_model = _app_model(tenant_id="tenant-1", app_id="app-1") + _, workflow_runs = service_dependencies + service = _service(service_dependencies) expected = {"total": 3, "succeeded": 2} - workflow_run_repo.get_workflow_runs_count.return_value = expected + workflow_runs.get_workflow_runs_count.return_value = expected result = service.get_workflow_runs_count( - app_model=app_model, + _request_context(), + app_id="app-1", status="succeeded", time_range="7d", triggered_from=WorkflowRunTriggeredFrom.APP_RUN, ) assert result == expected - workflow_run_repo.get_workflow_runs_count.assert_called_once_with( + workflow_runs.get_workflow_runs_count.assert_called_once_with( tenant_id="tenant-1", app_id="app-1", triggered_from=WorkflowRunTriggeredFrom.APP_RUN, @@ -279,83 +173,44 @@ class TestWorkflowRunServiceQueries: def test_get_workflow_run_node_executions_should_return_empty_list_when_run_not_found( self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], + service_dependencies: tuple[MagicMock, MagicMock], monkeypatch: pytest.MonkeyPatch, - sqlite_session_factory: sessionmaker[Session], ) -> None: - service = WorkflowRunService(session_factory=sqlite_session_factory) + service = _service(service_dependencies) monkeypatch.setattr(service, "get_workflow_run", MagicMock(return_value=None)) - app_model = _app_model(app_id="app-1") - user = _account(current_tenant_id="tenant-1") - result = service.get_workflow_run_node_executions(app_model=app_model, run_id="run-1", user=user) + result = service.get_workflow_run_node_executions( + _request_context(), + app_id="app-1", + run_id="run-1", + ) assert result == [] - def test_get_workflow_run_node_executions_should_use_end_user_tenant_id( + def test_get_workflow_run_node_executions_should_use_request_workspace( self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], + service_dependencies: tuple[MagicMock, MagicMock], monkeypatch: pytest.MonkeyPatch, - sqlite_session_factory: sessionmaker[Session], ) -> None: - node_repo, _, _ = repository_factory_mocks - service = WorkflowRunService(session_factory=sqlite_session_factory) + node_executions, _ = service_dependencies + service = _service(service_dependencies) monkeypatch.setattr(service, "get_workflow_run", MagicMock(return_value=_workflow_run())) - user = _end_user(tenant_id="tenant-end-user") - app_model = _app_model(app_id="app-1") - expected_executions = [SimpleNamespace(id="exec-1")] - expected_traces = [SimpleNamespace(id="exec-1:retry:1")] - node_repo.get_executions_by_workflow_run.return_value = expected_executions - mock_assemble = MagicMock(return_value=expected_traces) - monkeypatch.setattr(service_module, "assemble_workflow_node_execution_traces", mock_assemble) - - result = service.get_workflow_run_node_executions(app_model=app_model, run_id="run-1", user=user) - - assert result == expected_traces - node_repo.get_executions_by_workflow_run.assert_called_once_with( - tenant_id="tenant-end-user", - app_id="app-1", - workflow_run_id="run-1", - ) - mock_assemble.assert_called_once_with(expected_executions, node_repo) - - def test_get_workflow_run_node_executions_should_use_account_current_tenant_id( - self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - monkeypatch: pytest.MonkeyPatch, - sqlite_session_factory: sessionmaker[Session], - ) -> None: - node_repo, _, _ = repository_factory_mocks - service = WorkflowRunService(session_factory=sqlite_session_factory) - monkeypatch.setattr(service, "get_workflow_run", MagicMock(return_value=_workflow_run())) - app_model = _app_model(app_id="app-1") - user = _account(current_tenant_id="tenant-account") expected_executions = [SimpleNamespace(id="exec-1"), SimpleNamespace(id="exec-2")] expected_traces = [SimpleNamespace(id="exec-1:retry:1"), SimpleNamespace(id="exec-1")] - node_repo.get_executions_by_workflow_run.return_value = expected_executions + node_executions.get_executions_by_workflow_run.return_value = expected_executions mock_assemble = MagicMock(return_value=expected_traces) monkeypatch.setattr(service_module, "assemble_workflow_node_execution_traces", mock_assemble) - result = service.get_workflow_run_node_executions(app_model=app_model, run_id="run-1", user=user) + result = service.get_workflow_run_node_executions( + _request_context(workspace_id="tenant-context"), + app_id="app-1", + run_id="run-1", + ) assert result == expected_traces - node_repo.get_executions_by_workflow_run.assert_called_once_with( - tenant_id="tenant-account", + node_executions.get_executions_by_workflow_run.assert_called_once_with( + tenant_id="tenant-context", app_id="app-1", workflow_run_id="run-1", ) - mock_assemble.assert_called_once_with(expected_executions, node_repo) - - def test_get_workflow_run_node_executions_should_raise_when_resolved_tenant_id_is_none( - self, - repository_factory_mocks: tuple[MagicMock, MagicMock, Any], - monkeypatch: pytest.MonkeyPatch, - sqlite_session_factory: sessionmaker[Session], - ) -> None: - service = WorkflowRunService(session_factory=sqlite_session_factory) - monkeypatch.setattr(service, "get_workflow_run", MagicMock(return_value=_workflow_run())) - app_model = _app_model(app_id="app-1") - user = _account(current_tenant_id=None) - - with pytest.raises(ValueError, match="tenant_id cannot be None"): - service.get_workflow_run_node_executions(app_model=app_model, run_id="run-1", user=user) + mock_assemble.assert_called_once_with(expected_executions, node_executions) diff --git a/api/tests/unit_tests/services/test_workflow_run_service_pause.py b/api/tests/unit_tests/services/test_workflow_run_service_pause.py index 5b24cfd8a66..3552cbd7c9d 100644 --- a/api/tests/unit_tests/services/test_workflow_run_service_pause.py +++ b/api/tests/unit_tests/services/test_workflow_run_service_pause.py @@ -1,57 +1,112 @@ -"""Tests for the session lifecycle owned by ``WorkflowRunService``.""" +"""Tests for Console workflow pause details.""" -from unittest.mock import create_autospec, patch +from datetime import datetime +from unittest.mock import MagicMock import pytest -from sqlalchemy import Engine, text -from sqlalchemy.orm import Session, sessionmaker -from repositories.api_workflow_run_repository import APIWorkflowRunRepository -from services.workflow_run_service import WorkflowRunService +from core.workflow.nodes.human_input.pause_reason import HumanInputRequired +from graphon.entities.pause_reason import SchedulingPause +from graphon.enums import WorkflowExecutionStatus +from machinery.context import RequestContext +from repositories.sqlalchemy_api_workflow_run_repository import WorkflowRunPauseRecord +from services.workflow_run_service import ( + WorkflowRunPauseDetails, + WorkflowRunPausedNode, + WorkflowRunService, +) @pytest.fixture -def sqlite_session_factory(sqlite_engine: Engine) -> sessionmaker[Session]: - """Return a real factory whose sessions are bound to the isolated SQLite engine.""" - return sessionmaker(bind=sqlite_engine, expire_on_commit=False) +def workflow_runs() -> MagicMock: + return MagicMock() -@pytest.fixture -def workflow_run_repository(): - """Keep the repository boundary mocked while exercising real session construction.""" - return create_autospec(APIWorkflowRunRepository) +def _service(workflow_runs: MagicMock) -> WorkflowRunService: + return WorkflowRunService( + workflow_runs=workflow_runs, + node_executions=MagicMock(), + ) -def test_init_with_session_factory( - sqlite_session_factory: sessionmaker[Session], workflow_run_repository: APIWorkflowRunRepository -) -> None: - with patch("services.workflow_run_service.DifyAPIRepositoryFactory", autospec=True) as repository_factory: - repository_factory.create_api_workflow_run_repository.return_value = workflow_run_repository - - service = WorkflowRunService(sqlite_session_factory) - - assert service._session_factory is sqlite_session_factory - repository_factory.create_api_workflow_run_repository.assert_called_once_with(sqlite_session_factory) - with service._session_factory() as session: - assert session.scalar(text("SELECT 1")) == 1 +def _request_context(*, workspace_id: str = "tenant-1") -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id="account-1", + active_workspace_id=workspace_id, + ) -def test_init_with_engine_creates_bound_session_factory( - sqlite_engine: Engine, workflow_run_repository: APIWorkflowRunRepository -) -> None: - with patch("services.workflow_run_service.DifyAPIRepositoryFactory", autospec=True) as repository_factory: - repository_factory.create_api_workflow_run_repository.return_value = workflow_run_repository +def test_get_pause_details_returns_none_when_run_is_not_found(workflow_runs: MagicMock) -> None: + workflow_runs.get_pause_record.return_value = None - service = WorkflowRunService(sqlite_engine) + result = _service(workflow_runs).get_pause_details(_request_context(), workflow_run_id="run-1") - assert service._session_factory.kw["bind"] is sqlite_engine - assert service._session_factory.kw["expire_on_commit"] is False - repository_factory.create_api_workflow_run_repository.assert_called_once_with(service._session_factory) - with service._session_factory() as session: - assert session.scalar(text("SELECT 1")) == 1 + assert result is None + workflow_runs.get_pause_record.assert_called_once_with( + workspace_id="tenant-1", + workflow_run_id="run-1", + ) -def test_init_with_default_repository_dependencies(sqlite_session_factory: sessionmaker[Session]) -> None: - service = WorkflowRunService(sqlite_session_factory) +def test_get_pause_details_returns_empty_details_for_non_paused_run(workflow_runs: MagicMock) -> None: + workflow_runs.get_pause_record.return_value = WorkflowRunPauseRecord( + status=WorkflowExecutionStatus.SUCCEEDED, + paused_at=None, + reasons=(), + form_tokens={}, + ) - assert service._session_factory is sqlite_session_factory + result = _service(workflow_runs).get_pause_details(_request_context(), workflow_run_id="run-1") + + assert result == WorkflowRunPauseDetails(paused_at=None, paused_nodes=()) + + +def test_get_pause_details_maps_human_input_and_token(workflow_runs: MagicMock) -> None: + reason = HumanInputRequired( + form_id="form-1", + form_content="Approve?", + node_id="node-1", + node_title="Approval", + ) + paused_at = datetime(2026, 1, 2, 3, 4, 5) + workflow_runs.get_pause_record.return_value = WorkflowRunPauseRecord( + status=WorkflowExecutionStatus.PAUSED, + paused_at=paused_at, + reasons=(reason,), + form_tokens={"form-1": "form-token"}, + ) + + result = _service(workflow_runs).get_pause_details( + _request_context(workspace_id="tenant-context"), + workflow_run_id="run-1", + ) + + assert result == WorkflowRunPauseDetails( + paused_at=paused_at, + paused_nodes=( + WorkflowRunPausedNode( + node_id="node-1", + node_title="Approval", + form_id="form-1", + form_token="form-token", + ), + ), + ) + workflow_runs.get_pause_record.assert_called_once_with( + workspace_id="tenant-context", + workflow_run_id="run-1", + ) + + +def test_get_pause_details_rejects_unsupported_pause_reason(workflow_runs: MagicMock) -> None: + workflow_runs.get_pause_record.return_value = WorkflowRunPauseRecord( + status=WorkflowExecutionStatus.PAUSED, + paused_at=None, + reasons=(SchedulingPause(message="Waiting for external input"),), + form_tokens={}, + ) + + with pytest.raises(NotImplementedError, match="Pause details do not support SchedulingPause"): + _service(workflow_runs).get_pause_details(_request_context(), workflow_run_id="run-1")