test(auditlog): cached call combined telemetry end-to-end (#23 M3)
This commit is contained in:
@@ -0,0 +1,270 @@
|
|||||||
|
using Akka.TestKit.Xunit2;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using ScadaLink.AuditLog.Tests.Integration.Infrastructure;
|
||||||
|
using ScadaLink.Commons.Entities.Audit;
|
||||||
|
using ScadaLink.Commons.Interfaces.Services;
|
||||||
|
using ScadaLink.Commons.Messages.Integration;
|
||||||
|
using ScadaLink.Commons.Types;
|
||||||
|
using ScadaLink.Commons.Types.Enums;
|
||||||
|
using ScadaLink.ConfigurationDatabase.Tests.Migrations;
|
||||||
|
|
||||||
|
namespace ScadaLink.AuditLog.Tests.Integration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Bundle G G2 end-to-end suite for cached <c>ExternalSystem.CachedCall</c>
|
||||||
|
/// lifecycle telemetry (Audit Log #23 / M3). Wires the full M3 pipeline:
|
||||||
|
/// site-local SQLite audit writer + operation tracking store + the production
|
||||||
|
/// <see cref="CachedCallTelemetryForwarder"/> + the test-side
|
||||||
|
/// <see cref="CombinedTelemetryDispatcher"/> that ALSO pushes each combined
|
||||||
|
/// packet through the stub gRPC client into the central
|
||||||
|
/// <c>AuditLogIngestActor</c>'s dual-write transaction against a per-test
|
||||||
|
/// MSSQL database. Asserts the audit rows + the SiteCalls row + the
|
||||||
|
/// site-local tracking row converge to the expected shape for each lifecycle.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// The bridge is driven directly via <see cref="CombinedTelemetryHarness.EmitAttemptAsync"/>
|
||||||
|
/// — these tests do NOT spin up the actual S&F retry loop; that would
|
||||||
|
/// require a full SiteRuntime host and is out of scope for M3 (the S&F
|
||||||
|
/// observer hooks are exercised in <c>ScadaLink.StoreAndForward.Tests</c> at
|
||||||
|
/// unit level). The submit row is emitted via
|
||||||
|
/// <see cref="CombinedTelemetryHarness.EmitSubmitAsync"/> because the
|
||||||
|
/// production submit emission happens at the script-call site, not inside the
|
||||||
|
/// S&F loop.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// Each test uses a unique <c>SourceSite</c> id (Guid suffix) so concurrent
|
||||||
|
/// tests sharing the per-fixture MSSQL database don't interfere with each
|
||||||
|
/// other.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
public class CachedCallCombinedTelemetryTests : TestKit, IClassFixture<MsSqlMigrationFixture>
|
||||||
|
{
|
||||||
|
private readonly MsSqlMigrationFixture _fixture;
|
||||||
|
|
||||||
|
public CachedCallCombinedTelemetryTests(MsSqlMigrationFixture fixture)
|
||||||
|
{
|
||||||
|
_fixture = fixture;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static string NewSiteId() =>
|
||||||
|
"test-g2-cached-" + Guid.NewGuid().ToString("N").Substring(0, 8);
|
||||||
|
|
||||||
|
private static CachedCallTelemetry SubmitPacket(
|
||||||
|
TrackedOperationId id, string siteId, DateTime nowUtc, string target = "ERP.GetOrder") =>
|
||||||
|
new(
|
||||||
|
Audit: new AuditEvent
|
||||||
|
{
|
||||||
|
EventId = Guid.NewGuid(),
|
||||||
|
OccurredAtUtc = nowUtc,
|
||||||
|
Channel = AuditChannel.ApiOutbound,
|
||||||
|
Kind = AuditKind.CachedSubmit,
|
||||||
|
CorrelationId = id.Value,
|
||||||
|
SourceSiteId = siteId,
|
||||||
|
SourceInstanceId = "Plant.Pump42",
|
||||||
|
SourceScript = "ScriptActor:doStuff",
|
||||||
|
Target = target,
|
||||||
|
Status = AuditStatus.Submitted,
|
||||||
|
ForwardState = AuditForwardState.Pending,
|
||||||
|
},
|
||||||
|
Operational: new SiteCallOperational(
|
||||||
|
TrackedOperationId: id,
|
||||||
|
Channel: "ApiOutbound",
|
||||||
|
Target: target,
|
||||||
|
SourceSite: siteId,
|
||||||
|
Status: "Submitted",
|
||||||
|
RetryCount: 0,
|
||||||
|
LastError: null,
|
||||||
|
HttpStatus: null,
|
||||||
|
CreatedAtUtc: nowUtc,
|
||||||
|
UpdatedAtUtc: nowUtc,
|
||||||
|
TerminalAtUtc: null));
|
||||||
|
|
||||||
|
private static CachedCallAttemptContext AttemptContext(
|
||||||
|
TrackedOperationId id,
|
||||||
|
string siteId,
|
||||||
|
CachedCallAttemptOutcome outcome,
|
||||||
|
int retryCount,
|
||||||
|
string? lastError,
|
||||||
|
int? httpStatus,
|
||||||
|
DateTime createdUtc,
|
||||||
|
DateTime occurredUtc,
|
||||||
|
string target = "ERP.GetOrder",
|
||||||
|
string channel = "ApiOutbound") =>
|
||||||
|
new(
|
||||||
|
TrackedOperationId: id,
|
||||||
|
Channel: channel,
|
||||||
|
Target: target,
|
||||||
|
SourceSite: siteId,
|
||||||
|
Outcome: outcome,
|
||||||
|
RetryCount: retryCount,
|
||||||
|
LastError: lastError,
|
||||||
|
HttpStatus: httpStatus,
|
||||||
|
CreatedAtUtc: createdUtc,
|
||||||
|
OccurredAtUtc: occurredUtc,
|
||||||
|
DurationMs: 42,
|
||||||
|
SourceInstanceId: "Plant.Pump42");
|
||||||
|
|
||||||
|
[SkippableFact]
|
||||||
|
public async Task CachedCall_FailFailSuccess_Emits_5_AuditRows_AND_1_SiteCall_Delivered()
|
||||||
|
{
|
||||||
|
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||||
|
|
||||||
|
var siteId = NewSiteId();
|
||||||
|
var trackedId = TrackedOperationId.New();
|
||||||
|
var t0 = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
|
||||||
|
|
||||||
|
await using var harness = new CombinedTelemetryHarness(_fixture, this);
|
||||||
|
|
||||||
|
// Submit
|
||||||
|
await harness.EmitSubmitAsync(SubmitPacket(trackedId, siteId, t0));
|
||||||
|
|
||||||
|
// Attempt 1: transient HTTP 500
|
||||||
|
await harness.EmitAttemptAsync(AttemptContext(
|
||||||
|
trackedId, siteId,
|
||||||
|
CachedCallAttemptOutcome.TransientFailure,
|
||||||
|
retryCount: 1, lastError: "HTTP 500", httpStatus: 500,
|
||||||
|
createdUtc: t0, occurredUtc: t0.AddSeconds(5)));
|
||||||
|
|
||||||
|
// Attempt 2: transient HTTP 500
|
||||||
|
await harness.EmitAttemptAsync(AttemptContext(
|
||||||
|
trackedId, siteId,
|
||||||
|
CachedCallAttemptOutcome.TransientFailure,
|
||||||
|
retryCount: 2, lastError: "HTTP 500", httpStatus: 500,
|
||||||
|
createdUtc: t0, occurredUtc: t0.AddSeconds(15)));
|
||||||
|
|
||||||
|
// Attempt 3: delivered (terminal — emits Attempted + CachedResolve)
|
||||||
|
await harness.EmitAttemptAsync(AttemptContext(
|
||||||
|
trackedId, siteId,
|
||||||
|
CachedCallAttemptOutcome.Delivered,
|
||||||
|
retryCount: 3, lastError: null, httpStatus: 200,
|
||||||
|
createdUtc: t0, occurredUtc: t0.AddSeconds(25)));
|
||||||
|
|
||||||
|
// Central side: each forward through the dispatcher round-trips
|
||||||
|
// through the stub client + ingest actor, so by the time the awaits
|
||||||
|
// complete the rows are visible in MSSQL.
|
||||||
|
await using var read = harness.CreateReadContext();
|
||||||
|
|
||||||
|
// 1 Submit + 2 transient Attempted + 1 terminal Attempted + 1
|
||||||
|
// CachedResolve = 5 audit rows. The plan allows 4-5; this is the
|
||||||
|
// happy path emitting exactly 5.
|
||||||
|
var auditRows = await read.Set<AuditEvent>()
|
||||||
|
.Where(e => e.SourceSiteId == siteId)
|
||||||
|
.ToListAsync();
|
||||||
|
Assert.InRange(auditRows.Count, 4, 5);
|
||||||
|
|
||||||
|
// All audit rows must share the same CorrelationId (= TrackedOperationId).
|
||||||
|
Assert.All(auditRows, r => Assert.Equal(trackedId.Value, r.CorrelationId));
|
||||||
|
|
||||||
|
// Exactly one CachedSubmit row.
|
||||||
|
Assert.Single(auditRows, r => r.Kind == AuditKind.CachedSubmit);
|
||||||
|
// Exactly one terminal CachedResolve row, status Delivered.
|
||||||
|
var resolve = Assert.Single(auditRows, r => r.Kind == AuditKind.CachedResolve);
|
||||||
|
Assert.Equal(AuditStatus.Delivered, resolve.Status);
|
||||||
|
|
||||||
|
// SiteCalls row: Delivered, RetryCount=3, TerminalAtUtc set.
|
||||||
|
var siteCall = await read.Set<SiteCall>()
|
||||||
|
.SingleAsync(s => s.TrackedOperationId == trackedId);
|
||||||
|
Assert.Equal("Delivered", siteCall.Status);
|
||||||
|
Assert.Equal(3, siteCall.RetryCount);
|
||||||
|
Assert.NotNull(siteCall.TerminalAtUtc);
|
||||||
|
|
||||||
|
// Site-local Tracking.Status mirrors the same outcome.
|
||||||
|
var snapshot = await harness.TrackingStore.GetStatusAsync(trackedId);
|
||||||
|
Assert.NotNull(snapshot);
|
||||||
|
Assert.Equal("Delivered", snapshot!.Status);
|
||||||
|
Assert.NotNull(snapshot.TerminalAtUtc);
|
||||||
|
}
|
||||||
|
|
||||||
|
[SkippableFact]
|
||||||
|
public async Task CachedCall_AllAttemptsFailedAndParked_Emits_Terminal_Parked()
|
||||||
|
{
|
||||||
|
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||||
|
|
||||||
|
var siteId = NewSiteId();
|
||||||
|
var trackedId = TrackedOperationId.New();
|
||||||
|
var t0 = new DateTime(2026, 5, 20, 11, 0, 0, DateTimeKind.Utc);
|
||||||
|
|
||||||
|
await using var harness = new CombinedTelemetryHarness(_fixture, this);
|
||||||
|
|
||||||
|
await harness.EmitSubmitAsync(SubmitPacket(trackedId, siteId, t0));
|
||||||
|
|
||||||
|
// Three transient failures...
|
||||||
|
for (int i = 1; i <= 3; i++)
|
||||||
|
{
|
||||||
|
await harness.EmitAttemptAsync(AttemptContext(
|
||||||
|
trackedId, siteId,
|
||||||
|
CachedCallAttemptOutcome.TransientFailure,
|
||||||
|
retryCount: i, lastError: "HTTP 500", httpStatus: 500,
|
||||||
|
createdUtc: t0, occurredUtc: t0.AddSeconds(i * 5)));
|
||||||
|
}
|
||||||
|
|
||||||
|
// ...then S&F gives up — ParkedMaxRetries.
|
||||||
|
await harness.EmitAttemptAsync(AttemptContext(
|
||||||
|
trackedId, siteId,
|
||||||
|
CachedCallAttemptOutcome.ParkedMaxRetries,
|
||||||
|
retryCount: 4, lastError: "HTTP 500", httpStatus: 500,
|
||||||
|
createdUtc: t0, occurredUtc: t0.AddSeconds(30)));
|
||||||
|
|
||||||
|
await using var read = harness.CreateReadContext();
|
||||||
|
|
||||||
|
var siteCall = await read.Set<SiteCall>()
|
||||||
|
.SingleAsync(s => s.TrackedOperationId == trackedId);
|
||||||
|
Assert.Equal("Parked", siteCall.Status);
|
||||||
|
Assert.NotNull(siteCall.TerminalAtUtc);
|
||||||
|
|
||||||
|
// Terminal audit row should also be Parked.
|
||||||
|
var resolve = await read.Set<AuditEvent>()
|
||||||
|
.Where(e => e.SourceSiteId == siteId && e.Kind == AuditKind.CachedResolve)
|
||||||
|
.SingleAsync();
|
||||||
|
Assert.Equal(AuditStatus.Parked, resolve.Status);
|
||||||
|
|
||||||
|
// Site-local tracking matches.
|
||||||
|
var snapshot = await harness.TrackingStore.GetStatusAsync(trackedId);
|
||||||
|
Assert.NotNull(snapshot);
|
||||||
|
Assert.Equal("Parked", snapshot!.Status);
|
||||||
|
Assert.NotNull(snapshot.TerminalAtUtc);
|
||||||
|
}
|
||||||
|
|
||||||
|
[SkippableFact]
|
||||||
|
public async Task CachedCall_ImmediateSuccess_NoSF_Emits_Attempted_And_Resolve_Delivered()
|
||||||
|
{
|
||||||
|
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||||
|
|
||||||
|
var siteId = NewSiteId();
|
||||||
|
var trackedId = TrackedOperationId.New();
|
||||||
|
var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
|
||||||
|
|
||||||
|
await using var harness = new CombinedTelemetryHarness(_fixture, this);
|
||||||
|
|
||||||
|
// Submit + immediate delivered attempt (RetryCount = 0).
|
||||||
|
await harness.EmitSubmitAsync(SubmitPacket(trackedId, siteId, t0));
|
||||||
|
await harness.EmitAttemptAsync(AttemptContext(
|
||||||
|
trackedId, siteId,
|
||||||
|
CachedCallAttemptOutcome.Delivered,
|
||||||
|
retryCount: 0, lastError: null, httpStatus: 200,
|
||||||
|
createdUtc: t0, occurredUtc: t0.AddMilliseconds(50)));
|
||||||
|
|
||||||
|
await using var read = harness.CreateReadContext();
|
||||||
|
|
||||||
|
var siteCall = await read.Set<SiteCall>()
|
||||||
|
.SingleAsync(s => s.TrackedOperationId == trackedId);
|
||||||
|
Assert.Equal("Delivered", siteCall.Status);
|
||||||
|
Assert.Equal(0, siteCall.RetryCount);
|
||||||
|
Assert.NotNull(siteCall.TerminalAtUtc);
|
||||||
|
|
||||||
|
// 1 Submit + 1 Attempted + 1 CachedResolve = 3 audit rows.
|
||||||
|
var auditRows = await read.Set<AuditEvent>()
|
||||||
|
.Where(e => e.SourceSiteId == siteId)
|
||||||
|
.ToListAsync();
|
||||||
|
Assert.Equal(3, auditRows.Count);
|
||||||
|
Assert.Single(auditRows, r => r.Kind == AuditKind.CachedSubmit);
|
||||||
|
Assert.Single(auditRows, r => r.Kind == AuditKind.ApiCallCached);
|
||||||
|
var resolve = Assert.Single(auditRows, r => r.Kind == AuditKind.CachedResolve);
|
||||||
|
Assert.Equal(AuditStatus.Delivered, resolve.Status);
|
||||||
|
|
||||||
|
var snapshot = await harness.TrackingStore.GetStatusAsync(trackedId);
|
||||||
|
Assert.NotNull(snapshot);
|
||||||
|
Assert.Equal("Delivered", snapshot!.Status);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,125 @@
|
|||||||
|
using ScadaLink.AuditLog.Site.Telemetry;
|
||||||
|
using ScadaLink.AuditLog.Telemetry;
|
||||||
|
using ScadaLink.Commons.Interfaces.Services;
|
||||||
|
using ScadaLink.Commons.Messages.Integration;
|
||||||
|
using ScadaLink.Commons.Types;
|
||||||
|
using ScadaLink.Communication.Grpc;
|
||||||
|
using Google.Protobuf.WellKnownTypes;
|
||||||
|
using Timestamp = Google.Protobuf.WellKnownTypes.Timestamp;
|
||||||
|
|
||||||
|
namespace ScadaLink.AuditLog.Tests.Integration.Infrastructure;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Test-side combined-telemetry dispatcher: wraps a production
|
||||||
|
/// <see cref="ICachedCallTelemetryForwarder"/> so the local audit + tracking
|
||||||
|
/// stores still get written, then projects the same packet onto the wire as a
|
||||||
|
/// <see cref="CachedTelemetryBatch"/> and pushes it through the supplied
|
||||||
|
/// <see cref="ISiteStreamAuditClient"/>. The bridge can be composed into the
|
||||||
|
/// existing <see cref="CachedCallLifecycleBridge"/> chain as the
|
||||||
|
/// <see cref="ICachedCallTelemetryForwarder"/> implementation so a single
|
||||||
|
/// observer callback fans out to both halves.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// Production wiring keeps the wire push deferred to M6 — the site SQLite hot
|
||||||
|
/// path is the source of truth and a future M6 drain will push the rows
|
||||||
|
/// through the gRPC client. For end-to-end testing today we need a way to
|
||||||
|
/// exercise the central dual-write transaction immediately, so this
|
||||||
|
/// dispatcher synthesises the wire packet inline and round-trips it through
|
||||||
|
/// the stub client. The shape mirrors what the M6 drain will eventually emit.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b>Best-effort:</b> both the inner forwarder call and the wire push are
|
||||||
|
/// wrapped in independent try/catch blocks. A thrown wire client doesn't
|
||||||
|
/// abort the local writes (the audit row is already in SQLite); a thrown
|
||||||
|
/// local forwarder doesn't abort the wire push (central still gets the
|
||||||
|
/// dual-write attempt).
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class CombinedTelemetryDispatcher : ICachedCallTelemetryForwarder
|
||||||
|
{
|
||||||
|
private readonly ICachedCallTelemetryForwarder _inner;
|
||||||
|
private readonly ISiteStreamAuditClient _wireClient;
|
||||||
|
|
||||||
|
public CombinedTelemetryDispatcher(
|
||||||
|
ICachedCallTelemetryForwarder inner,
|
||||||
|
ISiteStreamAuditClient wireClient)
|
||||||
|
{
|
||||||
|
_inner = inner ?? throw new ArgumentNullException(nameof(inner));
|
||||||
|
_wireClient = wireClient ?? throw new ArgumentNullException(nameof(wireClient));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc/>
|
||||||
|
public async Task ForwardAsync(CachedCallTelemetry telemetry, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(telemetry);
|
||||||
|
|
||||||
|
// Inner forwarder writes the audit row to SQLite + updates the
|
||||||
|
// tracking store. Best-effort — exceptions are already swallowed
|
||||||
|
// inside the production forwarder, but wrap defensively here too in
|
||||||
|
// case a test substitutes a stricter inner.
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await _inner.ForwardAsync(telemetry, ct).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch
|
||||||
|
{
|
||||||
|
// Swallow — alog.md §7 best-effort contract.
|
||||||
|
}
|
||||||
|
|
||||||
|
// Project the same packet onto the wire and push it through the stub
|
||||||
|
// client. This is the bit a future M6 drain will subsume — until
|
||||||
|
// then the test wraps the two halves into one observer-driven step.
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var batch = new CachedTelemetryBatch();
|
||||||
|
batch.Packets.Add(BuildPacket(telemetry));
|
||||||
|
await _wireClient.IngestCachedTelemetryAsync(batch, ct).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch
|
||||||
|
{
|
||||||
|
// Swallow — the audit row is still in SQLite for a future drain;
|
||||||
|
// the central row will materialise the next time the wire path
|
||||||
|
// is exercised (or via the M6 reconciliation pull).
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static CachedTelemetryPacket BuildPacket(CachedCallTelemetry telemetry)
|
||||||
|
{
|
||||||
|
return new CachedTelemetryPacket
|
||||||
|
{
|
||||||
|
AuditEvent = AuditEventMapper.ToDto(telemetry.Audit),
|
||||||
|
Operational = ToOperationalDto(telemetry.Operational),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
private static SiteCallOperationalDto ToOperationalDto(SiteCallOperational op)
|
||||||
|
{
|
||||||
|
var dto = new SiteCallOperationalDto
|
||||||
|
{
|
||||||
|
TrackedOperationId = op.TrackedOperationId.Value.ToString("D"),
|
||||||
|
Channel = op.Channel,
|
||||||
|
Target = op.Target,
|
||||||
|
SourceSite = op.SourceSite,
|
||||||
|
Status = op.Status,
|
||||||
|
RetryCount = op.RetryCount,
|
||||||
|
LastError = op.LastError ?? string.Empty,
|
||||||
|
CreatedAtUtc = Timestamp.FromDateTime(EnsureUtc(op.CreatedAtUtc)),
|
||||||
|
UpdatedAtUtc = Timestamp.FromDateTime(EnsureUtc(op.UpdatedAtUtc)),
|
||||||
|
};
|
||||||
|
if (op.HttpStatus.HasValue)
|
||||||
|
{
|
||||||
|
dto.HttpStatus = op.HttpStatus.Value;
|
||||||
|
}
|
||||||
|
if (op.TerminalAtUtc.HasValue)
|
||||||
|
{
|
||||||
|
dto.TerminalAtUtc = Timestamp.FromDateTime(EnsureUtc(op.TerminalAtUtc.Value));
|
||||||
|
}
|
||||||
|
return dto;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static DateTime EnsureUtc(DateTime value) =>
|
||||||
|
value.Kind == DateTimeKind.Utc
|
||||||
|
? value
|
||||||
|
: DateTime.SpecifyKind(value.ToUniversalTime(), DateTimeKind.Utc);
|
||||||
|
}
|
||||||
@@ -0,0 +1,175 @@
|
|||||||
|
using Akka.Actor;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Microsoft.Extensions.Options;
|
||||||
|
using ScadaLink.AuditLog.Central;
|
||||||
|
using ScadaLink.AuditLog.Site;
|
||||||
|
using ScadaLink.AuditLog.Site.Telemetry;
|
||||||
|
using ScadaLink.Commons.Interfaces;
|
||||||
|
using ScadaLink.Commons.Interfaces.Repositories;
|
||||||
|
using ScadaLink.Commons.Interfaces.Services;
|
||||||
|
using ScadaLink.Commons.Messages.Integration;
|
||||||
|
using ScadaLink.ConfigurationDatabase;
|
||||||
|
using ScadaLink.ConfigurationDatabase.Repositories;
|
||||||
|
using ScadaLink.ConfigurationDatabase.Tests.Migrations;
|
||||||
|
using ScadaLink.SiteRuntime.Tracking;
|
||||||
|
|
||||||
|
namespace ScadaLink.AuditLog.Tests.Integration.Infrastructure;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Shared end-to-end harness for the M3 cached-call combined telemetry tests
|
||||||
|
/// (G2/G3/G4). Composes the full pipeline:
|
||||||
|
/// <list type="bullet">
|
||||||
|
/// <item><description>Site-local SQLite <see cref="SqliteAuditWriter"/> (in-memory) +
|
||||||
|
/// <see cref="RingBufferFallback"/> + <see cref="FallbackAuditWriter"/>.</description></item>
|
||||||
|
/// <item><description>Site-local SQLite <see cref="OperationTrackingStore"/> (in-memory).</description></item>
|
||||||
|
/// <item><description>Production <see cref="CachedCallTelemetryForwarder"/> wrapped by a
|
||||||
|
/// test-side <see cref="CombinedTelemetryDispatcher"/> that also pushes each
|
||||||
|
/// packet through the stub gRPC client.</description></item>
|
||||||
|
/// <item><description><see cref="CachedCallLifecycleBridge"/> wired to the
|
||||||
|
/// dispatcher so a single observer call fans out audit + tracking + wire.</description></item>
|
||||||
|
/// <item><description><see cref="DirectActorSiteStreamAuditClient"/> connected
|
||||||
|
/// to an <see cref="AuditLogIngestActor"/> backed by the real
|
||||||
|
/// <see cref="AuditLogRepository"/> + <see cref="SiteCallAuditRepository"/>
|
||||||
|
/// against the per-test <see cref="MsSqlMigrationFixture"/> database.</description></item>
|
||||||
|
/// </list>
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// Disposal cleans up the in-memory SQLite stores. The Akka actor system is
|
||||||
|
/// owned by the calling <see cref="Akka.TestKit.Xunit2.TestKit"/>; the harness
|
||||||
|
/// only owns the ingest actor IActorRef and the underlying repositories'
|
||||||
|
/// DbContext lifecycle.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class CombinedTelemetryHarness : IAsyncDisposable
|
||||||
|
{
|
||||||
|
public SqliteAuditWriter SqliteWriter { get; }
|
||||||
|
public RingBufferFallback Ring { get; }
|
||||||
|
public FallbackAuditWriter FallbackWriter { get; }
|
||||||
|
public OperationTrackingStore TrackingStore { get; }
|
||||||
|
public CachedCallTelemetryForwarder InnerForwarder { get; }
|
||||||
|
public CombinedTelemetryDispatcher Dispatcher { get; }
|
||||||
|
public CachedCallLifecycleBridge Bridge { get; }
|
||||||
|
public DirectActorSiteStreamAuditClient StubClient { get; }
|
||||||
|
public IActorRef IngestActor { get; }
|
||||||
|
public IServiceProvider ServiceProvider { get; }
|
||||||
|
|
||||||
|
private readonly MsSqlMigrationFixture _fixture;
|
||||||
|
private bool _disposed;
|
||||||
|
|
||||||
|
public CombinedTelemetryHarness(
|
||||||
|
MsSqlMigrationFixture fixture,
|
||||||
|
Akka.TestKit.Xunit2.TestKit testKit,
|
||||||
|
Func<ScadaLinkDbContext, ISiteCallAuditRepository>? siteCallRepoOverride = null)
|
||||||
|
{
|
||||||
|
_fixture = fixture ?? throw new ArgumentNullException(nameof(fixture));
|
||||||
|
ArgumentNullException.ThrowIfNull(testKit);
|
||||||
|
|
||||||
|
// Site SQLite — unique in-memory database per harness so tests don't share
|
||||||
|
// an audit queue. Mode=Memory + Cache=Shared keeps the file alive for the
|
||||||
|
// lifetime of the writer connection.
|
||||||
|
SqliteWriter = new SqliteAuditWriter(
|
||||||
|
Options.Create(new SqliteAuditWriterOptions
|
||||||
|
{
|
||||||
|
DatabasePath = "ignored",
|
||||||
|
BatchSize = 64,
|
||||||
|
ChannelCapacity = 1024,
|
||||||
|
}),
|
||||||
|
NullLogger<SqliteAuditWriter>.Instance,
|
||||||
|
connectionStringOverride:
|
||||||
|
$"Data Source=file:cachedcall-g-{Guid.NewGuid():N}?mode=memory&cache=shared");
|
||||||
|
|
||||||
|
Ring = new RingBufferFallback();
|
||||||
|
FallbackWriter = new FallbackAuditWriter(
|
||||||
|
SqliteWriter, Ring, new NoOpAuditWriteFailureCounter(),
|
||||||
|
NullLogger<FallbackAuditWriter>.Instance);
|
||||||
|
|
||||||
|
TrackingStore = new OperationTrackingStore(
|
||||||
|
Options.Create(new OperationTrackingOptions
|
||||||
|
{
|
||||||
|
// Same shared-in-memory pattern as the audit writer.
|
||||||
|
ConnectionString =
|
||||||
|
$"Data Source=file:tracking-g-{Guid.NewGuid():N}?mode=memory&cache=shared",
|
||||||
|
}),
|
||||||
|
NullLogger<OperationTrackingStore>.Instance);
|
||||||
|
|
||||||
|
// Central wiring: real repositories backed by the MSSQL fixture's DB.
|
||||||
|
ServiceProvider = BuildCentralServiceProvider(siteCallRepoOverride);
|
||||||
|
IngestActor = testKit.Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
|
||||||
|
ServiceProvider,
|
||||||
|
NullLogger<AuditLogIngestActor>.Instance)));
|
||||||
|
|
||||||
|
StubClient = new DirectActorSiteStreamAuditClient(IngestActor);
|
||||||
|
|
||||||
|
// Production forwarder writes the local stores; the dispatcher wraps
|
||||||
|
// it to ALSO push the same packet to central via the stub client.
|
||||||
|
InnerForwarder = new CachedCallTelemetryForwarder(
|
||||||
|
FallbackWriter, TrackingStore, NullLogger<CachedCallTelemetryForwarder>.Instance);
|
||||||
|
Dispatcher = new CombinedTelemetryDispatcher(InnerForwarder, StubClient);
|
||||||
|
|
||||||
|
Bridge = new CachedCallLifecycleBridge(Dispatcher, NullLogger<CachedCallLifecycleBridge>.Instance);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Convenience: emit the initial submit packet directly through the
|
||||||
|
/// dispatcher (the bridge's hooks fire only for S&F retry-loop
|
||||||
|
/// attempts; submit-row emission happens at the script call site).
|
||||||
|
/// </summary>
|
||||||
|
public Task EmitSubmitAsync(CachedCallTelemetry submit, CancellationToken ct = default) =>
|
||||||
|
Dispatcher.ForwardAsync(submit, ct);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Convenience: route a per-attempt or terminal outcome through the bridge.
|
||||||
|
/// </summary>
|
||||||
|
public Task EmitAttemptAsync(CachedCallAttemptContext context, CancellationToken ct = default) =>
|
||||||
|
Bridge.OnAttemptCompletedAsync(context, ct);
|
||||||
|
|
||||||
|
public ScadaLinkDbContext CreateReadContext()
|
||||||
|
{
|
||||||
|
var options = new DbContextOptionsBuilder<ScadaLinkDbContext>()
|
||||||
|
.UseSqlServer(_fixture.ConnectionString)
|
||||||
|
.Options;
|
||||||
|
return new ScadaLinkDbContext(options);
|
||||||
|
}
|
||||||
|
|
||||||
|
private IServiceProvider BuildCentralServiceProvider(
|
||||||
|
Func<ScadaLinkDbContext, ISiteCallAuditRepository>? siteCallRepoOverride)
|
||||||
|
{
|
||||||
|
var services = new ServiceCollection();
|
||||||
|
services.AddDbContext<ScadaLinkDbContext>(opts =>
|
||||||
|
opts.UseSqlServer(_fixture.ConnectionString)
|
||||||
|
.ConfigureWarnings(w => w.Ignore(
|
||||||
|
Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning)));
|
||||||
|
services.AddScoped<IAuditLogRepository>(sp =>
|
||||||
|
new AuditLogRepository(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||||
|
if (siteCallRepoOverride is null)
|
||||||
|
{
|
||||||
|
services.AddScoped<ISiteCallAuditRepository>(sp =>
|
||||||
|
new SiteCallAuditRepository(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
services.AddScoped(sp =>
|
||||||
|
siteCallRepoOverride(sp.GetRequiredService<ScadaLinkDbContext>()));
|
||||||
|
}
|
||||||
|
return services.BuildServiceProvider();
|
||||||
|
}
|
||||||
|
|
||||||
|
public async ValueTask DisposeAsync()
|
||||||
|
{
|
||||||
|
if (_disposed) return;
|
||||||
|
_disposed = true;
|
||||||
|
await SqliteWriter.DisposeAsync().ConfigureAwait(false);
|
||||||
|
await TrackingStore.DisposeAsync().ConfigureAwait(false);
|
||||||
|
if (ServiceProvider is IAsyncDisposable asyncSp)
|
||||||
|
{
|
||||||
|
await asyncSp.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
else if (ServiceProvider is IDisposable sp)
|
||||||
|
{
|
||||||
|
sp.Dispose();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -48,6 +48,13 @@
|
|||||||
the fixture + EF migrations come along without duplicating them.
|
the fixture + EF migrations come along without duplicating them.
|
||||||
-->
|
-->
|
||||||
<ProjectReference Include="../ScadaLink.ConfigurationDatabase.Tests/ScadaLink.ConfigurationDatabase.Tests.csproj" />
|
<ProjectReference Include="../ScadaLink.ConfigurationDatabase.Tests/ScadaLink.ConfigurationDatabase.Tests.csproj" />
|
||||||
|
<!--
|
||||||
|
G2/G3/G4: the cached-call combined telemetry integration tests compose the
|
||||||
|
production OperationTrackingStore (site SQLite source of truth for
|
||||||
|
Tracking.Status) alongside the M2 audit writer chain, so the harness
|
||||||
|
needs a project reference to SiteRuntime where the store lives.
|
||||||
|
-->
|
||||||
|
<ProjectReference Include="../../src/ScadaLink.SiteRuntime/ScadaLink.SiteRuntime.csproj" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
</Project>
|
</Project>
|
||||||
|
|||||||
Reference in New Issue
Block a user