diff --git a/tests/ScadaLink.AuditLog.Tests/Integration/CachedCallCombinedTelemetryTests.cs b/tests/ScadaLink.AuditLog.Tests/Integration/CachedCallCombinedTelemetryTests.cs new file mode 100644 index 0000000..342b028 --- /dev/null +++ b/tests/ScadaLink.AuditLog.Tests/Integration/CachedCallCombinedTelemetryTests.cs @@ -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; + +/// +/// Bundle G G2 end-to-end suite for cached ExternalSystem.CachedCall +/// lifecycle telemetry (Audit Log #23 / M3). Wires the full M3 pipeline: +/// site-local SQLite audit writer + operation tracking store + the production +/// + the test-side +/// that ALSO pushes each combined +/// packet through the stub gRPC client into the central +/// AuditLogIngestActor'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. +/// +/// +/// +/// The bridge is driven directly via +/// — 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 ScadaLink.StoreAndForward.Tests at +/// unit level). The submit row is emitted via +/// because the +/// production submit emission happens at the script-call site, not inside the +/// S&F loop. +/// +/// +/// Each test uses a unique SourceSite id (Guid suffix) so concurrent +/// tests sharing the per-fixture MSSQL database don't interfere with each +/// other. +/// +/// +public class CachedCallCombinedTelemetryTests : TestKit, IClassFixture +{ + 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() + .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() + .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() + .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() + .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() + .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() + .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); + } +} diff --git a/tests/ScadaLink.AuditLog.Tests/Integration/Infrastructure/CombinedTelemetryDispatcher.cs b/tests/ScadaLink.AuditLog.Tests/Integration/Infrastructure/CombinedTelemetryDispatcher.cs new file mode 100644 index 0000000..83c8303 --- /dev/null +++ b/tests/ScadaLink.AuditLog.Tests/Integration/Infrastructure/CombinedTelemetryDispatcher.cs @@ -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; + +/// +/// Test-side combined-telemetry dispatcher: wraps a production +/// so the local audit + tracking +/// stores still get written, then projects the same packet onto the wire as a +/// and pushes it through the supplied +/// . The bridge can be composed into the +/// existing chain as the +/// implementation so a single +/// observer callback fans out to both halves. +/// +/// +/// +/// 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. +/// +/// +/// Best-effort: 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). +/// +/// +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)); + } + + /// + 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); +} diff --git a/tests/ScadaLink.AuditLog.Tests/Integration/Infrastructure/CombinedTelemetryHarness.cs b/tests/ScadaLink.AuditLog.Tests/Integration/Infrastructure/CombinedTelemetryHarness.cs new file mode 100644 index 0000000..538b269 --- /dev/null +++ b/tests/ScadaLink.AuditLog.Tests/Integration/Infrastructure/CombinedTelemetryHarness.cs @@ -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; + +/// +/// Shared end-to-end harness for the M3 cached-call combined telemetry tests +/// (G2/G3/G4). Composes the full pipeline: +/// +/// Site-local SQLite (in-memory) + +/// + . +/// Site-local SQLite (in-memory). +/// Production wrapped by a +/// test-side that also pushes each +/// packet through the stub gRPC client. +/// wired to the +/// dispatcher so a single observer call fans out audit + tracking + wire. +/// connected +/// to an backed by the real +/// + +/// against the per-test database. +/// +/// +/// +/// +/// Disposal cleans up the in-memory SQLite stores. The Akka actor system is +/// owned by the calling ; the harness +/// only owns the ingest actor IActorRef and the underlying repositories' +/// DbContext lifecycle. +/// +/// +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? 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.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.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.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.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.Instance); + Dispatcher = new CombinedTelemetryDispatcher(InnerForwarder, StubClient); + + Bridge = new CachedCallLifecycleBridge(Dispatcher, NullLogger.Instance); + } + + /// + /// 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). + /// + public Task EmitSubmitAsync(CachedCallTelemetry submit, CancellationToken ct = default) => + Dispatcher.ForwardAsync(submit, ct); + + /// + /// Convenience: route a per-attempt or terminal outcome through the bridge. + /// + public Task EmitAttemptAsync(CachedCallAttemptContext context, CancellationToken ct = default) => + Bridge.OnAttemptCompletedAsync(context, ct); + + public ScadaLinkDbContext CreateReadContext() + { + var options = new DbContextOptionsBuilder() + .UseSqlServer(_fixture.ConnectionString) + .Options; + return new ScadaLinkDbContext(options); + } + + private IServiceProvider BuildCentralServiceProvider( + Func? siteCallRepoOverride) + { + var services = new ServiceCollection(); + services.AddDbContext(opts => + opts.UseSqlServer(_fixture.ConnectionString) + .ConfigureWarnings(w => w.Ignore( + Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning))); + services.AddScoped(sp => + new AuditLogRepository(sp.GetRequiredService())); + if (siteCallRepoOverride is null) + { + services.AddScoped(sp => + new SiteCallAuditRepository(sp.GetRequiredService())); + } + else + { + services.AddScoped(sp => + siteCallRepoOverride(sp.GetRequiredService())); + } + 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(); + } + } +} diff --git a/tests/ScadaLink.AuditLog.Tests/ScadaLink.AuditLog.Tests.csproj b/tests/ScadaLink.AuditLog.Tests/ScadaLink.AuditLog.Tests.csproj index 625f9c6..bd99f49 100644 --- a/tests/ScadaLink.AuditLog.Tests/ScadaLink.AuditLog.Tests.csproj +++ b/tests/ScadaLink.AuditLog.Tests/ScadaLink.AuditLog.Tests.csproj @@ -48,6 +48,13 @@ the fixture + EF migrations come along without duplicating them. --> + +