262 lines
10 KiB
C#
262 lines
10 KiB
C#
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using w4c_workflows.Data;
|
|
using w4c_workflows.Models;
|
|
using w4c_workflows.Models.Nodes;
|
|
using w4c_workflows.Services.Credentials;
|
|
using w4c_workflows.Services.Nodes;
|
|
using w4c_workflows.Services.Nodes.Executors;
|
|
using w4c_workflows.Services.Nodes.Interpolation;
|
|
using Xunit;
|
|
|
|
namespace w4c_workflows.Tests;
|
|
|
|
/// <summary>
|
|
/// End-to-end coverage for in-process sub-workflows (<c>core.executeWorkflow</c>):
|
|
/// the child run is linked to its parent, output items are mapped back, and the
|
|
/// depth/recursion guards stop runaway nesting. Uses the shared Postgres fixture
|
|
/// because child-run history is persisted.
|
|
/// </summary>
|
|
[Collection("WorkflowsPostgres")]
|
|
public class SubWorkflowInvokerTests
|
|
{
|
|
private readonly WorkflowsPostgresFixture _fixture;
|
|
|
|
public SubWorkflowInvokerTests(WorkflowsPostgresFixture fixture)
|
|
{
|
|
_fixture = fixture;
|
|
}
|
|
|
|
private static string Tenant() => "t" + Guid.NewGuid().ToString("N")[..12];
|
|
|
|
private static NodeWorkflowRunner NewRunner(int maxDepth = SubWorkflowInvoker.DefaultMaxDepth)
|
|
{
|
|
var catalog = NodeTestData.CoreCatalog();
|
|
var executors = new List<INodeExecutor>
|
|
{
|
|
new NoOpNodeExecutor(),
|
|
new SetNodeExecutor(),
|
|
new ExecuteWorkflowNodeExecutor(),
|
|
};
|
|
var graphRunner = new NodeGraphRunner(new NodeExecutorRegistry(executors), new NodeParameterInterpolator());
|
|
var vault = new CredentialVault(new ReversibleTestCipher(), new CredentialTypeCatalog());
|
|
return new NodeWorkflowRunner(catalog, graphRunner, vault, NullLogger<NodeWorkflowRunner>.Instance, maxDepth);
|
|
}
|
|
|
|
/// <summary>Child workflow: adds a <c>child</c> marker to each input item.</summary>
|
|
private const string ChildYaml = """
|
|
name: child-transform
|
|
tasks:
|
|
- id: mark
|
|
node: { type: core.set }
|
|
parameters:
|
|
mode: manual
|
|
fields:
|
|
- name: child
|
|
value: true
|
|
""";
|
|
|
|
private static string ParentYaml(string workflow, string extra = "")
|
|
=> $$"""
|
|
name: parent-call
|
|
tasks:
|
|
- id: call
|
|
node: { type: core.executeWorkflow }
|
|
parameters:
|
|
workflow: {{workflow}}
|
|
{{extra}}
|
|
""";
|
|
|
|
[Fact]
|
|
public async Task Parent_run_invokes_child_and_maps_its_output()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
var child = WorkflowDataHelpers.CompileAndSave(db, tenant, ChildYaml, "workflows/child.yaml");
|
|
var parent = WorkflowDataHelpers.CompileAndSave(db, tenant, ParentYaml("child-transform"), "workflows/parent.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1},{"n":2}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.True(outcome.Succeeded, outcome.Error);
|
|
var output = FlowItemJson.Parse(run.OutputJson);
|
|
Assert.Equal(2, output.Count);
|
|
Assert.All(output, item => Assert.True(item.Json["child"]!.GetValue<bool>()));
|
|
|
|
var childRun = await db.WorkflowRuns.SingleAsync(r => r.ParentRunId == run.Id);
|
|
Assert.Equal(child.Id, childRun.WorkflowId);
|
|
Assert.Equal(RunStatus.Succeeded, childRun.Status);
|
|
Assert.Equal(1, childRun.Depth);
|
|
Assert.Equal(WorkflowDataHelpers.TaskByKey(parent, "call").Id, childRun.ParentTaskId);
|
|
|
|
var childTaskRuns = await db.TaskRuns.Where(t => t.RunId == childRun.Id).ToListAsync();
|
|
Assert.Single(childTaskRuns);
|
|
Assert.Equal(TaskRunStatus.Succeeded, childTaskRuns[0].Status);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Child_can_be_referenced_by_id()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
var child = WorkflowDataHelpers.CompileAndSave(db, tenant, ChildYaml, "workflows/child.yaml");
|
|
var parent = WorkflowDataHelpers.CompileAndSave(
|
|
db, tenant, ParentYaml(child.Id.ToString()), "workflows/parent-id.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.True(outcome.Succeeded, outcome.Error);
|
|
Assert.True(await db.WorkflowRuns.AnyAsync(r => r.ParentRunId == run.Id));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Each_item_mode_starts_one_child_run_per_item()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
WorkflowDataHelpers.CompileAndSave(db, tenant, ChildYaml, "workflows/child.yaml");
|
|
var parent = WorkflowDataHelpers.CompileAndSave(
|
|
db, tenant,
|
|
ParentYaml("child-transform", " mode: eachItem"),
|
|
"workflows/parent-each.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1},{"n":2},{"n":3}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.True(outcome.Succeeded, outcome.Error);
|
|
var childRuns = await db.WorkflowRuns.Where(r => r.ParentRunId == run.Id).ToListAsync();
|
|
Assert.Equal(3, childRuns.Count);
|
|
Assert.All(childRuns, r => Assert.Equal(1, r.Depth));
|
|
Assert.Equal(3, FlowItemJson.Parse(run.OutputJson).Count);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Dropping_history_skips_the_child_task_rows()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
WorkflowDataHelpers.CompileAndSave(db, tenant, ChildYaml, "workflows/child.yaml");
|
|
var parent = WorkflowDataHelpers.CompileAndSave(
|
|
db, tenant,
|
|
ParentYaml("child-transform", " recordHistory: false"),
|
|
"workflows/parent-nohistory.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.True(outcome.Succeeded, outcome.Error);
|
|
var childRun = await db.WorkflowRuns.SingleAsync(r => r.ParentRunId == run.Id);
|
|
Assert.Empty(await db.TaskRuns.Where(t => t.RunId == childRun.Id).ToListAsync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Not_waiting_runs_the_child_but_emits_no_items()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
WorkflowDataHelpers.CompileAndSave(db, tenant, ChildYaml, "workflows/child.yaml");
|
|
var parent = WorkflowDataHelpers.CompileAndSave(
|
|
db, tenant,
|
|
ParentYaml("child-transform", " waitForCompletion: false"),
|
|
"workflows/parent-nowait.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.True(outcome.Succeeded, outcome.Error);
|
|
Assert.Null(run.OutputJson);
|
|
Assert.True(await db.WorkflowRuns.AnyAsync(r => r.ParentRunId == run.Id));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Recursive_self_call_is_rejected()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
const string selfYaml = """
|
|
name: self-call
|
|
tasks:
|
|
- id: call
|
|
node: { type: core.executeWorkflow }
|
|
parameters:
|
|
workflow: self-call
|
|
""";
|
|
var self = WorkflowDataHelpers.CompileAndSave(db, tenant, selfYaml, "workflows/self.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, self, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, self, null, default);
|
|
|
|
Assert.False(outcome.Succeeded);
|
|
Assert.Contains("recursive sub-workflow call", run.Error);
|
|
// The cycle was caught before any child run was created.
|
|
Assert.Empty(await db.WorkflowRuns.Where(r => r.ParentRunId == run.Id).ToListAsync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Nesting_beyond_the_depth_limit_fails_cleanly()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
WorkflowDataHelpers.CompileAndSave(db, tenant,
|
|
"name: depth-a\ntasks:\n - id: call\n node: { type: core.executeWorkflow }\n parameters:\n workflow: depth-b\n",
|
|
"workflows/depth-a.yaml");
|
|
WorkflowDataHelpers.CompileAndSave(db, tenant,
|
|
"name: depth-b\ntasks:\n - id: call\n node: { type: core.executeWorkflow }\n parameters:\n workflow: depth-c\n",
|
|
"workflows/depth-b.yaml");
|
|
WorkflowDataHelpers.CompileAndSave(db, tenant,
|
|
"name: depth-c\ntasks:\n - id: noop\n node: { type: core.noop }\n",
|
|
"workflows/depth-c.yaml");
|
|
|
|
var a = await db.Workflows.SingleAsync(w => w.Name == "depth-a" && w.TenantId == tenant);
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, a, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner(maxDepth: 1).RunAsync(db, run, a, null, default);
|
|
|
|
Assert.False(outcome.Succeeded);
|
|
Assert.Contains("maximum depth", run.Error);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Unknown_child_workflow_fails_the_run()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
var parent = WorkflowDataHelpers.CompileAndSave(
|
|
db, tenant, ParentYaml("does-not-exist"), "workflows/parent-missing.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.False(outcome.Succeeded);
|
|
Assert.Contains("was not found", run.Error);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Uncompiled_child_workflow_fails_the_run()
|
|
{
|
|
var tenant = Tenant();
|
|
await using var db = _fixture.CreateContext();
|
|
db.Workflows.Add(new Workflow
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
TenantId = tenant,
|
|
Name = "broken-child",
|
|
Path = "workflows/broken.yaml",
|
|
Status = WorkflowStatus.Invalid,
|
|
Mode = WorkflowMode.Function,
|
|
Target = "local",
|
|
});
|
|
await db.SaveChangesAsync();
|
|
|
|
var parent = WorkflowDataHelpers.CompileAndSave(
|
|
db, tenant, ParentYaml("broken-child"), "workflows/parent-broken.yaml");
|
|
var run = WorkflowDataHelpers.AddPendingRun(db, parent, """[{"n":1}]""");
|
|
|
|
var outcome = await NewRunner().RunAsync(db, run, parent, null, default);
|
|
|
|
Assert.False(outcome.Succeeded);
|
|
Assert.Contains("is not compiled", run.Error);
|
|
}
|
|
}
|