diff --git a/api/tests/unit_tests/services/workflow/test_node_output_inspector_service.py b/api/tests/unit_tests/services/workflow/test_node_output_inspector_service.py index 7f720575154..2c659b2621d 100644 --- a/api/tests/unit_tests/services/workflow/test_node_output_inspector_service.py +++ b/api/tests/unit_tests/services/workflow/test_node_output_inspector_service.py @@ -1,19 +1,20 @@ """Unit tests for NodeOutputInspectorService (Stage 4 §8). -The service reads from postgres and resolves agent v2 bindings; this suite -mocks the DB session and binding resolver so we exercise the view-construction -logic without DB / network access. +The service reads persisted workflow runs and node executions while the agent +v2 binding resolver and file URL boundaries remain isolated from network I/O. """ from __future__ import annotations import json +from collections.abc import Sequence from datetime import UTC, datetime from types import SimpleNamespace -from typing import Any +from typing import Any, Protocol from unittest.mock import MagicMock, patch import pytest +from sqlalchemy.orm import Session from core.workflow.file_reference import build_file_reference from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus @@ -22,7 +23,13 @@ from models.agent_config_entities import ( DeclaredOutputConfig, DeclaredOutputType, ) -from models.enums import WorkflowRunTriggeredFrom +from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom +from models.workflow import ( + WorkflowNodeExecutionModel, + WorkflowNodeExecutionTriggeredFrom, + WorkflowRun, + WorkflowType, +) from services.workflow.node_output_inspector_service import ( NodeOutputInspectorError, NodeOutputInspectorService, @@ -49,15 +56,28 @@ def _workflow_run( triggered_from: WorkflowRunTriggeredFrom = WorkflowRunTriggeredFrom.DEBUGGING, status: WorkflowExecutionStatus = WorkflowExecutionStatus.RUNNING, nodes: list[dict[str, Any]] | None = None, -): - return SimpleNamespace( + graph: str | None = None, +) -> WorkflowRun: + return WorkflowRun( id=run_id, workflow_id=workflow_id, tenant_id=tenant_id, app_id=app_id, + type=WorkflowType.WORKFLOW, triggered_from=triggered_from, + version="1", status=status, - graph=json.dumps({"nodes": nodes or []}), + graph=graph if graph is not None else json.dumps({"nodes": nodes or []}), + inputs="{}", + outputs="{}", + error=None, + elapsed_time=0, + total_tokens=0, + total_steps=0, + created_by_role=CreatorUserRole.ACCOUNT, + created_by="account-1", + created_at=datetime(2024, 1, 1, tzinfo=UTC), + finished_at=None, ) @@ -72,16 +92,33 @@ def _execution( index: int = 1, created_at: datetime | None = None, finished_at: datetime | None = None, -): - return SimpleNamespace( + tenant_id: str = "tenant-1", + app_id: str = "app-1", + workflow_id: str = "workflow-1", + workflow_run_id: str = "run-1", +) -> WorkflowNodeExecutionModel: + return WorkflowNodeExecutionModel( + tenant_id=tenant_id, + app_id=app_id, + workflow_id=workflow_id, + triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN, + workflow_run_id=workflow_run_id, + predecessor_node_id=None, + node_execution_id=None, node_id=node_id, node_type=node_type, title=title or node_id, status=status, + inputs=None, + process_data=None, outputs=json.dumps(outputs) if outputs is not None else None, + error=None, + elapsed_time=0, execution_metadata=json.dumps(execution_metadata) if execution_metadata is not None else None, index=index, created_at=created_at or datetime.now(UTC), + created_by_role=CreatorUserRole.ACCOUNT, + created_by="account-1", finished_at=finished_at, ) @@ -100,17 +137,32 @@ def _non_agent_node(*, node_id: str = "tool-node-1", node_type: str = "tool", ti } -def _mock_session( - *, - workflow_run: SimpleNamespace | None, - executions: list[SimpleNamespace] | None = None, -): - """Build a mock DB session with the configured rows.""" - executions = executions or [] - session = MagicMock() - session.scalar.return_value = workflow_run - session.scalars.return_value.all.return_value = executions - return session +class SessionFor(Protocol): + def __call__( + self, + *, + workflow_run: WorkflowRun | None, + executions: Sequence[WorkflowNodeExecutionModel] = (), + ) -> Session: ... + + +@pytest.fixture +def session_for(sqlite_session: Session) -> SessionFor: + """Persist one scenario in the shared SQLite session.""" + + def create_session( + *, + workflow_run: WorkflowRun | None, + executions: Sequence[WorkflowNodeExecutionModel] = (), + ) -> Session: + if workflow_run is not None: + sqlite_session.add(workflow_run) + sqlite_session.add_all(executions) + sqlite_session.commit() + sqlite_session.expunge_all() + return sqlite_session + + return create_session def _stub_binding_resolver(*, declared_outputs: list[DeclaredOutputConfig]): @@ -138,55 +190,56 @@ def _make_service(declared_outputs: list[DeclaredOutputConfig] | None = None) -> # ────────────────────────────────────────────────────────────────────────────── -def test_snapshot_404_when_workflow_run_missing(): +def test_snapshot_404_when_workflow_run_missing(session_for: SessionFor) -> None: service = _make_service() - session = _mock_session(workflow_run=None) + other_tenant_run = _workflow_run(run_id="missing", tenant_id="tenant-2") + session = session_for(workflow_run=other_tenant_run) with pytest.raises(NodeOutputInspectorError) as exc: service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="missing", session=session) assert exc.value.code == "workflow_run_not_found" -def test_snapshot_accepts_published_run_d1_lifted(): +def test_snapshot_accepts_published_run_d1_lifted(session_for: SessionFor) -> None: """D-1 was lifted 2026-05-26: any ``triggered_from`` is now accepted.""" service = _make_service() run = _workflow_run( nodes=[_agent_v2_node(node_id="agent-1")], triggered_from=WorkflowRunTriggeredFrom.APP_RUN, ) - session = _mock_session(workflow_run=run, executions=[]) + session = session_for(workflow_run=run, executions=[]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert snapshot.workflow_run_id == "run-1" assert [n.node_id for n in snapshot.node_outputs] == ["agent-1"] -def test_snapshot_accepts_webhook_triggered_run(): +def test_snapshot_accepts_webhook_triggered_run(session_for: SessionFor) -> None: """Webhook / schedule / plugin triggers are also published-side.""" service = _make_service() run = _workflow_run( nodes=[_agent_v2_node(node_id="agent-1")], triggered_from=WorkflowRunTriggeredFrom.WEBHOOK, ) - session = _mock_session(workflow_run=run, executions=[]) + session = session_for(workflow_run=run, executions=[]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert snapshot.workflow_run_id == "run-1" -def test_node_detail_404_when_node_id_absent_from_graph(): +def test_node_detail_404_when_node_id_absent_from_graph(session_for: SessionFor) -> None: service = _make_service() run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) - session = _mock_session(workflow_run=run, executions=[]) + session = session_for(workflow_run=run, executions=[]) with pytest.raises(NodeOutputInspectorError) as exc: service.node_detail(app_model=_app_model(), workflow_run_id="run-1", node_id="ghost", session=session) assert exc.value.code == "node_not_in_workflow_run" -def test_output_preview_404_when_output_name_unknown(): +def test_output_preview_404_when_output_name_unknown(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING)], ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) ex = _execution(node_id="agent-1", outputs={"text": "hello"}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with pytest.raises(NodeOutputInspectorError) as exc: service.output_preview( app_model=_app_model(), @@ -198,10 +251,10 @@ def test_output_preview_404_when_output_name_unknown(): assert exc.value.code == "node_output_not_declared" -def test_output_preview_404_when_node_id_absent_from_graph(): +def test_output_preview_404_when_node_id_absent_from_graph(session_for: SessionFor) -> None: service = _make_service() run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) - session = _mock_session(workflow_run=run, executions=[]) + session = session_for(workflow_run=run, executions=[]) with pytest.raises(NodeOutputInspectorError) as exc: service.output_preview( app_model=_app_model(), @@ -218,12 +271,12 @@ def test_output_preview_404_when_node_id_absent_from_graph(): # ────────────────────────────────────────────────────────────────────────────── -def test_snapshot_status_pending_when_node_has_no_execution(): +def test_snapshot_status_pending_when_node_has_no_execution(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING)], ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) - session = _mock_session(workflow_run=run, executions=[]) + session = session_for(workflow_run=run, executions=[]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert len(snapshot.node_outputs) == 1 @@ -232,19 +285,19 @@ def test_snapshot_status_pending_when_node_has_no_execution(): assert node.outputs[0].status == NodeOutputStatus.PENDING -def test_snapshot_status_running(): +def test_snapshot_status_running(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING)], ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) ex = _execution(node_id="agent-1", status=WorkflowNodeExecutionStatus.RUNNING) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert snapshot.node_outputs[0].node_status == NodeStatus.RUNNING assert snapshot.node_outputs[0].outputs[0].status == NodeOutputStatus.RUNNING -def test_snapshot_status_failed_node_marks_all_outputs_failed(): +def test_snapshot_status_failed_node_marks_all_outputs_failed(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig(name="a", type=DeclaredOutputType.STRING), @@ -253,26 +306,26 @@ def test_snapshot_status_failed_node_marks_all_outputs_failed(): ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) ex = _execution(node_id="agent-1", status=WorkflowNodeExecutionStatus.FAILED) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) statuses = {o.name: o.status for o in snapshot.node_outputs[0].outputs} assert statuses == {"a": NodeOutputStatus.FAILED, "b": NodeOutputStatus.FAILED} -def test_snapshot_status_ready_when_outputs_present_and_no_failure_metadata(): +def test_snapshot_status_ready_when_outputs_present_and_no_failure_metadata(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING)], ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) ex = _execution(node_id="agent-1", outputs={"text": "hello"}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) output = snapshot.node_outputs[0].outputs[0] assert output.status == NodeOutputStatus.READY assert output.value_preview == "hello" -def test_snapshot_marks_type_check_failure(): +def test_snapshot_marks_type_check_failure(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING)], ) @@ -287,7 +340,7 @@ def test_snapshot_marks_type_check_failure(): } }, ) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) output = snapshot.node_outputs[0].outputs[0] assert output.status == NodeOutputStatus.TYPE_CHECK_FAILED @@ -296,7 +349,7 @@ def test_snapshot_marks_type_check_failure(): assert output.type_check.reason == "wrong shape" -def test_snapshot_marks_output_check_failure_when_type_check_passed(): +def test_snapshot_marks_output_check_failure_when_type_check_passed(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig( @@ -317,7 +370,7 @@ def test_snapshot_marks_output_check_failure_when_type_check_passed(): }, }, ) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with patch( "services.workflow.node_output_inspector_service.file_helpers.get_signed_file_url", return_value="https://signed.example/x", @@ -330,7 +383,7 @@ def test_snapshot_marks_output_check_failure_when_type_check_passed(): assert output.output_check.reason == "benchmark mismatch" -def test_snapshot_marks_not_produced_when_declared_output_missing_from_payload(): +def test_snapshot_marks_not_produced_when_declared_output_missing_from_payload(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING), @@ -339,7 +392,7 @@ def test_snapshot_marks_not_produced_when_declared_output_missing_from_payload() ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) ex = _execution(node_id="agent-1", outputs={"text": "hi"}) # optional_meta missing - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) statuses = {o.name: o.status for o in snapshot.node_outputs[0].outputs} assert statuses == {"text": NodeOutputStatus.READY, "optional_meta": NodeOutputStatus.NOT_PRODUCED} @@ -350,7 +403,7 @@ def test_snapshot_marks_not_produced_when_declared_output_missing_from_payload() # ────────────────────────────────────────────────────────────────────────────── -def test_non_agent_node_outputs_inferred_from_payload_keys(): +def test_non_agent_node_outputs_inferred_from_payload_keys(session_for: SessionFor) -> None: service = _make_service() run = _workflow_run(nodes=[_non_agent_node(node_id="tool-1", node_type="tool")]) ex = _execution( @@ -358,7 +411,7 @@ def test_non_agent_node_outputs_inferred_from_payload_keys(): node_type="tool", outputs={"message": "sent", "thread_ts": "1234"}, ) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) output_names = sorted(o.name for o in snapshot.node_outputs[0].outputs) assert output_names == ["message", "thread_ts"] @@ -372,7 +425,7 @@ def test_non_agent_node_outputs_inferred_from_payload_keys(): # ────────────────────────────────────────────────────────────────────────────── -def test_file_output_preview_includes_signed_url(): +def test_file_output_preview_includes_signed_url(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig(name="report", type=DeclaredOutputType.FILE), @@ -384,7 +437,7 @@ def test_file_output_preview_includes_signed_url(): "reference": build_file_reference(record_id="550e8400-e29b-41d4-a716-446655440000"), } ex = _execution(node_id="agent-1", outputs={"report": file_payload}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with patch( "services.workflow.node_output_inspector_service._resolve_preview_url", return_value="https://signed.example/x.pdf", @@ -396,7 +449,7 @@ def test_file_output_preview_includes_signed_url(): assert preview_value["reference"] == file_payload["reference"] -def test_file_output_preview_endpoint_returns_full_value_with_signed_url(): +def test_file_output_preview_endpoint_returns_full_value_with_signed_url(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig(name="report", type=DeclaredOutputType.FILE), @@ -408,7 +461,7 @@ def test_file_output_preview_endpoint_returns_full_value_with_signed_url(): "reference": build_file_reference(record_id="550e8400-e29b-41d4-a716-446655440000"), } ex = _execution(node_id="agent-1", outputs={"report": file_payload}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with patch( "services.workflow.node_output_inspector_service._resolve_preview_url", return_value="https://signed.example/x.pdf", @@ -450,7 +503,7 @@ def test_resolve_preview_url_uses_standard_file_factory(): resolve_file_url.assert_called_once_with(file) -def test_array_file_output_preview_includes_signed_urls_for_each_item(): +def test_array_file_output_preview_includes_signed_urls_for_each_item(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig( @@ -472,7 +525,7 @@ def test_array_file_output_preview_includes_signed_urls_for_each_item(): }, ] ex = _execution(node_id="agent-1", outputs={"files": file_payloads}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with patch( "services.workflow.node_output_inspector_service._resolve_preview_url", side_effect=[ @@ -506,7 +559,7 @@ def test_array_file_output_preview_includes_signed_urls_for_each_item(): ] -def test_file_output_preview_uses_none_when_signed_url_resolution_fails(): +def test_file_output_preview_uses_none_when_signed_url_resolution_fails(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig(name="report", type=DeclaredOutputType.FILE), @@ -518,7 +571,7 @@ def test_file_output_preview_uses_none_when_signed_url_resolution_fails(): "reference": build_file_reference(record_id="550e8400-e29b-41d4-a716-446655440000"), } ex = _execution(node_id="agent-1", outputs={"report": file_payload}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with patch( "services.workflow.node_output_inspector_service._resolve_preview_url", side_effect=RuntimeError("boom"), @@ -530,7 +583,7 @@ def test_file_output_preview_uses_none_when_signed_url_resolution_fails(): assert preview_value["preview_url"] is None -def test_object_output_preview_does_not_augment_canonical_file_mapping_shape(): +def test_object_output_preview_does_not_augment_canonical_file_mapping_shape(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig(name="meta", type=DeclaredOutputType.OBJECT), @@ -542,7 +595,7 @@ def test_object_output_preview_does_not_augment_canonical_file_mapping_shape(): "reference": build_file_reference(record_id="550e8400-e29b-41d4-a716-446655440000"), } ex = _execution(node_id="agent-1", outputs={"meta": raw_value}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) with patch( "services.workflow.node_output_inspector_service._resolve_preview_url", return_value="https://signed.example/x.pdf", @@ -565,7 +618,7 @@ def test_object_output_preview_does_not_augment_canonical_file_mapping_shape(): # ────────────────────────────────────────────────────────────────────────────── -def test_retried_count_pulled_from_attempt_metadata(): +def test_retried_count_pulled_from_attempt_metadata(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING)], ) @@ -575,7 +628,7 @@ def test_retried_count_pulled_from_attempt_metadata(): outputs={"text": "ok"}, execution_metadata={"attempt": 2}, ) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert snapshot.node_outputs[0].outputs[0].retried == 2 @@ -585,7 +638,7 @@ def test_retried_count_pulled_from_attempt_metadata(): # ────────────────────────────────────────────────────────────────────────────── -def test_keeps_latest_execution_per_node_by_index(): +def test_keeps_latest_execution_per_node_by_index(session_for: SessionFor) -> None: """When a node has multiple executions (retries / iterations) keep the canonical one — the row with the highest ``index``.""" service = _make_service( @@ -594,7 +647,9 @@ def test_keeps_latest_execution_per_node_by_index(): run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) older = _execution(node_id="agent-1", outputs={"text": "old"}, index=1) newer = _execution(node_id="agent-1", outputs={"text": "new"}, index=5) - session = _mock_session(workflow_run=run, executions=[older, newer]) + other_tenant = _execution(node_id="agent-1", outputs={"text": "other tenant"}, index=99, tenant_id="tenant-2") + other_run = _execution(node_id="agent-1", outputs={"text": "other run"}, index=100, workflow_run_id="run-2") + session = session_for(workflow_run=run, executions=[older, newer, other_tenant, other_run]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert snapshot.node_outputs[0].outputs[0].value_preview == "new" @@ -604,7 +659,7 @@ def test_keeps_latest_execution_per_node_by_index(): # ────────────────────────────────────────────────────────────────────────────── -def test_array_typed_output_with_array_item_renders_correctly(): +def test_array_typed_output_with_array_item_renders_correctly(session_for: SessionFor) -> None: service = _make_service( declared_outputs=[ DeclaredOutputConfig( @@ -616,7 +671,7 @@ def test_array_typed_output_with_array_item_renders_correctly(): ) run = _workflow_run(nodes=[_agent_v2_node(node_id="agent-1")]) ex = _execution(node_id="agent-1", outputs={"files": []}) - session = _mock_session(workflow_run=run, executions=[ex]) + session = session_for(workflow_run=run, executions=[ex]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) output = snapshot.node_outputs[0].outputs[0] assert output.type == DeclaredOutputType.ARRAY @@ -627,17 +682,9 @@ def test_array_typed_output_with_array_item_renders_correctly(): # ────────────────────────────────────────────────────────────────────────────── -def test_unparseable_graph_blob_yields_empty_snapshot_not_500(): +def test_unparseable_graph_blob_yields_empty_snapshot_not_500(session_for: SessionFor) -> None: service = _make_service() - run = SimpleNamespace( - id="run-1", - workflow_id="workflow-1", - tenant_id="tenant-1", - app_id="app-1", - triggered_from=WorkflowRunTriggeredFrom.DEBUGGING, - status=WorkflowExecutionStatus.RUNNING, - graph="{not valid json", - ) - session = _mock_session(workflow_run=run, executions=[]) + run = _workflow_run(graph="{not valid json") + session = session_for(workflow_run=run, executions=[]) snapshot = service.snapshot_workflow_run(app_model=_app_model(), workflow_run_id="run-1", session=session) assert snapshot.node_outputs == []