Joseph Doherty
e79452d2de
Every emit site hands FlowTraceCollector the same AuditEvent it hands the writer, but redaction and payload capping lived inside the writers, which return a NEW filtered record — so the inline copy that rides ScriptCallResult.TraceEvents / RouteToCallResponse.TraceEvents was the RAW event: no per-target SQL-parameter redaction, no header scrubbing, no 8 KB / 64 KB truncation. Live and Replay of the same execution could therefore show different payloads, and the 500-event-only cap left the reply free to breach the 4 MB gRPC receive limit — which would fail a real call, since the same reply carries the script's return value. The collector now applies the host IAuditRedactor itself (outside the lock; the writers keep their own idempotent Apply) and enforces a byte budget (DefaultMaxBytes = 1 MB over the redacted DetailsJson) alongside the event cap, flagging Truncated when either trips. It moves to Commons (Types/Debugging) so the central FlowDebugRunner can share it. The redactor is a singleton, so ScriptRunLauncher resolves it from the ROOT provider and keeps constructing the collector before the try — the traced ScriptRun row is emitted from the catch blocks too, including the shed-at-dequeue path that throws before the DI scope exists. Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
560 lines
25 KiB
C#
560 lines
25 KiB
C#
using Akka.Actor;
|
|
using Akka.TestKit.Xunit2;
|
|
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Moq;
|
|
using ZB.MOM.WW.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
|
|
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 IAuditWriter = ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.IAuditWriter;
|
|
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
|
|
using ZB.MOM.WW.ScadaBridge.StoreAndForward;
|
|
using ZB.MOM.WW.ScadaBridge.TestSupport;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Scripts;
|
|
|
|
/// <summary>
|
|
/// Flow Debugger design 2026-09-05 §5.1/§5.2: a traced <see cref="ScriptRuntimeContext"/>
|
|
/// emits <c>ScriptTrace</c> rows for attribute writes and waits (into the inline collector
|
|
/// AND the audit writer), stamps <c>Trace</c> on the commands it sends, and inherits the flag
|
|
/// into nested calls. An untraced context is byte-for-byte the pre-feature behaviour.
|
|
/// Outbound-calls addendum (plan 2026-09-06 Task 3): the <c>ApiOutbound</c> /
|
|
/// <c>DbOutbound</c> / <c>Notification</c> rows a run emits ride the same collector — the
|
|
/// SAME <see cref="AuditEvent"/> instance the writer / telemetry forwarder receives — and
|
|
/// nothing lands after the collector is <see cref="FlowTraceCollector.Complete"/>d.
|
|
/// </summary>
|
|
public class ScriptRuntimeContextTraceEmissionTests : TestKit
|
|
{
|
|
private const string InstanceName = "Plant.Pump42";
|
|
|
|
private sealed class CapturingAuditWriter : IAuditWriter
|
|
{
|
|
public List<AuditEvent> Events { get; } = new();
|
|
public Task WriteAsync(AuditEvent evt, CancellationToken ct = default)
|
|
{
|
|
lock (Events) Events.Add(evt);
|
|
return Task.CompletedTask;
|
|
}
|
|
}
|
|
|
|
private sealed class CapturingForwarder : ICachedCallTelemetryForwarder
|
|
{
|
|
public List<CachedCallTelemetry> Telemetry { get; } = new();
|
|
public Task ForwardAsync(CachedCallTelemetry telemetry, CancellationToken ct = default)
|
|
{
|
|
Telemetry.Add(telemetry);
|
|
return Task.CompletedTask;
|
|
}
|
|
}
|
|
|
|
/// <summary>A real S&F service over a temp LocalDb file — <c>Notify.Send</c> needs the
|
|
/// concrete <see cref="StoreAndForwardService"/>, there is no interface to stub.</summary>
|
|
private sealed class TempStoreAndForward : IDisposable
|
|
{
|
|
private readonly TestLocalDb _localDb;
|
|
public StoreAndForwardService Service { get; }
|
|
|
|
private TempStoreAndForward(TestLocalDb localDb, StoreAndForwardService service)
|
|
{
|
|
_localDb = localDb;
|
|
Service = service;
|
|
}
|
|
|
|
public static async Task<TempStoreAndForward> CreateAsync()
|
|
{
|
|
var localDb = TestLocalDb.CreateTemp("TraceEmissionNotify");
|
|
var storage = new StoreAndForwardStorage(localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
|
|
await storage.InitializeAsync();
|
|
var options = new StoreAndForwardOptions
|
|
{
|
|
DefaultRetryInterval = TimeSpan.Zero,
|
|
DefaultMaxRetries = 3,
|
|
RetryTimerInterval = TimeSpan.FromMinutes(10),
|
|
};
|
|
return new TempStoreAndForward(localDb,
|
|
new StoreAndForwardService(storage, options, NullLogger<StoreAndForwardService>.Instance));
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
_localDb.Dispose();
|
|
TestLocalDb.DeleteFiles(_localDb.Path);
|
|
}
|
|
}
|
|
|
|
private static Mock<IExternalSystemClient> SyncClient(ExternalCallResult result)
|
|
{
|
|
var client = new Mock<IExternalSystemClient>();
|
|
client
|
|
.Setup(c => c.CallAsync("ERP", "GetOrder", It.IsAny<IReadOnlyDictionary<string, object?>?>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(result);
|
|
return client;
|
|
}
|
|
|
|
private static Mock<IExternalSystemClient> CachedClient(ExternalCallResult result)
|
|
{
|
|
var client = new Mock<IExternalSystemClient>();
|
|
client
|
|
.Setup(c => c.CachedCallAsync(
|
|
"ERP", "GetOrder",
|
|
It.IsAny<IReadOnlyDictionary<string, object?>?>(),
|
|
InstanceName,
|
|
It.IsAny<CancellationToken>(),
|
|
It.IsAny<TrackedOperationId?>(),
|
|
It.IsAny<Guid?>(), It.IsAny<string?>(), It.IsAny<Guid?>()))
|
|
.ReturnsAsync(result);
|
|
return client;
|
|
}
|
|
|
|
/// <summary>In-memory SQLite handed out by a stub gateway; the keep-alive root must outlive
|
|
/// the auditing wrapper so the database survives the wrapper's dispose.</summary>
|
|
private static Mock<IDatabaseGateway> SqliteGateway(out SqliteConnection keepAlive)
|
|
{
|
|
var connStr = $"Data Source=trace-{Guid.NewGuid():N};Mode=Memory;Cache=Shared";
|
|
keepAlive = new SqliteConnection(connStr);
|
|
keepAlive.Open();
|
|
using (var seed = keepAlive.CreateCommand())
|
|
{
|
|
seed.CommandText = "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT NOT NULL);";
|
|
seed.ExecuteNonQuery();
|
|
}
|
|
var live = new SqliteConnection(connStr);
|
|
live.Open();
|
|
var gateway = new Mock<IDatabaseGateway>();
|
|
gateway
|
|
.Setup(g => g.GetConnectionAsync("machineData", It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(live);
|
|
return gateway;
|
|
}
|
|
|
|
private static AuditRowProjection.AuditRowValues Row(AuditEvent evt) => evt.AsRow();
|
|
|
|
private static SharedScriptLibrary NewLibrary() =>
|
|
new(new ScriptCompilationService(NullLogger<ScriptCompilationService>.Instance),
|
|
NullLogger<SharedScriptLibrary>.Instance);
|
|
|
|
private static ScriptRuntimeContext CreateContext(
|
|
IActorRef instanceActor, IAuditWriter? writer, bool traced, FlowTraceCollector? collector,
|
|
Guid? executionId = null, Guid? parentExecutionId = null,
|
|
IExternalSystemClient? externalSystemClient = null,
|
|
IDatabaseGateway? databaseGateway = null,
|
|
ICachedCallTelemetryForwarder? cachedForwarder = null,
|
|
StoreAndForwardService? storeAndForward = null)
|
|
=> new(
|
|
instanceActor, NewLibrary(),
|
|
currentCallDepth: 0, maxCallDepth: 10, askTimeout: TimeSpan.FromSeconds(5),
|
|
instanceName: InstanceName, logger: NullLogger.Instance,
|
|
externalSystemClient: externalSystemClient, databaseGateway: databaseGateway,
|
|
storeAndForward: storeAndForward,
|
|
siteId: "site-a", sourceScript: "ScriptActor:A", auditWriter: writer,
|
|
cachedForwarder: cachedForwarder,
|
|
executionId: executionId, parentExecutionId: parentExecutionId, sourceNode: "node-a",
|
|
isTraced: traced, traceCollector: collector);
|
|
|
|
[Fact]
|
|
public async Task SetAttribute_WhenTraced_EmitsAttributeWriteRow_AndPassesTraceOnCommand()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector();
|
|
var exec = Guid.NewGuid();
|
|
var parent = Guid.NewGuid();
|
|
var ctx = CreateContext(probe.Ref, writer, traced: true, collector, exec, parent);
|
|
|
|
var call = ctx.SetAttribute("Setpoint", "42");
|
|
var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
|
|
Assert.True(cmd.Trace);
|
|
Assert.Equal(exec, cmd.SourceExecutionId);
|
|
probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", true, null, DateTimeOffset.UtcNow));
|
|
await call;
|
|
|
|
var evt = Assert.Single(writer.Events);
|
|
var row = evt.AsRow();
|
|
Assert.Equal("ScriptTrace", evt.Category);
|
|
Assert.Equal("ScriptTrace.AttributeWrite", evt.Action);
|
|
Assert.Equal(exec, row.ExecutionId);
|
|
Assert.Equal(parent, row.ParentExecutionId);
|
|
Assert.Equal("site-a", row.SourceSiteId);
|
|
Assert.Equal("node-a", evt.SourceNode);
|
|
Assert.Equal("Setpoint", evt.Target);
|
|
Assert.Equal(AuditOutcome.Success, evt.Outcome);
|
|
var extra = ScriptTraceExtra.TryParse(row.Extra);
|
|
var write = Assert.Single(extra!.Writes!);
|
|
Assert.Equal(("Setpoint", "42", true), (write.Attribute, write.Written, write.Success));
|
|
Assert.Contains("\"value\":\"42\"", row.RequestSummary);
|
|
|
|
Assert.Same(evt, Assert.Single(collector.Snapshot()));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SetAttribute_WhenNotTraced_EmitsNothing_AndCommandTraceFalse()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var ctx = CreateContext(probe.Ref, writer, traced: false, collector: null);
|
|
|
|
var call = ctx.SetAttribute("Setpoint", "42");
|
|
var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
|
|
Assert.False(cmd.Trace);
|
|
probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", true, null, DateTimeOffset.UtcNow));
|
|
await call;
|
|
|
|
Assert.Empty(writer.Events);
|
|
Assert.False(ctx.IsTraced);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Arch-review F1: the collector applies the host redactor, so the inline copy the reply
|
|
/// carries is the FILTERED record — while the writer still receives the raw event and
|
|
/// redacts it itself (redaction is idempotent, so the stored row is identical).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task SetAttribute_WhenTraced_InlineCopyIsRedacted_WriterStillSeesTheRawEvent()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector(redactor: new UpperRedactor());
|
|
var ctx = CreateContext(probe.Ref, writer, traced: true, collector);
|
|
|
|
var call = ctx.SetAttribute("Setpoint", "42");
|
|
var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
|
|
probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", true, null, DateTimeOffset.UtcNow));
|
|
await call;
|
|
|
|
var written = Assert.Single(writer.Events);
|
|
var collected = Assert.Single(collector.Complete());
|
|
Assert.NotSame(written, collected);
|
|
Assert.Contains("\"channel\":\"ScriptTrace\"", written.DetailsJson);
|
|
Assert.NotEqual(written.DetailsJson, collected.DetailsJson);
|
|
Assert.Equal(written.DetailsJson!.ToUpperInvariant(), collected.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() };
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SetAttribute_WhenTraced_FailedWrite_EmitsFailedRow_ThenThrows()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var ctx = CreateContext(probe.Ref, writer, traced: true, new FlowTraceCollector());
|
|
|
|
var call = ctx.SetAttribute("Setpoint", "42");
|
|
var cmd = probe.ExpectMsg<SetStaticAttributeCommand>(TimeSpan.FromSeconds(5));
|
|
probe.Reply(new SetStaticAttributeResponse(cmd.CorrelationId, InstanceName, "Setpoint", false, "device rejected", DateTimeOffset.UtcNow));
|
|
|
|
await Assert.ThrowsAsync<InvalidOperationException>(() => call);
|
|
var row = Assert.Single(writer.Events).AsRow();
|
|
Assert.Equal("Failed", row.Status.ToString());
|
|
Assert.Equal("device rejected", row.ErrorMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WaitInternal_WhenTraced_EmitsAttributeWait_WithOutcomeSatisfied()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(probe.Ref, writer, traced: true, collector);
|
|
|
|
var wait = ctx.WaitAttributeFull("Processed", "true", null, TimeSpan.FromSeconds(3));
|
|
var req = probe.ExpectMsg<WaitForAttributeRequest>(TimeSpan.FromSeconds(5));
|
|
probe.Reply(new WaitForAttributeResponse(req.CorrelationId, Matched: true, true, "Good", TimedOut: false));
|
|
var result = await wait;
|
|
|
|
Assert.True(result.Matched);
|
|
var evt = Assert.Single(writer.Events);
|
|
Assert.Equal("ScriptTrace.AttributeWait", evt.Action);
|
|
Assert.Equal("Processed", evt.Target);
|
|
var extra = ScriptTraceExtra.TryParse(evt.AsRow().Extra)!;
|
|
Assert.Equal("satisfied", extra.WaitOutcome);
|
|
Assert.Equal("Processed", extra.WaitAttribute);
|
|
Assert.Equal("true", extra.WaitTarget);
|
|
Assert.Equal(3000, extra.WaitTimeoutMs);
|
|
Assert.Equal("true", extra.ObservedValue, ignoreCase: true);
|
|
Assert.Single(collector.Snapshot());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WaitInternal_WhenTraced_Timeout_EmitsFailedRow_WithTimeoutOutcome()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var ctx = CreateContext(probe.Ref, writer, traced: true, new FlowTraceCollector());
|
|
|
|
var wait = ctx.WaitAttribute("Processed", "true", null, TimeSpan.FromSeconds(1));
|
|
var req = probe.ExpectMsg<WaitForAttributeRequest>(TimeSpan.FromSeconds(5));
|
|
probe.Reply(new WaitForAttributeResponse(req.CorrelationId, Matched: false, null, null, TimedOut: true));
|
|
|
|
Assert.False(await wait);
|
|
var row = Assert.Single(writer.Events).AsRow();
|
|
Assert.Equal("Failed", row.Status.ToString());
|
|
Assert.Contains("timed out", row.ErrorMessage);
|
|
Assert.Equal("timeout", ScriptTraceExtra.TryParse(row.Extra)!.WaitOutcome);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WaitInternal_WhenNotTraced_EmitsNothing()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var writer = new CapturingAuditWriter();
|
|
var ctx = CreateContext(probe.Ref, writer, traced: false, collector: null);
|
|
|
|
var wait = ctx.WaitAttribute("Processed", "true", null, TimeSpan.FromSeconds(1));
|
|
var req = probe.ExpectMsg<WaitForAttributeRequest>(TimeSpan.FromSeconds(5));
|
|
probe.Reply(new WaitForAttributeResponse(req.CorrelationId, Matched: true, true, "Good", TimedOut: false));
|
|
await wait;
|
|
|
|
Assert.Empty(writer.Events);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task CallScript_WhenTraced_PropagatesTraceOnScriptCallRequest()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var ctx = CreateContext(probe.Ref, null, traced: true, new FlowTraceCollector());
|
|
|
|
var call = ctx.CallScript("B");
|
|
var req = probe.ExpectMsg<ScriptCallRequest>(TimeSpan.FromSeconds(5));
|
|
Assert.True(req.Trace);
|
|
probe.Reply(new ScriptCallResult(req.CorrelationId, true, null, null));
|
|
await call;
|
|
|
|
// The inline shared-script child context inherits the flag too.
|
|
Assert.True(ctx.CreateChildContextForSharedScript(1).IsTraced);
|
|
Assert.False(CreateContext(probe.Ref, null, traced: false, null).CreateChildContextForSharedScript(1).IsTraced);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task CallScript_WhenNotTraced_RequestTraceIsFalse()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var ctx = CreateContext(probe.Ref, null, traced: false, null);
|
|
|
|
var call = ctx.CallScript("B");
|
|
var req = probe.ExpectMsg<ScriptCallRequest>(TimeSpan.FromSeconds(5));
|
|
Assert.False(req.Trace);
|
|
probe.Reply(new ScriptCallResult(req.CorrelationId, true, null, null));
|
|
await call;
|
|
}
|
|
|
|
// ---- Outbound rows ride the collector (plan 2026-09-06 Task 3) -------------------------
|
|
|
|
[Fact]
|
|
public async Task ExternalSystemCall_WhenTraced_CollectsTheSameApiCallRow_TheWriterReceives()
|
|
{
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector();
|
|
var exec = Guid.NewGuid();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector, exec,
|
|
externalSystemClient: SyncClient(new ExternalCallResult(true, "{}", null)).Object);
|
|
|
|
var result = await ctx.ExternalSystem.Call("ERP", "GetOrder");
|
|
|
|
Assert.True(result.Success);
|
|
var written = Assert.Single(writer.Events);
|
|
var collected = Assert.Single(collector.Snapshot());
|
|
Assert.Same(written, collected);
|
|
Assert.Equal(AuditChannel.ApiOutbound, Row(collected).Channel);
|
|
Assert.Equal(AuditKind.ApiCall, Row(collected).Kind);
|
|
Assert.Equal(exec, Row(collected).ExecutionId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ExternalSystemCachedCall_Buffered_CollectsCachedSubmitOnly_SameInstanceAsTelemetry()
|
|
{
|
|
var forwarder = new CapturingForwarder();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, new CapturingAuditWriter(), traced: true, collector,
|
|
externalSystemClient: CachedClient(new ExternalCallResult(true, null, null, WasBuffered: true)).Object,
|
|
cachedForwarder: forwarder);
|
|
|
|
var trackedId = await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
|
|
|
|
var packet = Assert.Single(forwarder.Telemetry);
|
|
var collected = Assert.Single(collector.Snapshot());
|
|
Assert.Same(packet.Audit, collected);
|
|
Assert.Equal(AuditKind.CachedSubmit, Row(collected).Kind);
|
|
Assert.Equal(trackedId.Value, Row(collected).CorrelationId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ExternalSystemCachedCall_ImmediateCompletion_CollectsSubmitAttemptedResolve_WhileRunIsOpen()
|
|
{
|
|
var forwarder = new CapturingForwarder();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, new CapturingAuditWriter(), traced: true, collector,
|
|
externalSystemClient: CachedClient(new ExternalCallResult(true, "{\"ok\":true}", null, WasBuffered: false)).Object,
|
|
cachedForwarder: forwarder);
|
|
|
|
await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
|
|
|
|
Assert.Equal(3, forwarder.Telemetry.Count);
|
|
var snap = collector.Snapshot();
|
|
Assert.Equal(3, snap.Count);
|
|
for (var i = 0; i < 3; i++) Assert.Same(forwarder.Telemetry[i].Audit, snap[i]);
|
|
Assert.Equal(
|
|
new[] { AuditKind.CachedSubmit, AuditKind.ApiCallCached, AuditKind.CachedResolve },
|
|
snap.Select(e => Row(e).Kind));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OutboundRows_AfterCollectorCompleted_DoNotTouchTheCollector_ButStillReachTelemetry()
|
|
{
|
|
// The S&F forwarder's later Forwarded/Attempted/Resolve rows run after the launcher
|
|
// has snapshotted TraceEvents. Complete() is that snapshot; anything emitted after it
|
|
// (here: a fast inline resolve on a context the run already returned from) is ignored
|
|
// by the collector while the writer/forwarder still receive every row.
|
|
var forwarder = new CapturingForwarder();
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector,
|
|
externalSystemClient: CachedClient(new ExternalCallResult(true, null, null, WasBuffered: false)).Object,
|
|
cachedForwarder: forwarder);
|
|
|
|
var first = await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
|
|
var snapshot = collector.Complete();
|
|
|
|
Assert.True(collector.Completed);
|
|
Assert.Equal(3, snapshot.Count);
|
|
Assert.Equal(first.Value, Row(snapshot[0]).CorrelationId);
|
|
|
|
await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
|
|
|
|
Assert.Equal(6, forwarder.Telemetry.Count);
|
|
Assert.Equal(3, collector.Count);
|
|
Assert.False(collector.Truncated);
|
|
Assert.Equal(snapshot.Select(e => e.EventId), collector.Snapshot().Select(e => e.EventId));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task DatabaseExecute_WhenTraced_CollectsTheSameDbWriteRow_TheWriterReceives()
|
|
{
|
|
var gateway = SqliteGateway(out var keepAlive);
|
|
using var _ = keepAlive;
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector,
|
|
databaseGateway: gateway.Object);
|
|
|
|
await using (var conn = await ctx.Database.Connection("machineData"))
|
|
await using (var cmd = conn.CreateCommand())
|
|
{
|
|
cmd.CommandText = "INSERT INTO t (id, name) VALUES (1, 'alpha')";
|
|
Assert.Equal(1, await cmd.ExecuteNonQueryAsync());
|
|
}
|
|
|
|
var written = Assert.Single(writer.Events);
|
|
var collected = Assert.Single(collector.Snapshot());
|
|
Assert.Same(written, collected);
|
|
Assert.Equal(AuditChannel.DbOutbound, Row(collected).Channel);
|
|
Assert.Equal(AuditKind.DbWrite, Row(collected).Kind);
|
|
Assert.Contains("\"rowsAffected\":1", Row(collected).Extra);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task DatabaseCachedWrite_Buffered_CollectsCachedSubmit_SameInstanceAsTelemetry()
|
|
{
|
|
var gateway = new Mock<IDatabaseGateway>();
|
|
gateway
|
|
.Setup(g => g.CachedWriteAsync(
|
|
"machineData", "UPDATE t SET name = 'x'",
|
|
It.IsAny<IReadOnlyDictionary<string, object?>?>(), InstanceName,
|
|
It.IsAny<CancellationToken>(), It.IsAny<TrackedOperationId?>(),
|
|
It.IsAny<Guid?>(), It.IsAny<string?>(), It.IsAny<Guid?>()))
|
|
.ReturnsAsync(new ExternalCallResult(true, null, null, WasBuffered: true));
|
|
var forwarder = new CapturingForwarder();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, new CapturingAuditWriter(), traced: true, collector,
|
|
databaseGateway: gateway.Object, cachedForwarder: forwarder);
|
|
|
|
var trackedId = await ctx.Database.CachedWrite("machineData", "UPDATE t SET name = 'x'");
|
|
|
|
var packet = Assert.Single(forwarder.Telemetry);
|
|
var collected = Assert.Single(collector.Snapshot());
|
|
Assert.Same(packet.Audit, collected);
|
|
Assert.Equal(AuditChannel.DbOutbound, Row(collected).Channel);
|
|
Assert.Equal(AuditKind.CachedSubmit, Row(collected).Kind);
|
|
Assert.Equal(trackedId.Value, Row(collected).CorrelationId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task NotifySend_WhenTraced_CollectsTheSameNotifySendRow_TheWriterReceives()
|
|
{
|
|
using var saf = await TempStoreAndForward.CreateAsync();
|
|
var writer = new CapturingAuditWriter();
|
|
var collector = new FlowTraceCollector();
|
|
var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: true, collector,
|
|
storeAndForward: saf.Service);
|
|
|
|
var notificationId = await ctx.Notify.To("Planners").Send("Recipe ready", "R-17 staged");
|
|
|
|
var written = Assert.Single(writer.Events);
|
|
var collected = Assert.Single(collector.Snapshot());
|
|
Assert.Same(written, collected);
|
|
Assert.Equal(AuditChannel.Notification, Row(collected).Channel);
|
|
Assert.Equal(AuditKind.NotifySend, Row(collected).Kind);
|
|
Assert.Equal("Planners", collected.Target);
|
|
Assert.Equal(Guid.Parse(notificationId), Row(collected).CorrelationId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OutboundRows_WhenNoCollector_WriterAndTelemetryStillReceiveEveryRow()
|
|
{
|
|
using var saf = await TempStoreAndForward.CreateAsync();
|
|
var gateway = SqliteGateway(out var keepAlive);
|
|
using var _ = keepAlive;
|
|
var writer = new CapturingAuditWriter();
|
|
var forwarder = new CapturingForwarder();
|
|
var client = CachedClient(new ExternalCallResult(true, null, null, WasBuffered: true));
|
|
client
|
|
.Setup(c => c.CallAsync("ERP", "GetOrder", It.IsAny<IReadOnlyDictionary<string, object?>?>(), It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new ExternalCallResult(true, "{}", null));
|
|
var ctx = CreateContext(CreateTestProbe().Ref, writer, traced: false, collector: null,
|
|
externalSystemClient: client.Object, databaseGateway: gateway.Object,
|
|
cachedForwarder: forwarder, storeAndForward: saf.Service);
|
|
|
|
await ctx.ExternalSystem.Call("ERP", "GetOrder");
|
|
await ctx.ExternalSystem.CachedCall("ERP", "GetOrder");
|
|
await using (var conn = await ctx.Database.Connection("machineData"))
|
|
await using (var cmd = conn.CreateCommand())
|
|
{
|
|
cmd.CommandText = "INSERT INTO t (id, name) VALUES (1, 'alpha')";
|
|
await cmd.ExecuteNonQueryAsync();
|
|
}
|
|
await ctx.Notify.To("Planners").Send("s", "b");
|
|
|
|
Assert.False(ctx.IsTraced);
|
|
Assert.Equal(
|
|
new[] { AuditKind.ApiCall, AuditKind.DbWrite, AuditKind.NotifySend },
|
|
writer.Events.Select(e => Row(e).Kind));
|
|
Assert.Equal(AuditKind.CachedSubmit, Row(Assert.Single(forwarder.Telemetry).Audit).Kind);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ThrowingTraceSink_NeverFailsTheCall_AndTheWriterStillGetsTheRow()
|
|
{
|
|
var writer = new CapturingAuditWriter();
|
|
var helper = new ScriptRuntimeContext.ExternalSystemHelper(
|
|
SyncClient(new ExternalCallResult(true, "{}", null)).Object, InstanceName, NullLogger.Instance, Guid.NewGuid(),
|
|
auditWriter: writer, siteId: "site-a", sourceScript: "S",
|
|
traceSink: _ => throw new InvalidOperationException("observer broke"));
|
|
|
|
var result = await helper.Call("ERP", "GetOrder");
|
|
|
|
Assert.True(result.Success);
|
|
var written = Assert.Single(writer.Events);
|
|
Assert.Equal(AuditKind.ApiCall, Row(written).Kind);
|
|
}
|
|
}
|