Files
ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs
T

455 lines
20 KiB
C#

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site.Telemetry;
/// <summary>
/// Bundle D D1 tests for <see cref="SiteAuditTelemetryActor"/>. The actor drains
/// the site SQLite queue via <see cref="ISiteAuditQueue"/>, pushes batches via
/// <see cref="ISiteStreamAuditClient"/>, and flips ack'd rows to Forwarded.
/// Both collaborators are NSubstitute mocks so the tests never touch real
/// SQLite or gRPC.
/// </summary>
public class SiteAuditTelemetryActorTests : TestKit
{
private readonly ISiteAuditQueue _queue = Substitute.For<ISiteAuditQueue>();
private readonly ISiteStreamAuditClient _client = Substitute.For<ISiteStreamAuditClient>();
private readonly IOperationTrackingStore _trackingStore = Substitute.For<IOperationTrackingStore>();
/// <summary>
/// Fast options so tests don't stall waiting for the scheduler. 1s busy /
/// 2s idle still exercises the busy-vs-idle branching, but each test
/// completes in &lt; 5 s wall-clock.
/// </summary>
private static IOptions<SiteAuditTelemetryOptions> Opts(
int batchSize = 256,
int busySeconds = 1,
int idleSeconds = 2) =>
Options.Create(new SiteAuditTelemetryOptions
{
BatchSize = batchSize,
BusyIntervalSeconds = busySeconds,
IdleIntervalSeconds = idleSeconds,
});
private IActorRef CreateActor(IOptions<SiteAuditTelemetryOptions>? options = null) =>
Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
_queue,
_client,
options ?? Opts(),
NullLogger<SiteAuditTelemetryActor>.Instance,
(IOperationTrackingStore?)null)));
/// <summary>
/// AuditLog-001: builds an actor with the optional
/// <see cref="IOperationTrackingStore"/> wired in so the cached-drain
/// scheduler is armed alongside the audit-only drain. Used by the new
/// cached-drain regression tests below.
/// </summary>
private IActorRef CreateActorWithCachedDrain(IOptions<SiteAuditTelemetryOptions>? options = null) =>
Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
_queue,
_client,
options ?? Opts(),
NullLogger<SiteAuditTelemetryActor>.Instance,
(IOperationTrackingStore?)_trackingStore)));
private static AuditEvent NewEvent(Guid? id = null) => ScadaBridgeAuditEventFactory.Create(
eventId: id ?? Guid.NewGuid(),
occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
channel: AuditChannel.ApiOutbound,
kind: AuditKind.ApiCall,
status: AuditStatus.Delivered,
sourceSiteId: "site-1");
private static IngestAck AckAll(IReadOnlyList<AuditEvent> events)
{
var ack = new IngestAck();
foreach (var e in events)
{
ack.AcceptedEventIds.Add(e.EventId.ToString());
}
return ack;
}
[Fact]
public async Task Drain_With_50PendingRows_Sends_OneBatch_Of_50_Then_FlipsToForwarded()
{
// Arrange — 50 pending rows on the first read, then empty on subsequent
// reads so the actor settles after one productive drain.
var pending = Enumerable.Range(0, 50).Select(_ => NewEvent()).ToList();
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(
Task.FromResult<IReadOnlyList<AuditEvent>>(pending),
Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
AuditEventBatch? capturedBatch = null;
_client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
.Returns(call =>
{
capturedBatch = call.Arg<AuditEventBatch>();
return Task.FromResult(AckAll(pending));
});
// Act
CreateActor();
// Assert — give the scheduler time to fire the initial Drain tick.
await AwaitAssertAsync(async () =>
{
await _client.Received(1).IngestAuditEventsAsync(
Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>());
await _queue.Received(1).MarkForwardedAsync(
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 50), Arg.Any<CancellationToken>());
}, TimeSpan.FromSeconds(5));
Assert.NotNull(capturedBatch);
Assert.Equal(50, capturedBatch!.Events.Count);
var expected = pending.Select(e => e.EventId).ToHashSet();
await _queue.Received(1).MarkForwardedAsync(
Arg.Is<IReadOnlyList<Guid>>(g => g.ToHashSet().SetEquals(expected)),
Arg.Any<CancellationToken>());
}
[Fact]
public async Task Drain_GrpcThrows_RowsStayPending_NextDrainRetries()
{
// Arrange — first read returns 3 rows; the gRPC client throws on the
// first push, then succeeds on the second. After the second push the
// queue returns empty so the actor settles.
var batch = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(
Task.FromResult<IReadOnlyList<AuditEvent>>(batch),
Task.FromResult<IReadOnlyList<AuditEvent>>(batch),
Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
var calls = 0;
_client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
.Returns(_ =>
{
calls++;
if (calls == 1)
{
throw new InvalidOperationException("simulated gRPC failure");
}
return Task.FromResult(AckAll(batch));
});
// Act
CreateActor();
// Assert — eventually MarkForwardedAsync is called exactly once (after
// the retry succeeded). The first failure must NOT have called
// MarkForwardedAsync because the rows stay Pending.
await AwaitAssertAsync(async () =>
{
await _queue.Received(1).MarkForwardedAsync(
Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
}, TimeSpan.FromSeconds(10));
Assert.True(calls >= 2, $"Expected at least 2 client calls (1 failure + 1 retry); saw {calls}");
}
[Fact]
public async Task Drain_ZeroPending_SchedulesAtIdleInterval_NoClientCall()
{
// Arrange — queue always empty.
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
// Idle interval = 2 s. Pause 3 s after the first tick (1 s busy on
// PreStart) and assert the empty-queue branch did NOT push to the
// client.
CreateActor(Opts(busySeconds: 1, idleSeconds: 2));
// Allow the initial tick (~1 s) + a generous window for the idle re-tick.
await Task.Delay(TimeSpan.FromSeconds(3));
await _client.DidNotReceiveWithAnyArgs().IngestAuditEventsAsync(default!, default);
// ReadPendingAsync was called at least once (initial tick), and at
// most twice within the 3 s window (initial + one idle re-tick).
var readCalls = _queue.ReceivedCalls()
.Count(c => c.GetMethodInfo().Name == nameof(ISiteAuditQueue.ReadPendingAsync));
Assert.InRange(readCalls, 1, 2);
}
[Fact]
public async Task Drain_NonZeroPending_SchedulesAtBusyInterval()
{
// Arrange — every read returns 1 row. With busy=1s the actor should
// re-drain quickly, producing multiple client calls inside a short
// window.
var single = new List<AuditEvent> { NewEvent() };
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(single));
_client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
.Returns(call => Task.FromResult(AckAll(single)));
CreateActor(Opts(busySeconds: 1, idleSeconds: 10));
// 3-second window with busy=1s should fit at least 2 drains.
await Task.Delay(TimeSpan.FromSeconds(3));
var pushCalls = _client.ReceivedCalls()
.Count(c => c.GetMethodInfo().Name == nameof(ISiteStreamAuditClient.IngestAuditEventsAsync));
Assert.True(pushCalls >= 2,
$"Expected ≥2 pushes within 3s when busy=1s; saw {pushCalls}");
}
[Fact]
public async Task Drain_AcceptedEventIdsSubset_OnlyMarksAccepted()
{
// Arrange — 5 rows pushed, but the central ack only lists 3.
var rows = Enumerable.Range(0, 5).Select(_ => NewEvent()).ToList();
var ackedIds = rows.Take(3).Select(r => r.EventId).ToList();
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(
Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
var partialAck = new IngestAck();
foreach (var id in ackedIds)
{
partialAck.AcceptedEventIds.Add(id.ToString());
}
_client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult(partialAck));
// Act
CreateActor();
await AwaitAssertAsync(async () =>
{
await _queue.Received(1).MarkForwardedAsync(
Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
}, TimeSpan.FromSeconds(5));
// Assert — exactly the 3 ack'd ids made it to MarkForwardedAsync, not
// the other 2.
var ackedSet = ackedIds.ToHashSet();
await _queue.Received(1).MarkForwardedAsync(
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3 && g.ToHashSet().SetEquals(ackedSet)),
Arg.Any<CancellationToken>());
}
// ────────────────────────────────────────────────────────────────────────
// AuditLog-001: combined-telemetry cached-drain regression tests. Verify
// that the production wiring of the previously-unreachable cached transport
// routes cached rows through ReadPendingCachedTelemetryAsync +
// IngestCachedTelemetryAsync (and NOT IngestAuditEventsAsync), and that
// orphaned audit rows (no tracking snapshot) are logged + skipped rather
// than crashing the drain.
// ────────────────────────────────────────────────────────────────────────
private static AuditEvent NewCachedEvent(
AuditKind kind = AuditKind.CachedSubmit,
Guid? eventId = null,
Guid? correlationId = null,
string sourceSiteId = "site-1") =>
// C3 (Task 2.5): canonical record via the shared factory; ForwardState is
// no longer a record field (the SQLite shim defaults it on INSERT).
ScadaBridgeAuditEventFactory.Create(
channel: AuditChannel.ApiOutbound,
kind: kind,
status: AuditStatus.Submitted,
eventId: eventId ?? Guid.NewGuid(),
occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
target: "ERP.GetOrder",
sourceSiteId: sourceSiteId,
correlationId: correlationId ?? Guid.NewGuid());
private static TrackingStatusSnapshot NewSnapshot(
TrackedOperationId id,
string status = "Submitted",
int retryCount = 0) => new(
Id: id,
Kind: nameof(AuditKind.ApiCallCached),
TargetSummary: "ERP.GetOrder",
Status: status,
RetryCount: retryCount,
LastError: null,
HttpStatus: null,
CreatedAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
UpdatedAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
TerminalAtUtc: null,
SourceInstanceId: "instance-1",
SourceScript: "script-1",
SourceNode: "node-a");
[Fact]
public async Task CachedDrain_CachedRows_RouteThrough_IngestCachedTelemetry_NotIngestAuditEvents()
{
// Arrange — three cached audit rows on the cached queue, each with a
// matching tracking snapshot. The audit-only queue is empty (those
// rows are excluded by ReadPendingAsync after AuditLog-001).
var cachedRows = new[]
{
NewCachedEvent(AuditKind.CachedSubmit),
NewCachedEvent(AuditKind.ApiCallCached),
NewCachedEvent(AuditKind.CachedResolve),
};
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
_queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(
Task.FromResult<IReadOnlyList<AuditEvent>>(cachedRows),
Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
foreach (var row in cachedRows)
{
var tid = new TrackedOperationId(row.CorrelationId!.Value);
_trackingStore.GetStatusAsync(tid, Arg.Any<CancellationToken>())
.Returns(Task.FromResult<TrackingStatusSnapshot?>(NewSnapshot(tid)));
}
CachedTelemetryBatch? capturedBatch = null;
_client.IngestCachedTelemetryAsync(Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>())
.Returns(call =>
{
capturedBatch = call.Arg<CachedTelemetryBatch>();
var ack = new IngestAck();
foreach (var packet in capturedBatch.Packets)
{
ack.AcceptedEventIds.Add(packet.AuditEvent.EventId);
}
return Task.FromResult(ack);
});
// Act
CreateActorWithCachedDrain();
// Assert — exactly one IngestCachedTelemetryAsync push containing all
// three packets, and zero IngestAuditEventsAsync pushes (the audit-only
// drain saw an empty queue).
await AwaitAssertAsync(async () =>
{
await _client.Received(1).IngestCachedTelemetryAsync(
Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>());
await _queue.Received(1).MarkForwardedAsync(
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3), Arg.Any<CancellationToken>());
}, TimeSpan.FromSeconds(5));
Assert.NotNull(capturedBatch);
Assert.Equal(3, capturedBatch!.Packets.Count);
await _client.DidNotReceiveWithAnyArgs().IngestAuditEventsAsync(default!, default);
var emittedEventIds = capturedBatch.Packets
.Select(p => Guid.Parse(p.AuditEvent.EventId))
.ToHashSet();
var expectedIds = cachedRows.Select(r => r.EventId).ToHashSet();
Assert.Equal(expectedIds, emittedEventIds);
}
[Fact]
public async Task CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash()
{
// Arrange — two cached audit rows: one with a tracking snapshot, one
// orphaned (the tracking store returns null). The orphaned row must be
// skipped without aborting the batch — the valid row still flows.
var orphan = NewCachedEvent(AuditKind.CachedSubmit);
var valid = NewCachedEvent(AuditKind.CachedResolve);
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
_queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(
Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { orphan, valid }),
Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
// orphan: tracking returns null
_trackingStore.GetStatusAsync(
new TrackedOperationId(orphan.CorrelationId!.Value),
Arg.Any<CancellationToken>())
.Returns(Task.FromResult<TrackingStatusSnapshot?>(null));
// valid: tracking returns a snapshot
var validTid = new TrackedOperationId(valid.CorrelationId!.Value);
_trackingStore.GetStatusAsync(validTid, Arg.Any<CancellationToken>())
.Returns(Task.FromResult<TrackingStatusSnapshot?>(NewSnapshot(validTid, "Delivered")));
CachedTelemetryBatch? capturedBatch = null;
_client.IngestCachedTelemetryAsync(Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>())
.Returns(call =>
{
capturedBatch = call.Arg<CachedTelemetryBatch>();
var ack = new IngestAck();
foreach (var packet in capturedBatch.Packets)
{
ack.AcceptedEventIds.Add(packet.AuditEvent.EventId);
}
return Task.FromResult(ack);
});
// Act
CreateActorWithCachedDrain();
// Assert — exactly one push containing ONLY the valid row; the orphan
// is skipped and stays Pending (not in MarkForwardedAsync's id list).
await AwaitAssertAsync(async () =>
{
await _client.Received(1).IngestCachedTelemetryAsync(
Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>());
}, TimeSpan.FromSeconds(5));
Assert.NotNull(capturedBatch);
Assert.Single(capturedBatch!.Packets);
Assert.Equal(valid.EventId.ToString(), capturedBatch.Packets[0].AuditEvent.EventId);
await _queue.Received(1).MarkForwardedAsync(
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == valid.EventId),
Arg.Any<CancellationToken>());
}
[Fact]
public async Task AuditOnlyDrain_StillFlows_When_CachedDrain_IsDisabled()
{
// Arrange — ordinary (non-cached) audit rows on the audit-only queue;
// the actor is constructed WITHOUT a tracking store so the cached
// scheduler is never armed. Regression guard against the audit-only
// drain regressing during the AuditLog-001 refactor.
var rows = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
.Returns(
Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
_client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
.Returns(_ => Task.FromResult(AckAll(rows)));
// Act — note: CreateActor (no tracking store), not CreateActorWithCachedDrain.
CreateActor();
// Assert — audit-only drain flows normally; the cached client is
// never called and ReadPendingCachedTelemetryAsync is never queried.
await AwaitAssertAsync(async () =>
{
await _client.Received(1).IngestAuditEventsAsync(
Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>());
await _queue.Received(1).MarkForwardedAsync(
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3), Arg.Any<CancellationToken>());
}, TimeSpan.FromSeconds(5));
await _client.DidNotReceiveWithAnyArgs().IngestCachedTelemetryAsync(default!, default);
await _queue.DidNotReceiveWithAnyArgs().ReadPendingCachedTelemetryAsync(default, default);
}
}