opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-08 02:43:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): extract console workflow run application service (#41336)

Browse Source
This commit is contained in:
非法操作 2026-09-04 08:16:49 +00:00 committed by GitHub
parent cb6c04637b
commit f8930aba34
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 980 additions and 906 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

297
api/controllers/console/app/workflow_run.py
View File

@ -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/<uuid:app_id>/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/<uuid:app_id>/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/<uuid:app_id>/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/<uuid:app_id>/workflow-runs/<uuid:run_id>")
@ -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/<uuid:app_id>/workflow-runs/<uuid:run_id>/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/<string:workflow_run_id>/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

11
api/extensions/ext_application_services.py
View File

@ -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(),

3
api/repositories/api_workflow_run_repository.py
View File

@ -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.

85
api/repositories/sqlalchemy_api_workflow_run_repository.py
View File

@ -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,

181
api/services/workflow_run_service.py
View File

@ -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
),
)

14
api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py
View File

@ -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,

17
api/tests/test_containers_integration_tests/core/app/layers/test_pause_state_persist_layer.py
View File

@ -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

12
api/tests/test_containers_integration_tests/repositories/test_sqlalchemy_workflow_run_cleanup_repository.py
View File

@ -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(

264
api/tests/test_containers_integration_tests/services/test_workflow_run_service.py
View File

@ -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

234
api/tests/unit_tests/controllers/console/app/test_workflow_pause_details_api.py
View File

@ -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": []}

175
api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py
View File

@ -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",
)

18
api/tests/unit_tests/extensions/test_ext_application_services.py
View File

@ -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],

137
api/tests/unit_tests/repositories/test_sqlalchemy_api_workflow_run_repository.py
View File

@ -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:

305
api/tests/unit_tests/services/test_workflow_run_service.py
View File

@ -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)

133
api/tests/unit_tests/services/test_workflow_run_service_pause.py
View File

@ -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")