123 lines
3.2 KiB
Python
123 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from wf_core import (
|
|
END,
|
|
Edge,
|
|
InterruptNode,
|
|
NodeDef,
|
|
NodeUse,
|
|
RunStatus,
|
|
RuntimeContext,
|
|
SchemaRef,
|
|
StateSchema,
|
|
Workflow,
|
|
execute_workflow_async,
|
|
execute_workflow_result_async,
|
|
resume_workflow_async,
|
|
resume_workflow_result_async,
|
|
)
|
|
|
|
|
|
async def explode(_payload: dict[str, Any], _context: RuntimeContext) -> dict[str, Any]:
|
|
raise ValueError("boom")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_result_api_returns_failed_state_without_changing_strict_execute() -> (
|
|
None
|
|
):
|
|
workflow = _failing_workflow()
|
|
|
|
failed = await execute_workflow_result_async(workflow, {}, {"explode": explode})
|
|
|
|
assert failed.status is RunStatus.FAILED
|
|
assert failed.error == "boom"
|
|
|
|
with pytest.raises(ValueError, match="boom"):
|
|
await execute_workflow_async(workflow, {}, {"explode": explode})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_result_api_returns_failed_state_without_changing_strict_resume() -> (
|
|
None
|
|
):
|
|
workflow = _interrupt_then_fail_workflow()
|
|
interrupted = await execute_workflow_async(workflow, {}, {"explode": explode})
|
|
|
|
failed = await resume_workflow_result_async(
|
|
workflow,
|
|
interrupted,
|
|
{"explode": explode},
|
|
resume_payload={},
|
|
)
|
|
|
|
assert failed.status is RunStatus.FAILED
|
|
assert failed.error == "boom"
|
|
|
|
interrupted = await execute_workflow_async(workflow, {}, {"explode": explode})
|
|
with pytest.raises(ValueError, match="boom"):
|
|
await resume_workflow_async(
|
|
workflow,
|
|
interrupted,
|
|
{"explode": explode},
|
|
resume_payload={},
|
|
)
|
|
|
|
|
|
def _failing_workflow() -> Workflow:
|
|
return Workflow(
|
|
name="failing",
|
|
input_schema=_schema(),
|
|
state_schema=StateSchema.from_field_map({}),
|
|
output_schema=_schema(),
|
|
outcomes=["ok"],
|
|
start="explode",
|
|
node_defs=[
|
|
NodeDef(
|
|
name="explode",
|
|
input_schema=_schema(),
|
|
output_schema=_schema(),
|
|
outcomes=["ok"],
|
|
)
|
|
],
|
|
nodes=[NodeUse(id="explode", type="node", node="explode")],
|
|
edges=[Edge.model_validate({"from": "explode", "outcome": "ok", "to": END})],
|
|
)
|
|
|
|
|
|
def _interrupt_then_fail_workflow() -> Workflow:
|
|
return Workflow(
|
|
name="interrupt_then_fail",
|
|
input_schema=_schema(),
|
|
state_schema=StateSchema.from_field_map({}),
|
|
output_schema=_schema(),
|
|
outcomes=["ok"],
|
|
start="ask",
|
|
node_defs=[
|
|
NodeDef(
|
|
name="explode",
|
|
input_schema=_schema(),
|
|
output_schema=_schema(),
|
|
outcomes=["ok"],
|
|
)
|
|
],
|
|
nodes=[
|
|
InterruptNode(id="ask", type="interrupt", kind="approval"),
|
|
NodeUse(id="explode", type="node", node="explode"),
|
|
],
|
|
edges=[
|
|
Edge.model_validate(
|
|
{"from": "ask", "outcome": "submitted", "to": "explode"}
|
|
),
|
|
Edge.model_validate({"from": "explode", "outcome": "ok", "to": END}),
|
|
],
|
|
)
|
|
|
|
|
|
def _schema() -> SchemaRef:
|
|
return SchemaRef(type="object", properties={})
|