using Microsoft.EntityFrameworkCore; using ScadaLink.Commons.Entities.Audit; using ScadaLink.Commons.Types; using ScadaLink.Commons.Types.Audit; using ScadaLink.ConfigurationDatabase; using ScadaLink.ConfigurationDatabase.Repositories; using ScadaLink.ConfigurationDatabase.Tests.Migrations; using Xunit; namespace ScadaLink.ConfigurationDatabase.Tests.Repositories; /// /// Bundle B3 (#22, #23 M3) integration tests for . /// Uses the same as the Bundle B2 migration tests so /// the monotonic-upsert SQL executes against the real SiteCalls schema. Each test /// scopes its data by minting a fresh (or a per-test /// SourceSite suffix) so tests neither collide nor require teardown. /// public class SiteCallAuditRepositoryTests : IClassFixture { private readonly MsSqlMigrationFixture _fixture; public SiteCallAuditRepositoryTests(MsSqlMigrationFixture fixture) { _fixture = fixture; } [SkippableFact] public async Task UpsertAsync_FreshId_InsertsOneRow() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); var row = NewRow(id, status: "Submitted", retryCount: 0); await repo.UpsertAsync(row); await using var readContext = CreateContext(); var loaded = await readContext.Set() .Where(s => s.TrackedOperationId == id) .ToListAsync(); Assert.Single(loaded); Assert.Equal("Submitted", loaded[0].Status); Assert.Equal(0, loaded[0].RetryCount); } [SkippableFact] public async Task UpsertAsync_AdvancedStatus_UpdatesRow() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); // Submitted (rank 0) → Forwarded (rank 1) → Attempted (rank 2) — every // step strictly advances the rank, so each upsert must mutate the row. await repo.UpsertAsync(NewRow(id, status: "Submitted", retryCount: 0)); await repo.UpsertAsync(NewRow(id, status: "Forwarded", retryCount: 0)); await repo.UpsertAsync(NewRow(id, status: "Attempted", retryCount: 1, lastError: "transient 503")); var loaded = await repo.GetAsync(id); Assert.NotNull(loaded); Assert.Equal("Attempted", loaded!.Status); Assert.Equal(1, loaded.RetryCount); Assert.Equal("transient 503", loaded.LastError); } [SkippableFact] public async Task UpsertAsync_OlderStatus_IsNoOp_RowUnchanged() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); // First land Attempted (rank 2). A late-arriving Submitted (rank 0) must // NOT roll the row back — silent no-op. await repo.UpsertAsync(NewRow(id, status: "Attempted", retryCount: 5, lastError: "transient")); var attemptedSnapshot = await repo.GetAsync(id); await repo.UpsertAsync(NewRow(id, status: "Submitted", retryCount: 0, lastError: null)); var afterStale = await repo.GetAsync(id); Assert.NotNull(afterStale); Assert.Equal("Attempted", afterStale!.Status); Assert.Equal(5, afterStale.RetryCount); Assert.Equal("transient", afterStale.LastError); // UpdatedAtUtc should not have moved when the stale write was rejected. Assert.Equal(attemptedSnapshot!.UpdatedAtUtc, afterStale.UpdatedAtUtc); } [SkippableFact] public async Task UpsertAsync_SameStatus_IsNoOp() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); await repo.UpsertAsync(NewRow(id, status: "Attempted", retryCount: 1, lastError: "first")); var snapshot = await repo.GetAsync(id); // Same rank (2) — repository must treat this as a no-op (no fields move). await repo.UpsertAsync(NewRow(id, status: "Attempted", retryCount: 2, lastError: "second")); var afterDuplicate = await repo.GetAsync(id); Assert.NotNull(afterDuplicate); Assert.Equal("Attempted", afterDuplicate!.Status); Assert.Equal(1, afterDuplicate.RetryCount); Assert.Equal("first", afterDuplicate.LastError); Assert.Equal(snapshot!.UpdatedAtUtc, afterDuplicate.UpdatedAtUtc); } [SkippableFact] public async Task UpsertAsync_TerminalOverTerminal_IsNoOp() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); // Bundle B3 plan: terminal statuses share rank 3 and are mutually // exclusive — Delivered cannot overwrite Parked. var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); await repo.UpsertAsync(NewRow(id, status: "Parked", retryCount: 3, lastError: "parked-reason", terminal: true)); var afterPark = await repo.GetAsync(id); await repo.UpsertAsync(NewRow(id, status: "Delivered", retryCount: 4, lastError: null, terminal: true)); var afterDeliveredAttempt = await repo.GetAsync(id); Assert.NotNull(afterDeliveredAttempt); Assert.Equal("Parked", afterDeliveredAttempt!.Status); Assert.Equal("parked-reason", afterDeliveredAttempt.LastError); Assert.Equal(afterPark!.UpdatedAtUtc, afterDeliveredAttempt.UpdatedAtUtc); } [SkippableFact] public async Task UpsertAsync_ConcurrentInserts_SameId_OnlyOneRow() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); // 50 parallel inserters with the same id. The IF NOT EXISTS … INSERT // pattern has a check-then-act race; concurrent losers must surface as // silent duplicate-key swallows, not thrown exceptions. Final row // count must be exactly 1. var id = TrackedOperationId.New(); var row = NewRow(id, status: "Submitted", retryCount: 0); await Parallel.ForEachAsync( Enumerable.Range(0, 50), new ParallelOptions { MaxDegreeOfParallelism = 50 }, async (_, ct) => { await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); await repo.UpsertAsync(row, ct); }); await using var readContext = CreateContext(); var count = await readContext.Set() .Where(s => s.TrackedOperationId == id) .CountAsync(); Assert.Equal(1, count); } [SkippableFact] public async Task GetAsync_KnownId_ReturnsRow() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); await repo.UpsertAsync(NewRow(id, status: "Submitted", retryCount: 0)); var loaded = await repo.GetAsync(id); Assert.NotNull(loaded); Assert.Equal(id, loaded!.TrackedOperationId); } [SkippableFact] public async Task GetAsync_UnknownId_ReturnsNull() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); var loaded = await repo.GetAsync(TrackedOperationId.New()); Assert.Null(loaded); } [SkippableFact] public async Task QueryAsync_FilterBySourceSite_ReturnsMatchingRows() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var siteA = NewSiteId(); var siteB = NewSiteId(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); var t0 = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); await repo.UpsertAsync(NewRow(TrackedOperationId.New(), sourceSite: siteA, createdAtUtc: t0)); await repo.UpsertAsync(NewRow(TrackedOperationId.New(), sourceSite: siteA, createdAtUtc: t0.AddMinutes(1))); await repo.UpsertAsync(NewRow(TrackedOperationId.New(), sourceSite: siteB, createdAtUtc: t0.AddMinutes(2))); var rows = await repo.QueryAsync( new SiteCallQueryFilter(SourceSite: siteA), new SiteCallPaging(PageSize: 10)); Assert.Equal(2, rows.Count); Assert.All(rows, r => Assert.Equal(siteA, r.SourceSite)); } [SkippableFact] public async Task QueryAsync_KeysetPaging_NoOverlap() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var site = NewSiteId(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); // Five rows with distinct CreatedAtUtc. Page-size 2 → page 1 returns // minutes 4,3; cursor (minutes 3) → page 2 returns minutes 2,1; cursor // (minutes 1) → page 3 returns minute 0. var t0 = new DateTime(2026, 5, 20, 9, 0, 0, DateTimeKind.Utc); for (var i = 0; i < 5; i++) { await repo.UpsertAsync(NewRow(TrackedOperationId.New(), sourceSite: site, createdAtUtc: t0.AddMinutes(i))); } var page1 = await repo.QueryAsync( new SiteCallQueryFilter(SourceSite: site), new SiteCallPaging(PageSize: 2)); Assert.Equal(2, page1.Count); Assert.Equal(t0.AddMinutes(4), page1[0].CreatedAtUtc); Assert.Equal(t0.AddMinutes(3), page1[1].CreatedAtUtc); var cursor1 = page1[^1]; var page2 = await repo.QueryAsync( new SiteCallQueryFilter(SourceSite: site), new SiteCallPaging( PageSize: 2, AfterCreatedAtUtc: cursor1.CreatedAtUtc, AfterId: cursor1.TrackedOperationId)); Assert.Equal(2, page2.Count); Assert.Equal(t0.AddMinutes(2), page2[0].CreatedAtUtc); Assert.Equal(t0.AddMinutes(1), page2[1].CreatedAtUtc); var cursor2 = page2[^1]; var page3 = await repo.QueryAsync( new SiteCallQueryFilter(SourceSite: site), new SiteCallPaging( PageSize: 2, AfterCreatedAtUtc: cursor2.CreatedAtUtc, AfterId: cursor2.TrackedOperationId)); Assert.Single(page3); Assert.Equal(t0.AddMinutes(0), page3[0].CreatedAtUtc); // No overlap across pages. var allIds = page1.Concat(page2).Concat(page3).Select(r => r.TrackedOperationId).ToHashSet(); Assert.Equal(5, allIds.Count); } [SkippableFact] public async Task PurgeTerminalAsync_RemovesTerminalAndOld() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var site = NewSiteId(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); // One row that's been Delivered for a long time (5 days ago) — should be purged. var oldId = TrackedOperationId.New(); var fiveDaysAgo = DateTime.UtcNow.AddDays(-5); await repo.UpsertAsync(NewRow( oldId, sourceSite: site, status: "Delivered", retryCount: 1, createdAtUtc: fiveDaysAgo.AddMinutes(-1), updatedAtUtc: fiveDaysAgo, terminal: true, terminalAtUtc: fiveDaysAgo)); var purged = await repo.PurgeTerminalAsync(DateTime.UtcNow.AddDays(-1)); Assert.True(purged >= 1, $"Expected at least one purged row; got {purged}."); Assert.Null(await repo.GetAsync(oldId)); } [SkippableFact] public async Task PurgeTerminalAsync_KeepsNonTerminalAndRecent() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var site = NewSiteId(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); // Non-terminal row: never eligible. var activeId = TrackedOperationId.New(); await repo.UpsertAsync(NewRow( activeId, sourceSite: site, status: "Attempted", retryCount: 1, createdAtUtc: DateTime.UtcNow.AddDays(-10), updatedAtUtc: DateTime.UtcNow.AddDays(-10), terminal: false)); // Recent terminal row: TerminalAtUtc within the keep window. var recentTerminalId = TrackedOperationId.New(); await repo.UpsertAsync(NewRow( recentTerminalId, sourceSite: site, status: "Delivered", retryCount: 0, createdAtUtc: DateTime.UtcNow.AddHours(-2), updatedAtUtc: DateTime.UtcNow.AddHours(-1), terminal: true, terminalAtUtc: DateTime.UtcNow.AddHours(-1))); // Purge older than 1 day — both rows must survive. await repo.PurgeTerminalAsync(DateTime.UtcNow.AddDays(-1)); Assert.NotNull(await repo.GetAsync(activeId)); Assert.NotNull(await repo.GetAsync(recentTerminalId)); } // --- KPI snapshot tests ------------------------------------------------- [SkippableFact] public async Task ComputeKpisAsync_CountsBufferedParkedFailedDeliveredAndStuck() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var site = NewSiteId(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); var now = DateTime.UtcNow; var stuckCutoff = now.AddMinutes(-10); var intervalSince = now.AddHours(-1); // Buffered + stuck (non-terminal Attempted, created 30 min ago). await repo.UpsertAsync(NewRow( TrackedOperationId.New(), site, status: "Attempted", createdAtUtc: now.AddMinutes(-30))); // Buffered but NOT stuck (non-terminal Attempted, created 2 min ago). await repo.UpsertAsync(NewRow( TrackedOperationId.New(), site, status: "Attempted", createdAtUtc: now.AddMinutes(-2))); // Parked (terminal). await repo.UpsertAsync(NewRow( TrackedOperationId.New(), site, status: "Parked", createdAtUtc: now.AddMinutes(-5), updatedAtUtc: now.AddMinutes(-4), terminal: true, terminalAtUtc: now.AddMinutes(-4))); // Delivered within the interval. await repo.UpsertAsync(NewRow( TrackedOperationId.New(), site, status: "Delivered", createdAtUtc: now.AddMinutes(-4), updatedAtUtc: now.AddMinutes(-1), terminal: true, terminalAtUtc: now.AddMinutes(-1))); // Failed within the interval. await repo.UpsertAsync(NewRow( TrackedOperationId.New(), site, status: "Failed", createdAtUtc: now.AddMinutes(-6), updatedAtUtc: now.AddMinutes(-2), terminal: true, terminalAtUtc: now.AddMinutes(-2))); // Delivered OUTSIDE the interval (2 hours ago) — must not count. await repo.UpsertAsync(NewRow( TrackedOperationId.New(), site, status: "Delivered", createdAtUtc: now.AddHours(-3), updatedAtUtc: now.AddHours(-2), terminal: true, terminalAtUtc: now.AddHours(-2))); var snapshot = await repo.ComputeKpisAsync(stuckCutoff, intervalSince); // Counts are global; assert the floor since the table is shared with // other tests. The OUTSIDE-interval Delivered row proves the window // bounds the throughput counts. Assert.True(snapshot.BufferedCount >= 2); Assert.True(snapshot.ParkedCount >= 1); Assert.True(snapshot.StuckCount >= 1); Assert.True(snapshot.DeliveredLastInterval >= 1); Assert.True(snapshot.FailedLastInterval >= 1); Assert.NotNull(snapshot.OldestPendingAge); Assert.True(snapshot.OldestPendingAge >= TimeSpan.FromMinutes(25)); } [SkippableFact] public async Task ComputePerSiteKpisAsync_ScopesCountsToEachSite() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var siteA = NewSiteId(); var siteB = NewSiteId(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); var now = DateTime.UtcNow; var stuckCutoff = now.AddMinutes(-10); var intervalSince = now.AddHours(-1); // siteA: 2 buffered (one stuck), 1 parked. await repo.UpsertAsync(NewRow(TrackedOperationId.New(), siteA, status: "Attempted", createdAtUtc: now.AddMinutes(-30))); await repo.UpsertAsync(NewRow(TrackedOperationId.New(), siteA, status: "Attempted", createdAtUtc: now.AddMinutes(-2))); await repo.UpsertAsync(NewRow( TrackedOperationId.New(), siteA, status: "Parked", createdAtUtc: now.AddMinutes(-5), updatedAtUtc: now.AddMinutes(-4), terminal: true, terminalAtUtc: now.AddMinutes(-4))); // siteB: 1 delivered within interval only. await repo.UpsertAsync(NewRow( TrackedOperationId.New(), siteB, status: "Delivered", createdAtUtc: now.AddMinutes(-4), updatedAtUtc: now.AddMinutes(-1), terminal: true, terminalAtUtc: now.AddMinutes(-1))); var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, intervalSince); var a = Assert.Single(perSite, s => s.SourceSite == siteA); Assert.Equal(2, a.BufferedCount); Assert.Equal(1, a.ParkedCount); Assert.Equal(1, a.StuckCount); Assert.NotNull(a.OldestPendingAge); var b = Assert.Single(perSite, s => s.SourceSite == siteB); Assert.Equal(0, b.BufferedCount); Assert.Equal(1, b.DeliveredLastInterval); // siteB has no non-terminal rows — no oldest-pending age. Assert.Null(b.OldestPendingAge); } // --- helpers ------------------------------------------------------------ private ScadaLinkDbContext CreateContext() { var options = new DbContextOptionsBuilder() .UseSqlServer(_fixture.ConnectionString) .Options; return new ScadaLinkDbContext(options); } private static string NewSiteId() => "site-b3-" + Guid.NewGuid().ToString("N").Substring(0, 8); private static SiteCall NewRow( TrackedOperationId id, string? sourceSite = null, string status = "Submitted", int retryCount = 0, string? lastError = null, int? httpStatus = null, DateTime? createdAtUtc = null, DateTime? updatedAtUtc = null, bool terminal = false, DateTime? terminalAtUtc = null) { var created = createdAtUtc ?? DateTime.UtcNow; var updated = updatedAtUtc ?? created; DateTime? terminalAt = terminal ? (terminalAtUtc ?? updated) : null; return new SiteCall { TrackedOperationId = id, Channel = "ApiOutbound", Target = "ERP.GetOrder", SourceSite = sourceSite ?? NewSiteId(), Status = status, RetryCount = retryCount, LastError = lastError, HttpStatus = httpStatus, CreatedAtUtc = created, UpdatedAtUtc = updated, TerminalAtUtc = terminalAt, IngestedAtUtc = DateTime.UtcNow, }; } }