988 lines
32 KiB
Python
988 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import (
|
|
CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
|
|
PrincipalRef,
|
|
)
|
|
from govoplan_core.core.automation import (
|
|
ActionDefinition,
|
|
ActionExecutionResult,
|
|
ActionPreview,
|
|
AutomationPrincipalResolution,
|
|
EffectDefinition,
|
|
EffectPreview,
|
|
ObservedEffect,
|
|
)
|
|
from govoplan_core.core.dataflows import (
|
|
CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
|
|
DataflowRunDescriptor,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_workflow_engine.backend.db.models import (
|
|
WorkflowDefinition,
|
|
WorkflowDefinitionRevision,
|
|
WorkflowInstance,
|
|
WorkflowInstanceEvent,
|
|
WorkflowInstanceStep,
|
|
)
|
|
from govoplan_workflow_engine.backend.instance_service import (
|
|
SqlWorkflowRuntimeWorker,
|
|
cancel_instance,
|
|
instance_response,
|
|
reconcile_instance,
|
|
resolve_step,
|
|
start_instance,
|
|
)
|
|
from govoplan_workflow_engine.backend.schemas import (
|
|
BpmnRevisionInput,
|
|
WorkflowDefinitionCreateRequest,
|
|
WorkflowEdge,
|
|
WorkflowGraph,
|
|
WorkflowInstanceStartRequest,
|
|
WorkflowNode,
|
|
WorkflowStepActionRequest,
|
|
)
|
|
from govoplan_workflow_engine.backend.bpmn_adapters import NATIVE_LINEAR_ADAPTER_ID
|
|
from govoplan_workflow_engine.backend.service import (
|
|
WorkflowConflictError,
|
|
activate_definition,
|
|
create_definition,
|
|
)
|
|
try:
|
|
from test_bpmn import NATIVE_BPMN
|
|
except ModuleNotFoundError as exc:
|
|
if exc.name != "test_bpmn":
|
|
raise
|
|
from tests.test_bpmn import NATIVE_BPMN
|
|
|
|
|
|
def principal() -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(
|
|
{
|
|
"workflow:definition:read",
|
|
"workflow:instance:read",
|
|
"workflow:instance:start",
|
|
"workflow:instance:transition",
|
|
"dataflow:pipeline:run",
|
|
}
|
|
),
|
|
),
|
|
account=object(),
|
|
user=object(),
|
|
)
|
|
|
|
|
|
def runtime_graph() -> WorkflowGraph:
|
|
return WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="start",
|
|
type="workflow.start.manual",
|
|
label="Start",
|
|
config={"input_schema_ref": ""},
|
|
),
|
|
WorkflowNode(
|
|
id="flow",
|
|
type="workflow.dataflow",
|
|
label="Prepare evidence",
|
|
config={
|
|
"pipeline_ref": "pipeline:pipeline-1",
|
|
"revision": 3,
|
|
"environment": "development",
|
|
"row_limit": 250,
|
|
"publication_target_ref": "",
|
|
"warning_policy": "review",
|
|
"input_mapping": {},
|
|
"view_surface_ids": [
|
|
"dataflow.module",
|
|
"dataflow.route.pipelines",
|
|
],
|
|
},
|
|
),
|
|
WorkflowNode(
|
|
id="complete",
|
|
type="workflow.end.completed",
|
|
label="Complete",
|
|
config={"output_mapping": {}},
|
|
),
|
|
WorkflowNode(
|
|
id="cancelled",
|
|
type="workflow.end.cancelled",
|
|
label="Rejected",
|
|
config={"reason": "Rejected during review"},
|
|
),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(
|
|
id="start-flow",
|
|
source="start",
|
|
target="flow",
|
|
),
|
|
WorkflowEdge(
|
|
id="flow-complete",
|
|
source="flow",
|
|
source_port="success",
|
|
target="complete",
|
|
),
|
|
WorkflowEdge(
|
|
id="flow-warning",
|
|
source="flow",
|
|
source_port="warning",
|
|
target="complete",
|
|
),
|
|
WorkflowEdge(
|
|
id="flow-review",
|
|
source="flow",
|
|
source_port="review_required",
|
|
target="complete",
|
|
),
|
|
WorkflowEdge(
|
|
id="flow-failure",
|
|
source="flow",
|
|
source_port="failure",
|
|
target="cancelled",
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
def action_graph() -> WorkflowGraph:
|
|
return WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="start",
|
|
type="workflow.start.manual",
|
|
config={"input_schema_ref": ""},
|
|
),
|
|
WorkflowNode(
|
|
id="action",
|
|
type="workflow.capability",
|
|
config={
|
|
"capability": "test.actions",
|
|
"operation": "test.case.record",
|
|
"input_mapping": {"case_id": "$input.case_id"},
|
|
"idempotency_key": "$input.case_id",
|
|
"failure_policy": "manual",
|
|
},
|
|
),
|
|
WorkflowNode(
|
|
id="complete",
|
|
type="workflow.end.completed",
|
|
config={"output_mapping": {}},
|
|
),
|
|
WorkflowNode(
|
|
id="failed",
|
|
type="workflow.end.cancelled",
|
|
config={"reason": "Action rejected"},
|
|
),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(
|
|
id="start-action",
|
|
source="start",
|
|
target="action",
|
|
),
|
|
WorkflowEdge(
|
|
id="action-complete",
|
|
source="action",
|
|
source_port="success",
|
|
target="complete",
|
|
),
|
|
WorkflowEdge(
|
|
id="action-failed",
|
|
source="action",
|
|
source_port="failure",
|
|
target="failed",
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
class FakeDataflowLifecycle:
|
|
def __init__(self) -> None:
|
|
self.runs: dict[str, DataflowRunDescriptor] = {}
|
|
self.requests = []
|
|
self.cancelled: list[str] = []
|
|
|
|
def start_run(self, _session, _principal, *, request):
|
|
self.requests.append(request)
|
|
run_ref = f"run:{len(self.requests)}"
|
|
descriptor = DataflowRunDescriptor(
|
|
ref=run_ref,
|
|
pipeline_ref=request.pipeline_ref,
|
|
revision=request.revision,
|
|
status="queued",
|
|
definition_hash="definition-hash",
|
|
executor_version="test",
|
|
metadata={"progress_percent": 0, "progress_phase": "queued"},
|
|
)
|
|
self.runs[run_ref] = descriptor
|
|
return descriptor
|
|
|
|
def get_run(self, _session, _principal, *, run_ref):
|
|
return self.runs.get(run_ref)
|
|
|
|
def cancel_run(self, _session, _principal, *, run_ref):
|
|
descriptor = self.runs[run_ref]
|
|
descriptor = replace(descriptor, status="cancelled")
|
|
self.runs[run_ref] = descriptor
|
|
self.cancelled.append(run_ref)
|
|
return descriptor
|
|
|
|
def finish(
|
|
self,
|
|
run_ref: str,
|
|
*,
|
|
diagnostics: list[dict[str, object]] | None = None,
|
|
) -> None:
|
|
self.runs[run_ref] = replace(
|
|
self.runs[run_ref],
|
|
status="succeeded",
|
|
output_publication_ref="publication:1",
|
|
output_datasource_ref="datasource:1",
|
|
output_materialization_ref="materialization:1",
|
|
input_row_count=12,
|
|
output_row_count=10,
|
|
metadata={"diagnostics": diagnostics or []},
|
|
)
|
|
|
|
def fail(self, run_ref: str) -> None:
|
|
self.runs[run_ref] = replace(
|
|
self.runs[run_ref],
|
|
status="failed",
|
|
error="Data quality gate failed.",
|
|
)
|
|
|
|
|
|
class FakeAutomationProvider:
|
|
def __init__(self) -> None:
|
|
self.requests = []
|
|
|
|
def resolve_automation_principal(self, _session, *, request):
|
|
self.requests.append(request)
|
|
return AutomationPrincipalResolution(
|
|
allowed=True,
|
|
principal=principal(),
|
|
granted_scopes=request.grant_scopes,
|
|
provenance={"status": "rechecked"},
|
|
)
|
|
|
|
|
|
class FakeActionProvider:
|
|
action = ActionDefinition(
|
|
action_key="test.case.record",
|
|
owner_module="test",
|
|
description="Record a test case.",
|
|
input_schema_ref="schema:test.case.record@1",
|
|
expected_effect_keys=("test.case.recorded",),
|
|
)
|
|
effect = EffectDefinition(
|
|
effect_key="test.case.recorded",
|
|
owner_module="test",
|
|
operation="created",
|
|
description="A test case was recorded.",
|
|
)
|
|
|
|
def __init__(self, *states: str) -> None:
|
|
self.states = list(states or ("completed",))
|
|
self.requests = []
|
|
|
|
def action_definitions(self):
|
|
return (self.action,)
|
|
|
|
def effect_definitions(self):
|
|
return (self.effect,)
|
|
|
|
def preview_action(self, _session, _principal, *, request):
|
|
return ActionPreview(
|
|
action_key=request.action_key,
|
|
allowed=True,
|
|
summary="Record one case.",
|
|
risk_level=self.action.risk_level,
|
|
reversibility=self.action.reversibility,
|
|
effects=(
|
|
EffectPreview(
|
|
effect_key=self.effect.effect_key,
|
|
summary="Record case.",
|
|
),
|
|
),
|
|
preview_ref="preview:test",
|
|
)
|
|
|
|
def execute_action(self, _session, _principal, *, request):
|
|
self.requests.append(request)
|
|
state = self.states.pop(0) if self.states else "completed"
|
|
if state != "completed":
|
|
return ActionExecutionResult(
|
|
state=state,
|
|
error="Temporary action failure.",
|
|
)
|
|
return ActionExecutionResult(
|
|
state="completed",
|
|
output={"case_ref": "case:1"},
|
|
observed_effects=(
|
|
ObservedEffect(
|
|
effect_key=self.effect.effect_key,
|
|
operation="created",
|
|
resource_ref="case:1",
|
|
),
|
|
),
|
|
audit_event_refs=("audit:1",),
|
|
)
|
|
|
|
|
|
class Registry:
|
|
def __init__(
|
|
self,
|
|
dataflow: FakeDataflowLifecycle,
|
|
automation: FakeAutomationProvider | None = None,
|
|
action: FakeActionProvider | None = None,
|
|
) -> None:
|
|
self.dataflow = dataflow
|
|
self.automation = automation
|
|
self.action = action
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name == CAPABILITY_DATAFLOW_RUN_LIFECYCLE or (
|
|
name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
|
|
and self.automation is not None
|
|
) or (
|
|
name == "test.actions"
|
|
and self.action is not None
|
|
)
|
|
|
|
def capability(self, name: str):
|
|
if name == CAPABILITY_DATAFLOW_RUN_LIFECYCLE:
|
|
return self.dataflow
|
|
if (
|
|
name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
|
|
and self.automation is not None
|
|
):
|
|
return self.automation
|
|
if name == "test.actions" and self.action is not None:
|
|
return self.action
|
|
raise KeyError(name)
|
|
|
|
|
|
class WorkflowInstanceServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[
|
|
WorkflowDefinition.__table__,
|
|
WorkflowDefinitionRevision.__table__,
|
|
WorkflowInstance.__table__,
|
|
WorkflowInstanceStep.__table__,
|
|
WorkflowInstanceEvent.__table__,
|
|
],
|
|
)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.session: Session = self.Session()
|
|
self.dataflow = FakeDataflowLifecycle()
|
|
self.registry = Registry(self.dataflow)
|
|
self.definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Monthly governed processing",
|
|
graph=runtime_graph(),
|
|
execution_mode="hybrid",
|
|
view_id="view-1",
|
|
view_revision_id="view-revision-1",
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=self.definition.id,
|
|
actor_id="account-1",
|
|
revision=1,
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
Base.metadata.drop_all(
|
|
self.engine,
|
|
tables=[
|
|
WorkflowInstanceEvent.__table__,
|
|
WorkflowInstanceStep.__table__,
|
|
WorkflowInstance.__table__,
|
|
WorkflowDefinitionRevision.__table__,
|
|
WorkflowDefinition.__table__,
|
|
],
|
|
)
|
|
self.engine.dispose()
|
|
|
|
def _start(self, key: str = "request-1") -> WorkflowInstance:
|
|
instance, replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=self.definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key=key,
|
|
input={"case_id": "case-1"},
|
|
correlation_id="correlation-1",
|
|
),
|
|
)
|
|
self.assertFalse(replayed)
|
|
return instance
|
|
|
|
def test_start_pins_revision_and_replays_idempotently(self) -> None:
|
|
instance = self._start()
|
|
replayed, was_replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=self.definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="request-1",
|
|
input={"case_id": "case-1"},
|
|
correlation_id="correlation-1",
|
|
),
|
|
)
|
|
|
|
self.assertTrue(was_replayed)
|
|
self.assertEqual(instance.id, replayed.id)
|
|
self.assertEqual("waiting", instance.status)
|
|
self.assertEqual("user", instance.start_origin)
|
|
self.assertEqual(1, len(self.dataflow.requests))
|
|
response = instance_response(self.session, instance)
|
|
self.assertEqual("hybrid", response.execution_mode)
|
|
self.assertEqual("user", response.start_origin)
|
|
self.assertIsNotNone(response.view_context)
|
|
self.assertEqual(
|
|
[
|
|
"dataflow.module",
|
|
"dataflow.route.pipelines",
|
|
],
|
|
response.view_context.visible_surface_ids,
|
|
)
|
|
self.assertEqual([1, 2], [step.sequence for step in response.steps])
|
|
self.assertEqual("run:1", response.steps[-1].external_ref)
|
|
self.assertGreaterEqual(len(response.events), 4)
|
|
|
|
with self.assertRaises(WorkflowConflictError):
|
|
start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=self.definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="request-1",
|
|
input={"case_id": "another-case"},
|
|
correlation_id="correlation-1",
|
|
),
|
|
)
|
|
|
|
def test_guided_workflow_rejects_automated_start_origin(self) -> None:
|
|
definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Guided review",
|
|
graph=WorkflowGraph(
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="start",
|
|
type="workflow.start.api",
|
|
config={
|
|
"input_schema_ref": "schema:input",
|
|
"authorization_policy_ref": "policy:start",
|
|
},
|
|
),
|
|
WorkflowNode(
|
|
id="activity",
|
|
type="workflow.activity",
|
|
config={"title": "Review"},
|
|
),
|
|
WorkflowNode(
|
|
id="done",
|
|
type="workflow.end.completed",
|
|
),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(
|
|
id="start-activity",
|
|
source="start",
|
|
target="activity",
|
|
),
|
|
WorkflowEdge(
|
|
id="activity-done",
|
|
source="activity",
|
|
target="done",
|
|
),
|
|
],
|
|
),
|
|
execution_mode="guided",
|
|
allow_automation=True,
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
WorkflowConflictError,
|
|
"Guided workflows must be started by a user",
|
|
):
|
|
start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="automated-guided",
|
|
),
|
|
start_origin="api",
|
|
)
|
|
|
|
def test_module_action_records_effects_and_completes_idempotently(
|
|
self,
|
|
) -> None:
|
|
action = FakeActionProvider()
|
|
registry = Registry(self.dataflow, action=action)
|
|
definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Action workflow",
|
|
graph=action_graph(),
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
)
|
|
|
|
instance, replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="action-instance",
|
|
input={"case_id": "case-1"},
|
|
),
|
|
)
|
|
response = instance_response(self.session, instance)
|
|
|
|
self.assertFalse(replayed)
|
|
self.assertEqual("completed", response.status)
|
|
self.assertEqual(1, len(action.requests))
|
|
self.assertEqual(
|
|
"test.actions:test.case.record:case-1",
|
|
action.requests[0].idempotency_key,
|
|
)
|
|
self.assertEqual(
|
|
"case:1",
|
|
response.steps[1].output["execution"]["observed_effects"][0][
|
|
"resource_ref"
|
|
],
|
|
)
|
|
self.assertIn(
|
|
"workflow.action.completed",
|
|
{event.kind for event in response.events},
|
|
)
|
|
|
|
def test_retryable_module_action_reuses_the_idempotency_key(self) -> None:
|
|
action = FakeActionProvider("retryable", "completed")
|
|
registry = Registry(self.dataflow, action=action)
|
|
definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Retry action",
|
|
graph=action_graph(),
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
)
|
|
instance, _replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="retry-action-instance",
|
|
input={"case_id": "case-2"},
|
|
),
|
|
)
|
|
waiting = instance_response(self.session, instance)
|
|
current = waiting.steps[-1]
|
|
self.assertEqual("retryable", current.handoff["state"])
|
|
|
|
resolved = resolve_step(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=instance.id,
|
|
step_id=current.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=registry,
|
|
payload=WorkflowStepActionRequest(action="retry"),
|
|
)
|
|
|
|
self.assertEqual("completed", resolved.status)
|
|
self.assertEqual(
|
|
action.requests[0].idempotency_key,
|
|
action.requests[1].idempotency_key,
|
|
)
|
|
|
|
def test_automated_dataflow_failure_policy_fails_without_handoff(
|
|
self,
|
|
) -> None:
|
|
graph = runtime_graph()
|
|
flow = next(node for node in graph.nodes if node.id == "flow")
|
|
flow.config["warning_policy"] = "continue"
|
|
flow.config["failure_policy"] = "fail"
|
|
definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Automated Dataflow",
|
|
graph=graph,
|
|
execution_mode="automated",
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
)
|
|
instance, _replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="automated-dataflow",
|
|
),
|
|
)
|
|
self.dataflow.fail("run:1")
|
|
|
|
changed = reconcile_instance(
|
|
self.session,
|
|
instance=instance,
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
)
|
|
|
|
self.assertTrue(changed)
|
|
self.assertEqual("failed", instance.status)
|
|
self.assertIsNone(instance.current_step_id)
|
|
self.assertFalse(
|
|
any(
|
|
step.handoff.get("kind") == "dataflow_failure"
|
|
for step in instance.steps
|
|
)
|
|
)
|
|
|
|
def test_native_bpmn_profile_runs_through_canonical_instance_state(self) -> None:
|
|
definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="BPMN governed review",
|
|
graph=runtime_graph(),
|
|
bpmn=BpmnRevisionInput(
|
|
xml=NATIVE_BPMN,
|
|
adapter_id=NATIVE_LINEAR_ADAPTER_ID,
|
|
),
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
)
|
|
instance, replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="bpmn-request-1",
|
|
input={"case_id": "case-bpmn"},
|
|
),
|
|
)
|
|
|
|
self.assertFalse(replayed)
|
|
self.assertEqual("waiting", instance.status)
|
|
waiting = instance_response(self.session, instance).steps[-1]
|
|
self.assertEqual("workflow.activity", waiting.node_type)
|
|
self.assertEqual("Review request", waiting.handoff["title"])
|
|
|
|
replay, was_replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="bpmn-request-1",
|
|
input={"case_id": "case-bpmn"},
|
|
),
|
|
)
|
|
self.assertTrue(was_replayed)
|
|
self.assertEqual(instance.id, replay.id)
|
|
|
|
resolved = resolve_step(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=instance.id,
|
|
step_id=waiting.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowStepActionRequest(
|
|
action="complete",
|
|
comment="Reviewed.",
|
|
evidence=["case:case-bpmn"],
|
|
),
|
|
)
|
|
response = instance_response(self.session, resolved)
|
|
self.assertEqual("completed", response.status)
|
|
self.assertEqual(
|
|
"workflow.instance.completed",
|
|
response.events[-1].kind,
|
|
)
|
|
|
|
def test_reconcile_completes_with_stable_dataflow_output_refs(self) -> None:
|
|
instance = self._start()
|
|
self.dataflow.finish("run:1")
|
|
|
|
changed = reconcile_instance(
|
|
self.session,
|
|
instance=instance,
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
actor_id="account-1",
|
|
)
|
|
response = instance_response(self.session, instance)
|
|
|
|
self.assertTrue(changed)
|
|
self.assertEqual("completed", response.status)
|
|
flow_output = response.context["steps"]["flow"]
|
|
self.assertEqual("publication:1", flow_output["output_publication_ref"])
|
|
self.assertEqual("datasource:1", flow_output["output_datasource_ref"])
|
|
self.assertEqual(
|
|
"materialization:1",
|
|
flow_output["output_materialization_ref"],
|
|
)
|
|
self.assertEqual("workflow.instance.completed", response.events[-1].kind)
|
|
|
|
def test_warning_requires_review_and_approve_resumes(self) -> None:
|
|
instance = self._start()
|
|
self.dataflow.finish(
|
|
"run:1",
|
|
diagnostics=[
|
|
{
|
|
"severity": "warning",
|
|
"code": "review.required",
|
|
"message": "Verify unmatched records.",
|
|
}
|
|
],
|
|
)
|
|
reconcile_instance(
|
|
self.session,
|
|
instance=instance,
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
)
|
|
step_id = str(instance.current_step_id)
|
|
|
|
self.assertEqual(
|
|
"review_required",
|
|
instance_response(self.session, instance).steps[-1].handoff["state"],
|
|
)
|
|
resolved = resolve_step(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=instance.id,
|
|
step_id=step_id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowStepActionRequest(
|
|
action="approve",
|
|
comment="Evidence verified.",
|
|
evidence=["publication:1"],
|
|
),
|
|
)
|
|
|
|
self.assertEqual("completed", resolved.status)
|
|
|
|
def test_failure_can_retry_and_reject_invalid_actions(self) -> None:
|
|
instance = self._start()
|
|
self.dataflow.fail("run:1")
|
|
reconcile_instance(
|
|
self.session,
|
|
instance=instance,
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
)
|
|
step_id = str(instance.current_step_id)
|
|
|
|
with self.assertRaises(WorkflowConflictError):
|
|
resolve_step(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=instance.id,
|
|
step_id=step_id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowStepActionRequest(action="approve"),
|
|
)
|
|
retried = resolve_step(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=instance.id,
|
|
step_id=step_id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
payload=WorkflowStepActionRequest(action="retry"),
|
|
)
|
|
|
|
self.assertEqual("waiting", retried.status)
|
|
self.assertEqual(2, len(self.dataflow.requests))
|
|
self.assertEqual("run:2", retried.steps[-1].external_ref)
|
|
self.assertEqual("superseded", retried.steps[-2].status)
|
|
|
|
def test_cancel_propagates_to_linked_dataflow(self) -> None:
|
|
instance = self._start()
|
|
|
|
cancelled = cancel_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=instance.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=self.registry,
|
|
)
|
|
|
|
self.assertEqual("cancelled", cancelled.status)
|
|
self.assertEqual(["run:1"], self.dataflow.cancelled)
|
|
|
|
def test_worker_rechecks_authorization_before_reconciling(self) -> None:
|
|
instance = self._start()
|
|
self.session.commit()
|
|
self.dataflow.finish("run:1")
|
|
automation = FakeAutomationProvider()
|
|
worker = SqlWorkflowRuntimeWorker(
|
|
registry=Registry(self.dataflow, automation),
|
|
)
|
|
|
|
summary = worker.reconcile_pending(self.session)
|
|
|
|
self.assertEqual(1, summary["advanced"])
|
|
self.assertEqual("completed", instance.status)
|
|
self.assertEqual(1, len(automation.requests))
|
|
self.assertEqual(
|
|
"rechecked",
|
|
instance.authorization_["last_resolution"]["status"],
|
|
)
|
|
|
|
def test_worker_reconciles_pending_module_action(self) -> None:
|
|
action = FakeActionProvider("pending", "completed")
|
|
automation = FakeAutomationProvider()
|
|
registry = Registry(
|
|
self.dataflow,
|
|
automation=automation,
|
|
action=action,
|
|
)
|
|
definition = create_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
actor_id="account-1",
|
|
payload=WorkflowDefinitionCreateRequest(
|
|
name="Asynchronous action",
|
|
graph=action_graph(),
|
|
),
|
|
)
|
|
activate_definition(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
)
|
|
instance, _replayed = start_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
definition_id=definition.id,
|
|
actor_id="account-1",
|
|
principal=principal(),
|
|
registry=registry,
|
|
payload=WorkflowInstanceStartRequest(
|
|
idempotency_key="pending-action",
|
|
input={"case_id": "case-3"},
|
|
),
|
|
)
|
|
self.session.commit()
|
|
worker = SqlWorkflowRuntimeWorker(registry=registry)
|
|
|
|
summary = worker.reconcile_pending(self.session)
|
|
|
|
self.assertEqual(1, summary["advanced"])
|
|
self.assertEqual("completed", instance.status)
|
|
self.assertEqual(2, len(action.requests))
|
|
self.assertEqual(
|
|
action.requests[0].idempotency_key,
|
|
action.requests[1].idempotency_key,
|
|
)
|
|
self.assertEqual(1, len(automation.requests))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|