Joseph Doherty
626 lines
34 KiB
C#
626 lines
34 KiB
C#
using System.Text.Json;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using NSubstitute;
|
|
using ZB.MOM.WW.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
|
using ZB.MOM.WW.ScadaBridge.DeploymentManager;
|
|
using ZB.MOM.WW.ScadaBridge.InboundAPI;
|
|
using ZB.MOM.WW.ScadaBridge.ManagementService.FlowDebug;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;
|
|
|
|
/// <summary>
|
|
/// Flow Debugger design 2026-09-05 §7.1: the runner behind the three FlowDebug* commands —
|
|
/// standby gate, schema validation before any routing, synthesized inbound row + sink events
|
|
/// on a Live run, the value-change poll loop, and Replay from the repository.
|
|
/// </summary>
|
|
public class FlowDebugRunnerTests
|
|
{
|
|
private static readonly AuthenticatedUser Deployer = new("ops", "Ops", new[] { "Deployer" }, Array.Empty<string>());
|
|
|
|
private readonly ServiceCollection _services = new();
|
|
private readonly IInboundApiRepository _apiRepo = Substitute.For<IInboundApiRepository>();
|
|
private readonly IInstanceLocator _locator = Substitute.For<IInstanceLocator>();
|
|
private readonly IInstanceRouter _router = Substitute.For<IInstanceRouter>();
|
|
private readonly List<AuditEvent> _central = new();
|
|
private readonly IActiveNodeGate _gate = Substitute.For<IActiveNodeGate>();
|
|
private readonly InboundScriptExecutor _executor = new(NullLogger<InboundScriptExecutor>.Instance, Substitute.For<IServiceProvider>());
|
|
private DateTimeOffset _clock = new(2026, 9, 5, 12, 0, 0, TimeSpan.Zero);
|
|
|
|
public FlowDebugRunnerTests()
|
|
{
|
|
_gate.IsActiveNode.Returns(true);
|
|
var writer = Substitute.For<ICentralAuditWriter>();
|
|
writer.WriteAsync(Arg.Do<AuditEvent>(_central.Add), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
|
|
_services.AddSingleton(_apiRepo);
|
|
_services.AddSingleton(_gate);
|
|
_services.AddSingleton(writer);
|
|
_services.AddSingleton(_locator);
|
|
_services.AddSingleton(_router);
|
|
_services.AddSingleton(_executor);
|
|
_services.AddScoped(sp => new RouteHelper(_locator, _router, sp.GetService<ICentralAuditWriter>(), null));
|
|
}
|
|
|
|
private FlowDebugRunner Runner() => new(
|
|
NullLogger<FlowDebugRunner>.Instance,
|
|
(d, _) => { _clock += d; return Task.CompletedTask; },
|
|
() => _clock);
|
|
|
|
private static JsonElement Json(string s) => JsonDocument.Parse(s).RootElement.Clone();
|
|
|
|
[Fact]
|
|
public async Task RunMethod_OnStandby_RefusesWithStandbyCode_BeforeTouchingTheMethod()
|
|
{
|
|
_gate.IsActiveNode.Returns(false);
|
|
var ex = await Assert.ThrowsAsync<ManagementCommandException>(() =>
|
|
Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(1, null), Deployer, CancellationToken.None));
|
|
Assert.Equal(FlowDebugRunner.StandbyNodeCode, ex.Code);
|
|
Assert.Contains("ACTIVE central node", ex.Message);
|
|
await _apiRepo.DidNotReceiveWithAnyArgs().GetApiMethodByIdAsync(default, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_InvalidParameters_FailsBeforeAnyRouteCall()
|
|
{
|
|
var method = new ApiMethod("M", "return 1;")
|
|
{
|
|
Id = 3, TimeoutSeconds = 5,
|
|
ParameterDefinitions = "{\"type\":\"object\",\"properties\":{\"n\":{\"type\":\"integer\"}},\"required\":[\"n\"]}",
|
|
};
|
|
_apiRepo.GetApiMethodByIdAsync(3, Arg.Any<CancellationToken>()).Returns(method);
|
|
|
|
var ex = await Assert.ThrowsAsync<ManagementCommandException>(() =>
|
|
Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(3, Json("{}")), Deployer, CancellationToken.None));
|
|
Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code);
|
|
await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default);
|
|
Assert.Empty(_central);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_HappyPath_ReturnsInboundRow_ThenSinkEvents_Ordered_AndWritesTheInboundRow()
|
|
{
|
|
var method = new ApiMethod("M", "x") { Id = 4, TimeoutSeconds = 5 };
|
|
_apiRepo.GetApiMethodByIdAsync(4, Arg.Any<CancellationToken>()).Returns(method);
|
|
_locator.GetSiteIdForInstanceAsync("inst", Arg.Any<CancellationToken>()).Returns("site-a");
|
|
var child = Guid.NewGuid();
|
|
var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
|
occurredAtUtc: DateTime.UtcNow.AddSeconds(1), executionId: child, target: "work");
|
|
RouteToCallRequest? captured = null;
|
|
_router.RouteToCallAsync("site-a", Arg.Do<RouteToCallRequest>(r => captured = r), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow, child, [siteRun]));
|
|
_executor.RegisterHandler("M", ctx => ctx.Route.To("inst").Call("work"));
|
|
|
|
var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunMethodCommand(4, Json("{\"a\":1}")), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
Assert.Null(result.ErrorCode);
|
|
Assert.True(captured!.Trace);
|
|
Assert.Equal(result.RootExecutionId, captured.ParentExecutionId);
|
|
|
|
// inbound (synthesized) → RouteCall (sink) → site ScriptRun (sink, from the reply)
|
|
Assert.Equal(3, result.Events.Count);
|
|
var inbound = result.Events[0].AsRow();
|
|
Assert.Equal(AuditKind.InboundRequest, inbound.Kind);
|
|
Assert.Equal("ops", result.Events[0].Actor);
|
|
Assert.Equal("M", result.Events[0].Target);
|
|
Assert.Equal(result.RootExecutionId, inbound.ExecutionId);
|
|
Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(inbound.Extra)!.Source);
|
|
Assert.Contains("\"a\":1", inbound.RequestSummary);
|
|
Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind);
|
|
Assert.Same(siteRun, result.Events[2]);
|
|
|
|
// Central writer saw the inbound row AND the always-on RouteCall row.
|
|
Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.InboundRequest);
|
|
Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.RouteCall);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_SiteClockBehindCentral_StillReturnsTheSinkEventsInEmissionOrder()
|
|
{
|
|
// Arch-review F5: the site's rows carry the SITE clock. A site 2 s behind central used to
|
|
// re-sort ahead of the always-on central RouteCall row.
|
|
var method = new ApiMethod("M", "x") { Id = 6, TimeoutSeconds = 5 };
|
|
_apiRepo.GetApiMethodByIdAsync(6, Arg.Any<CancellationToken>()).Returns(method);
|
|
_locator.GetSiteIdForInstanceAsync("inst", Arg.Any<CancellationToken>()).Returns("site-a");
|
|
var child = Guid.NewGuid();
|
|
var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
|
occurredAtUtc: DateTime.UtcNow.AddSeconds(-2), executionId: child, target: "work");
|
|
_router.RouteToCallAsync("site-a", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow, child, [siteRun]));
|
|
_executor.RegisterHandler("M", ctx => ctx.Route.To("inst").Call("work"));
|
|
|
|
var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunMethodCommand(6, null), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
Assert.Equal(3, result.Events.Count);
|
|
Assert.Equal(AuditKind.InboundRequest, result.Events[0].AsRow().Kind);
|
|
Assert.Equal(AuditKind.RouteCall, result.Events[1].AsRow().Kind);
|
|
Assert.Same(siteRun, result.Events[2]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_UnknownMethod_IsNotFound()
|
|
{
|
|
var ex = await Assert.ThrowsAsync<ManagementCommandException>(() =>
|
|
Runner().RunMethodAsync(_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(99, null), Deployer, CancellationToken.None));
|
|
Assert.Equal(FlowDebugRunner.NotFoundCode, ex.Code);
|
|
}
|
|
|
|
private (ITemplateEngineRepository Templates, ISiteRepository Sites, IPullTraceEventsClient Puller) WireValueChange()
|
|
{
|
|
var templates = Substitute.For<ITemplateEngineRepository>();
|
|
var sites = Substitute.For<ISiteRepository>();
|
|
var puller = Substitute.For<IPullTraceEventsClient>();
|
|
templates.GetInstanceByIdAsync(5, Arg.Any<CancellationToken>()).Returns(new Instance("Pump1") { Id = 5, SiteId = 2 });
|
|
sites.GetSiteByIdAsync(2, Arg.Any<CancellationToken>()).Returns(new Site("Site A", "site-a") { Id = 2 });
|
|
_router.RouteToSetAttributesAsync("site-a", Arg.Any<RouteToSetAttributesRequest>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
|
|
_services.AddSingleton(templates);
|
|
_services.AddSingleton(sites);
|
|
_services.AddSingleton(puller);
|
|
return (templates, sites, puller);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunValueChange_TracedWrite_ThenPollsUntilSpawnedRunSettles()
|
|
{
|
|
var (_, _, puller) = WireValueChange();
|
|
RouteToSetAttributesRequest? write = null;
|
|
_router.RouteToSetAttributesAsync("site-a", Arg.Do<RouteToSetAttributesRequest>(r => write = r), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
|
|
var calls = 0;
|
|
puller.PullAsync("site-a", Arg.Any<IReadOnlyCollection<Guid>>(), Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci =>
|
|
{
|
|
calls++;
|
|
var ids = (IReadOnlyCollection<Guid>)ci[1];
|
|
var root = ids.First();
|
|
// Second poll onwards: a ScriptRun parented to the root exists.
|
|
return calls >= 2
|
|
? (IReadOnlyList<AuditEvent>)[ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
|
eventId: Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001"), executionId: Guid.NewGuid(), parentExecutionId: root, target: "OnStatus")]
|
|
: Array.Empty<AuditEvent>();
|
|
});
|
|
|
|
var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 30), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
Assert.True(write!.Trace);
|
|
Assert.Equal(result.RootExecutionId, write.ParentExecutionId);
|
|
Assert.Equal("Pump1", write.InstanceUniqueName);
|
|
Assert.Equal("Running", write.AttributeValues["Status"]);
|
|
Assert.Equal(2, result.Events.Count);
|
|
Assert.Equal(AuditKind.AttributeWrite, result.Events[0].AsRow().Kind);
|
|
Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(result.Events[0].AsRow().Extra)!.Source);
|
|
Assert.Equal(AuditKind.ScriptRun, result.Events[1].AsRow().Kind);
|
|
// Settled: at least one quiet poll (2 s window) after the last new row, then stopped.
|
|
Assert.True(calls >= 3, $"expected settle polls, got {calls}");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunValueChange_NoSpawnWithinWindow_ReportsNoTriggerCode()
|
|
{
|
|
var (_, _, puller) = WireValueChange();
|
|
puller.PullAsync(default!, default!, default, default, default).ReturnsForAnyArgs(Array.Empty<AuditEvent>());
|
|
|
|
var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 3), Deployer, CancellationToken.None);
|
|
|
|
Assert.False(result.Completed);
|
|
Assert.Equal(FlowDebugRunner.NoTriggerCode, result.ErrorCode);
|
|
Assert.Single(result.Events); // just the synthesized write row
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunValueChange_OutOfScopeSite_IsRefused()
|
|
{
|
|
WireValueChange();
|
|
var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "9" });
|
|
await Assert.ThrowsAsync<SiteScopeViolationException>(() => Runner().RunValueChangeAsync(
|
|
_services.BuildServiceProvider(), new FlowDebugRunValueChangeCommand(5, "Status", "x"), scoped, CancellationToken.None));
|
|
await _router.DidNotReceiveWithAnyArgs().RouteToSetAttributesAsync(default!, default!, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Replay_ReturnsRepositoryRows_WithRootFromTheFirstRow()
|
|
{
|
|
var repo = Substitute.For<IAuditLogRepository>();
|
|
var root = Guid.NewGuid();
|
|
var child = Guid.NewGuid();
|
|
var rows = new List<AuditEvent>
|
|
{
|
|
ScadaBridgeAuditEventFactory.Create(AuditChannel.ApiInbound, AuditKind.InboundRequest, AuditStatus.Delivered, executionId: root),
|
|
ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, executionId: child, parentExecutionId: root),
|
|
};
|
|
repo.GetTraceEventsAsync(child, Arg.Any<CancellationToken>()).Returns(rows);
|
|
_services.AddSingleton(repo);
|
|
|
|
var result = await Runner().ReplayAsync(_services.BuildServiceProvider(), new FlowDebugReplayCommand(child), CancellationToken.None);
|
|
|
|
Assert.True(result.Completed);
|
|
Assert.Equal(root, result.RootExecutionId);
|
|
Assert.Equal(rows, result.Events);
|
|
|
|
var empty = await Runner().ReplayAsync(_services.BuildServiceProvider(), new FlowDebugReplayCommand(Guid.NewGuid()), CancellationToken.None);
|
|
Assert.False(empty.Completed);
|
|
Assert.Equal(FlowDebugRunner.NotFoundCode, empty.ErrorCode);
|
|
}
|
|
|
|
// --- Arch-review F2: a method-entry Live run is bound by the operator's site scope ---
|
|
//
|
|
// RunValueChange already called EnforceSiteScope on the target instance; a method
|
|
// Live run has no single target site up front, so the scope is pushed into the
|
|
// RouteHelper and enforced at route time (RouteTarget.ResolveSiteAsync), before any
|
|
// site traffic. The denial comes back as SITE_SCOPE_DENIED and the runner converts it.
|
|
|
|
/// <summary>Wires a method whose script routes to <paramref name="instanceCode"/>, resident at <paramref name="siteIdentifier"/>.</summary>
|
|
private ISiteRepository WireRoutedMethod(int methodId, string instanceCode, string siteIdentifier)
|
|
{
|
|
var method = new ApiMethod("M", "x") { Id = methodId, TimeoutSeconds = 5 };
|
|
_apiRepo.GetApiMethodByIdAsync(methodId, Arg.Any<CancellationToken>()).Returns(method);
|
|
_locator.GetSiteIdForInstanceAsync(instanceCode, Arg.Any<CancellationToken>()).Returns(siteIdentifier);
|
|
_router.RouteToCallAsync(siteIdentifier, Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToCallResponse(
|
|
((RouteToCallRequest)ci[1]).CorrelationId, true, 7, null, DateTimeOffset.UtcNow));
|
|
_executor.RegisterHandler("M", ctx => ctx.Route.To(instanceCode).Call("work"));
|
|
|
|
// The user's PermittedSiteIds are NUMERIC ids; the locator speaks site IDENTIFIERS,
|
|
// so the runner resolves one to the other through the repository.
|
|
var sites = Substitute.For<ISiteRepository>();
|
|
sites.GetSiteByIdAsync(1, Arg.Any<CancellationToken>()).Returns(new Site("Site A", "site-a") { Id = 1 });
|
|
_services.AddSingleton(sites);
|
|
return sites;
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_ScopedDeployer_RouteToOtherSite_IsUnauthorized_AndInboundRowIsWritten()
|
|
{
|
|
WireRoutedMethod(11, "inst", "site-b");
|
|
var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "1" });
|
|
|
|
await Assert.ThrowsAsync<SiteScopeViolationException>(() => Runner().RunMethodAsync(
|
|
_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(11, null), scoped, CancellationToken.None));
|
|
|
|
// The site was never contacted...
|
|
await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default);
|
|
// ...but the denied attempt IS recorded against the operator's name.
|
|
Assert.Contains(_central, e => e.AsRow().Kind == AuditKind.InboundRequest && e.Actor == "ops");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_ScopedDeployer_RouteWithinScope_Runs()
|
|
{
|
|
// The same scoped user, routing to a site their numeric id DOES resolve to.
|
|
WireRoutedMethod(12, "inst", "site-a");
|
|
var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "1" });
|
|
|
|
var result = await Runner().RunMethodAsync(
|
|
_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(12, null), scoped, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
await _router.ReceivedWithAnyArgs(1).RouteToCallAsync(default!, default!, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_Administrator_IsNeverScoped()
|
|
{
|
|
WireRoutedMethod(13, "inst", "site-b");
|
|
var admin = new AuthenticatedUser("boss", "Boss", new[] { "Deployer", "Administrator" }, new[] { "1" });
|
|
|
|
var result = await Runner().RunMethodAsync(
|
|
_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(13, null), admin, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
await _router.ReceivedWithAnyArgs(1).RouteToCallAsync(default!, default!, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_UnscopedUser_IsNeverScoped()
|
|
{
|
|
WireRoutedMethod(14, "inst", "site-b");
|
|
|
|
// Deployer with an EMPTY PermittedSiteIds = global scope, as EnforceSiteScope treats it.
|
|
var result = await Runner().RunMethodAsync(
|
|
_services.BuildServiceProvider(), new FlowDebugRunMethodCommand(14, null), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
await _router.ReceivedWithAnyArgs(1).RouteToCallAsync(default!, default!, default);
|
|
}
|
|
|
|
[Fact]
|
|
public void RoleGate_TheThreeRunCommands_AreDeployerOnly()
|
|
{
|
|
// arch-review F3: a Live run executes scripts and writes attributes, so it stays Deployer-only.
|
|
foreach (var cmd in new object[]
|
|
{
|
|
new FlowDebugRunMethodCommand(1, null),
|
|
new FlowDebugRunValueChangeCommand(1, "a", "b"),
|
|
new FlowDebugRunScriptCommand(1, "Manual", null, null),
|
|
})
|
|
Assert.Equal(new[] { "Deployer" }, ManagementActor.GetRequiredRoles(cmd));
|
|
}
|
|
|
|
[Fact]
|
|
public void RoleGate_Replay_AlsoAcceptsTheOperationalAuditRoles()
|
|
{
|
|
// arch-review F3: Replay only reads the audit chain, and the Audit Log / Execution
|
|
// Chain drill-ins must work for an auditor who holds no Deployer role.
|
|
var actual = ManagementActor.GetRequiredRoles(new FlowDebugReplayCommand(Guid.NewGuid()));
|
|
Assert.NotNull(actual);
|
|
Assert.Equal(new[] { "Administrator", "Deployer", "Viewer" }, actual!.OrderBy(r => r, StringComparer.Ordinal));
|
|
}
|
|
|
|
// --- Arch-review F1: the Live result is the REDACTED, capped trace copy ---
|
|
//
|
|
// The runner collects sink events through the same FlowTraceCollector the site launcher
|
|
// uses, so central and site filter identically and a Live run shows exactly the rows a
|
|
// later Replay reads back out of the audit store.
|
|
|
|
/// <summary>Wires a method whose handler pushes <paramref name="events"/> straight at the trace sink.</summary>
|
|
private void WireSinkPushingMethod(int methodId, IEnumerable<AuditEvent> events)
|
|
{
|
|
var method = new ApiMethod("M", "x") { Id = methodId, TimeoutSeconds = 5 };
|
|
_apiRepo.GetApiMethodByIdAsync(methodId, Arg.Any<CancellationToken>()).Returns(method);
|
|
_executor.RegisterHandler("M", ctx =>
|
|
{
|
|
foreach (var e in events) ctx.Route.TraceSink!(e);
|
|
return Task.FromResult<object?>(null);
|
|
});
|
|
}
|
|
|
|
private static AuditEvent TraceRow(string? detailsJson) =>
|
|
ScadaBridgeAuditEventFactory.Create(
|
|
AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
|
executionId: Guid.NewGuid(), target: "work") with
|
|
{ DetailsJson = detailsJson };
|
|
|
|
[Fact]
|
|
public async Task RunMethod_SinkEvents_AreRedacted_WhenARedactorIsRegistered()
|
|
{
|
|
WireSinkPushingMethod(31, [TraceRow("{\"secret\":\"abc\"}")]);
|
|
_services.AddSingleton<IAuditRedactor>(new UpperRedactor());
|
|
|
|
var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunMethodCommand(31, null), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
Assert.Equal(2, result.Events.Count); // inbound row + the one sink event
|
|
Assert.Equal("{\"SECRET\":\"ABC\"}", result.Events[1].DetailsJson);
|
|
Assert.False(result.Truncated);
|
|
// Review follow-up: the row the RUNNER synthesizes is redacted too — the central writer
|
|
// only redacts its own private copy, so the caller's copy would otherwise be the raw one.
|
|
var written = Assert.Single(_central, e => e.AsRow().Kind == AuditKind.InboundRequest);
|
|
Assert.Equal(written.DetailsJson!.ToUpperInvariant(), result.Events[0].DetailsJson);
|
|
Assert.NotEqual(written.DetailsJson, result.Events[0].DetailsJson); // the raw row still reaches the writer
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunValueChange_SynthesizedRootRow_IsRedacted_WhenARedactorIsRegistered()
|
|
{
|
|
var (_, _, puller) = WireValueChange();
|
|
puller.PullAsync(default!, default!, default, default, default).ReturnsForAnyArgs(Array.Empty<AuditEvent>());
|
|
_services.AddSingleton<IAuditRedactor>(new UpperRedactor());
|
|
|
|
var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 1), Deployer, CancellationToken.None);
|
|
|
|
var root = Assert.Single(result.Events);
|
|
var written = Assert.Single(_central);
|
|
Assert.Equal(written.DetailsJson!.ToUpperInvariant(), root.DetailsJson);
|
|
Assert.NotEqual(written.DetailsJson, root.DetailsJson);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunValueChange_FailedWrite_ReturnsTheRedactedRootRow()
|
|
{
|
|
WireValueChange();
|
|
_router.RouteToSetAttributesAsync("site-a", Arg.Any<RouteToSetAttributesRequest>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, false, "no such attribute", DateTimeOffset.UtcNow));
|
|
_services.AddSingleton<IAuditRedactor>(new UpperRedactor());
|
|
|
|
var result = await Runner().RunValueChangeAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunValueChangeCommand(5, "Status", "Running", WaitSeconds: 1), Deployer, CancellationToken.None);
|
|
|
|
Assert.False(result.Completed);
|
|
Assert.Equal("WRITE_FAILED", result.ErrorCode);
|
|
var root = Assert.Single(result.Events);
|
|
var written = Assert.Single(_central);
|
|
Assert.Equal(written.DetailsJson!.ToUpperInvariant(), root.DetailsJson);
|
|
Assert.NotEqual(written.DetailsJson, root.DetailsJson);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunMethod_OverCap_ReportsTruncated()
|
|
{
|
|
WireSinkPushingMethod(32, Enumerable.Range(0, 501).Select(_ => TraceRow("{}")).ToList());
|
|
|
|
var result = await Runner().RunMethodAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunMethodCommand(32, null), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Truncated);
|
|
Assert.Equal(501, result.Events.Count); // inbound row + the 500 kept
|
|
}
|
|
|
|
// --- Flow Debugger script entry (design 2026-09-08) ---
|
|
//
|
|
// The runner resolves the script from the FLATTENED configuration and branches on its
|
|
// server-side trigger kind: ValueChange/Conditional fire by WRITING the monitored attribute
|
|
// (the shared WriteAndPollAsync helper), every other kind is a direct traced routed call
|
|
// with ONE operator-attributed RouteCall root row.
|
|
|
|
/// <summary>Wires instance 5 / site-a like <see cref="WireValueChange"/> plus a flattened configuration holding the given scripts.</summary>
|
|
private (ITemplateEngineRepository Templates, ISiteRepository Sites, IPullTraceEventsClient Puller, IFlatteningPipeline Pipeline)
|
|
WireScript(params ResolvedScript[] scripts)
|
|
{
|
|
var (templates, sites, puller) = WireValueChange();
|
|
var pipeline = Substitute.For<IFlatteningPipeline>();
|
|
var config = new FlattenedConfiguration { InstanceUniqueName = "Pump1", TemplateId = 10, SiteId = 2, Scripts = scripts };
|
|
pipeline.FlattenAndValidateAsync(5, Arg.Any<CancellationToken>(), false, null)
|
|
.Returns(Result<FlatteningPipelineResult>.Success(new FlatteningPipelineResult(config, "hash", new ValidationResult())));
|
|
_services.AddSingleton(pipeline);
|
|
return (templates, sites, puller, pipeline);
|
|
}
|
|
|
|
private static ResolvedScript CallScript(string name, string? parameterDefinitions = null, int? timeout = null) =>
|
|
new() { CanonicalName = name, Code = "return 1;", TriggerType = "Call", ParameterDefinitions = parameterDefinitions, ExecutionTimeoutSeconds = timeout };
|
|
|
|
private static ResolvedScript WriteScript(string name, string kind, string attribute) =>
|
|
new() { CanonicalName = name, Code = "return 1;", TriggerType = kind, TriggerConfiguration = $$"""{"attributeName":"{{attribute}}"}""" };
|
|
|
|
[Fact]
|
|
public async Task RunScript_WriteKind_WritesTheTriggerAttribute_ThenPolls()
|
|
{
|
|
var (_, _, puller, _) = WireScript(WriteScript("OnLevel", "Conditional", "Level"));
|
|
RouteToSetAttributesRequest? write = null;
|
|
_router.RouteToSetAttributesAsync("site-a", Arg.Do<RouteToSetAttributesRequest>(r => write = r), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow));
|
|
puller.PullAsync("site-a", Arg.Any<IReadOnlyCollection<Guid>>(), Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => (IReadOnlyList<AuditEvent>)[ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
|
eventId: Guid.Parse("bbbbbbbb-0000-0000-0000-000000000001"),
|
|
executionId: Guid.NewGuid(), parentExecutionId: ((IReadOnlyCollection<Guid>)ci[1]).First(), target: "OnLevel")]);
|
|
|
|
var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "onlevel", null, "7", WaitSeconds: 10), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
Assert.Equal("7", write!.AttributeValues["Level"]);
|
|
Assert.True(write.Trace);
|
|
Assert.Equal(AuditKind.AttributeWrite, result.Events[0].AsRow().Kind);
|
|
await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_WriteKind_WithoutValue_FailsValidation_BeforeAnyWrite()
|
|
{
|
|
WireScript(WriteScript("OnLevel", "ValueChange", "Level"));
|
|
var ex = await Assert.ThrowsAsync<ManagementCommandException>(() => Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "OnLevel", null, null), Deployer, CancellationToken.None));
|
|
Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code);
|
|
await _router.DidNotReceiveWithAnyArgs().RouteToSetAttributesAsync(default!, default!, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_CallKind_RoutesATracedCall_AndReturnsRootRouteCallRow_ThenSiteRows()
|
|
{
|
|
WireScript(CallScript("Manual", """{"type":"object","properties":{"n":{"type":"integer"}},"required":["n"]}""", timeout: 5));
|
|
var child = Guid.NewGuid();
|
|
var siteRun = ScadaBridgeAuditEventFactory.Create(AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered,
|
|
executionId: child, target: "Manual");
|
|
RouteToCallRequest? captured = null;
|
|
_router.RouteToCallAsync("site-a", Arg.Do<RouteToCallRequest>(r => captured = r), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 42, null, DateTimeOffset.UtcNow, child, [siteRun]));
|
|
|
|
var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "Manual", Json("{\"n\":3}"), null), Deployer, CancellationToken.None);
|
|
|
|
Assert.True(result.Completed, result.Error);
|
|
Assert.Null(result.ErrorCode);
|
|
Assert.True(captured!.Trace);
|
|
Assert.Equal("Pump1", captured.InstanceUniqueName);
|
|
Assert.Equal("Manual", captured.ScriptName);
|
|
Assert.Equal(result.RootExecutionId, captured.ParentExecutionId);
|
|
Assert.Equal(3L, Convert.ToInt64(captured.Parameters!["n"]));
|
|
|
|
Assert.Equal(2, result.Events.Count);
|
|
var root = result.Events[0].AsRow();
|
|
Assert.Equal(AuditKind.RouteCall, root.Kind);
|
|
Assert.Equal(result.RootExecutionId, root.ExecutionId);
|
|
Assert.Equal("ops", result.Events[0].Actor);
|
|
Assert.Equal("Pump1.Manual", result.Events[0].Target);
|
|
Assert.Equal("site-a", root.SourceSiteId);
|
|
Assert.Equal("flow-debugger", ScriptTraceExtra.TryParse(root.Extra)!.Source);
|
|
Assert.Equal(child, ScriptTraceExtra.TryParse(root.Extra)!.ChildExecutionId);
|
|
Assert.Contains("\"n\":3", root.RequestSummary);
|
|
Assert.Contains("42", root.ResponseSummary);
|
|
Assert.Same(siteRun, result.Events[1]);
|
|
Assert.Single(_central); // the root row was written centrally
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_CallKind_InvalidParameters_FailsBeforeRouting()
|
|
{
|
|
WireScript(CallScript("Manual", """{"type":"object","properties":{"n":{"type":"integer"}},"required":["n"]}"""));
|
|
var ex = await Assert.ThrowsAsync<ManagementCommandException>(() => Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "Manual", Json("{}"), null), Deployer, CancellationToken.None));
|
|
Assert.Equal(FlowDebugRunner.ValidationFailedCode, ex.Code);
|
|
await _router.DidNotReceiveWithAnyArgs().RouteToCallAsync(default!, default!, default);
|
|
Assert.Empty(_central);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_CallKind_SiteFailure_ReportsCallFailed_WithTheRootRow()
|
|
{
|
|
WireScript(CallScript("Manual"));
|
|
_router.RouteToCallAsync("site-a", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, false, null, "Script 'Manual' not found on instance 'Pump1'.", DateTimeOffset.UtcNow));
|
|
|
|
var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "Manual", null, null), Deployer, CancellationToken.None);
|
|
|
|
Assert.False(result.Completed);
|
|
Assert.Equal(FlowDebugRunner.CallFailedCode, result.ErrorCode);
|
|
Assert.Contains("not found", result.Error);
|
|
Assert.Single(result.Events);
|
|
Assert.Equal(AuditStatus.Failed, result.Events[0].AsRow().Status);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_UnknownScript_IsNotFound()
|
|
{
|
|
WireScript(CallScript("Manual"));
|
|
var ex = await Assert.ThrowsAsync<ManagementCommandException>(() => Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "Nope", null, null), Deployer, CancellationToken.None));
|
|
Assert.Equal(FlowDebugRunner.NotFoundCode, ex.Code);
|
|
Assert.Contains("Nope", ex.Message);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_OutOfScopeSite_IsRefused_BeforeFlattening()
|
|
{
|
|
var (_, _, _, pipeline) = WireScript(CallScript("Manual"));
|
|
var scoped = new AuthenticatedUser("ops", "Ops", new[] { "Deployer" }, new[] { "9" });
|
|
await Assert.ThrowsAsync<SiteScopeViolationException>(() => Runner().RunScriptAsync(
|
|
_services.BuildServiceProvider(), new FlowDebugRunScriptCommand(5, "Manual", null, null), scoped, CancellationToken.None));
|
|
await pipeline.DidNotReceiveWithAnyArgs().FlattenAndValidateAsync(default, default, default, default);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RunScript_CallKind_RootRow_IsRedacted_WhenARedactorIsRegistered()
|
|
{
|
|
WireScript(CallScript("Manual"));
|
|
_router.RouteToCallAsync("site-a", Arg.Any<RouteToCallRequest>(), Arg.Any<CancellationToken>())
|
|
.Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 1, null, DateTimeOffset.UtcNow));
|
|
_services.AddSingleton<IAuditRedactor>(new UpperRedactor());
|
|
|
|
var result = await Runner().RunScriptAsync(_services.BuildServiceProvider(),
|
|
new FlowDebugRunScriptCommand(5, "Manual", null, null), Deployer, CancellationToken.None);
|
|
|
|
var root = Assert.Single(result.Events);
|
|
var written = Assert.Single(_central);
|
|
Assert.Equal(written.DetailsJson!.ToUpperInvariant(), root.DetailsJson);
|
|
Assert.NotEqual(written.DetailsJson, root.DetailsJson);
|
|
}
|
|
|
|
/// <summary>Upper-cases <c>DetailsJson</c> so a test can see the filtered copy land.</summary>
|
|
private sealed class UpperRedactor : IAuditRedactor
|
|
{
|
|
public AuditEvent Apply(AuditEvent rawEvent)
|
|
=> rawEvent with { DetailsJson = rawEvent.DetailsJson?.ToUpperInvariant() };
|
|
}
|
|
}
|