Joseph Doherty
573 lines
26 KiB
C#
573 lines
26 KiB
C#
using System.Data.Common;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.EntityFrameworkCore.Diagnostics;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
|
|
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
|
|
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Repositories;
|
|
|
|
/// <summary>
|
|
/// M6 (K2) coverage for <see cref="KpiHistoryRepository"/> over the in-memory SQLite
|
|
/// harness. Exercises the bulk write, the single-series query (including the
|
|
/// null-ScopeKey Global match), and the retention purge.
|
|
/// </summary>
|
|
public class KpiHistoryRepositoryTests
|
|
{
|
|
// Fixed UTC base so the time assertions are deterministic.
|
|
private static readonly DateTime Base = new(2026, 6, 1, 0, 0, 0, DateTimeKind.Utc);
|
|
|
|
private static ScadaBridgeDbContext NewContext() => SqliteTestHelper.CreateInMemoryContext();
|
|
|
|
private static KpiSample Sample(
|
|
string source, string metric, string scope, string? scopeKey, double value, DateTime capturedAtUtc) =>
|
|
new()
|
|
{
|
|
Source = source,
|
|
Metric = metric,
|
|
Scope = scope,
|
|
ScopeKey = scopeKey,
|
|
Value = value,
|
|
CapturedAtUtc = capturedAtUtc,
|
|
};
|
|
|
|
[Fact]
|
|
public async Task GetRawSeriesAsync_ReturnsOnlyMatchingSeries_InAscendingTimeOrder()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// Two metrics ("queueDepth", "parkedCount") under one source; the target
|
|
// series ("queueDepth", Global, null key) has two points at different
|
|
// timestamps — inserted out of order to prove the OrderBy.
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
// Target series — Global / null ScopeKey, two points (later first).
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 7, capturedAtUtc: Base.AddMinutes(10)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(5)),
|
|
// Different metric — must be excluded.
|
|
Sample("NotificationOutbox", "parkedCount", "Global", null, value: 99, capturedAtUtc: Base.AddMinutes(5)),
|
|
// Same metric but a Site scope with a non-null key — must be excluded
|
|
// from the Global (null-key) query.
|
|
Sample("NotificationOutbox", "queueDepth", "Site", "plant-a", value: 42, capturedAtUtc: Base.AddMinutes(5)),
|
|
});
|
|
|
|
var series = await repo.GetRawSeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddMinutes(60));
|
|
|
|
// Only the two Global/null-key queueDepth points, ascending by capture time.
|
|
Assert.Equal(2, series.Count);
|
|
Assert.Equal(Base.AddMinutes(5), series[0].BucketStartUtc);
|
|
Assert.Equal(3, series[0].Value);
|
|
Assert.Equal(Base.AddMinutes(10), series[1].BucketStartUtc);
|
|
Assert.Equal(7, series[1].Value);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetRawSeriesAsync_SiteScopedKey_MatchesOnlyThatKey()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("SiteCallAudit", "buffered", "Site", "plant-a", value: 5, capturedAtUtc: Base.AddMinutes(5)),
|
|
Sample("SiteCallAudit", "buffered", "Site", "plant-b", value: 8, capturedAtUtc: Base.AddMinutes(5)),
|
|
// A Global (null-key) row with the same metric must NOT leak into a
|
|
// site-keyed query.
|
|
Sample("SiteCallAudit", "buffered", "Global", null, value: 13, capturedAtUtc: Base.AddMinutes(5)),
|
|
});
|
|
|
|
var series = await repo.GetRawSeriesAsync(
|
|
"SiteCallAudit", "buffered", "Site", scopeKey: "plant-a",
|
|
fromUtc: Base, toUtc: Base.AddMinutes(60));
|
|
|
|
Assert.Single(series);
|
|
Assert.Equal(5, series[0].Value);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetRawSeriesAsync_HonorsTimeWindowBounds()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("Health", "ageSeconds", "Global", null, value: 1, capturedAtUtc: Base),
|
|
Sample("Health", "ageSeconds", "Global", null, value: 2, capturedAtUtc: Base.AddMinutes(30)),
|
|
// Outside the [from, to] window — excluded.
|
|
Sample("Health", "ageSeconds", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(120)),
|
|
});
|
|
|
|
var series = await repo.GetRawSeriesAsync(
|
|
"Health", "ageSeconds", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddMinutes(60));
|
|
|
|
Assert.Equal(2, series.Count);
|
|
Assert.Equal(new[] { 1d, 2d }, series.Select(p => p.Value).ToArray());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RecordSamplesAsync_EmptyBatch_IsNoOp_AndGetRawSeriesAsync_EmptyTable_ReturnsEmpty()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// Empty batch must not throw (early-return guard, no SaveChanges round-trip).
|
|
await repo.RecordSamplesAsync(Array.Empty<KpiSample>());
|
|
|
|
// With nothing written, GetRawSeriesAsync must return a non-null empty list.
|
|
var series = await repo.GetRawSeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddMinutes(60));
|
|
|
|
Assert.NotNull(series);
|
|
Assert.Empty(series);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PurgeOlderThanAsync_DeletesOnlyRowsOlderThanCutoff_AndReturnsCount()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
// Two rows strictly older than the cutoff — should be purged.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 1, capturedAtUtc: Base.AddDays(-10)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 2, capturedAtUtc: Base.AddDays(-8)),
|
|
// One row AT the cutoff — strictly-older predicate keeps it.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddDays(-7)),
|
|
// One row newer than the cutoff — kept.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 4, capturedAtUtc: Base.AddDays(-1)),
|
|
});
|
|
|
|
var cutoff = Base.AddDays(-7);
|
|
var deleted = await repo.PurgeOlderThanAsync(cutoff);
|
|
|
|
Assert.Equal(2, deleted);
|
|
|
|
var remaining = await repo.GetRawSeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base.AddDays(-30), toUtc: Base.AddDays(30));
|
|
|
|
// The cutoff row (==) and the newer row survive.
|
|
Assert.Equal(2, remaining.Count);
|
|
Assert.Equal(new[] { 3d, 4d }, remaining.Select(p => p.Value).ToArray());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PurgeOlderThanAsync_MultiWindow_IssuesMoreThanOneDelete_AndReturnsTotal()
|
|
{
|
|
// Rows span several days, each in its own hour-bucket, all older than the
|
|
// cutoff. The batched purge must slice them into per-hour DELETE windows
|
|
// (arch-review 04, P4) rather than one giant DELETE — so more than one
|
|
// DELETE statement is issued, and the return value still equals the total.
|
|
var counter = new DeleteCountingInterceptor();
|
|
await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
// Three distinct hour-buckets strictly older than the cutoff — purged.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 1, capturedAtUtc: Base.AddDays(-5)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 2, capturedAtUtc: Base.AddDays(-4)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddDays(-3)),
|
|
// Newer than the cutoff — kept, and never scanned into a delete window.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 4, capturedAtUtc: Base.AddDays(-1)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 5, capturedAtUtc: Base),
|
|
});
|
|
|
|
// Only count DELETEs the purge itself issues (RecordSamples above emits none).
|
|
counter.Reset();
|
|
|
|
var deleted = await repo.PurgeOlderThanAsync(Base.AddDays(-2));
|
|
|
|
// All three old rows gone; the batched total is the sum across windows.
|
|
Assert.Equal(3, deleted);
|
|
Assert.True(counter.DeleteCount > 1,
|
|
$"expected the batched purge to issue more than one DELETE, got {counter.DeleteCount}");
|
|
|
|
var remaining = await repo.GetRawSeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base.AddDays(-30), toUtc: Base.AddDays(30));
|
|
Assert.Equal(new[] { 4d, 5d }, remaining.Select(p => p.Value).ToArray());
|
|
}
|
|
|
|
// ---- Hourly rollup fold / query / purge (plan #22 T3) ------------------------------
|
|
|
|
[Fact]
|
|
public async Task FoldHourlyRollupsAsync_GaugeMetric_FoldsToLastValuePerHour()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// "queueDepth" is a Gauge → last (latest-timestamp) value in the hour wins.
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(5)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 9, capturedAtUtc: Base.AddMinutes(59)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 7, capturedAtUtc: Base.AddMinutes(30)),
|
|
});
|
|
|
|
// toHourUtc = Base + 1 h (next hour start) → the Base hour is complete and folded.
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
|
|
|
|
var series = await repo.GetHourlySeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddHours(24));
|
|
|
|
var row = Assert.Single(series);
|
|
Assert.Equal(Base, row.BucketStartUtc);
|
|
Assert.Equal(9, row.Value); // last-value (max CapturedAtUtc)
|
|
}
|
|
|
|
[Fact]
|
|
public async Task FoldHourlyRollupsAsync_RateMetric_FoldsToSumPerHour_WithMinMaxCount()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// "deliveredLastInterval" is a Rate → sum of the hour's per-interval deltas.
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 2, capturedAtUtc: Base.AddMinutes(5)),
|
|
Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 5, capturedAtUtc: Base.AddMinutes(25)),
|
|
Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 4, capturedAtUtc: Base.AddMinutes(45)),
|
|
});
|
|
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
|
|
|
|
// Read the entity directly to assert Min/Max/Count fidelity, not just Value.
|
|
var rollup = Assert.Single(await ctx.KpiRollupHourly.ToListAsync());
|
|
Assert.Equal(Base, rollup.HourStartUtc);
|
|
Assert.Equal(11, rollup.Value); // sum 2+5+4
|
|
Assert.Equal(2, rollup.MinValue);
|
|
Assert.Equal(5, rollup.MaxValue);
|
|
Assert.Equal(3, rollup.SampleCount);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task FoldHourlyRollupsAsync_SplitsIntoSeparateHourBuckets_ForGlobalScope()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// Two hours of one Global (null ScopeKey) gauge series.
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("SiteHealth", "connectionsUp", "Global", null, value: 4, capturedAtUtc: Base.AddMinutes(10)),
|
|
Sample("SiteHealth", "connectionsUp", "Global", null, value: 6, capturedAtUtc: Base.AddMinutes(50)),
|
|
Sample("SiteHealth", "connectionsUp", "Global", null, value: 2, capturedAtUtc: Base.AddHours(1).AddMinutes(20)),
|
|
});
|
|
|
|
// Fold two complete hours: [Base, Base+2h).
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(2));
|
|
|
|
var series = await repo.GetHourlySeriesAsync(
|
|
"SiteHealth", "connectionsUp", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddHours(24));
|
|
|
|
Assert.Equal(2, series.Count);
|
|
Assert.Equal(Base, series[0].BucketStartUtc);
|
|
Assert.Equal(6, series[0].Value); // hour 0 last-value
|
|
Assert.Equal(Base.AddHours(1), series[1].BucketStartUtc);
|
|
Assert.Equal(2, series[1].Value); // hour 1 last-value
|
|
}
|
|
|
|
[Fact]
|
|
public async Task FoldHourlyRollupsAsync_ExcludesIncompleteHour_AtExclusiveUpperBound()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
// Complete hour (Base) — folded.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(30)),
|
|
// Next, in-progress hour — must NOT be folded when toHourUtc == Base+1h.
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 99, capturedAtUtc: Base.AddHours(1).AddMinutes(15)),
|
|
});
|
|
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
|
|
|
|
var series = await repo.GetHourlySeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddHours(24));
|
|
|
|
// Only the complete Base hour is present; the in-progress hour is excluded.
|
|
var row = Assert.Single(series);
|
|
Assert.Equal(Base, row.BucketStartUtc);
|
|
Assert.Equal(3, row.Value);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task FoldHourlyRollupsAsync_IsIdempotent_AcrossRepeatedRunsOverSameWindow()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 2, capturedAtUtc: Base.AddMinutes(5)),
|
|
Sample("NotificationOutbox", "deliveredLastInterval", "Global", null, value: 5, capturedAtUtc: Base.AddMinutes(45)),
|
|
});
|
|
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1)); // re-run same window
|
|
|
|
// Exactly one row for the (series, hour); the rate sum is not double-counted.
|
|
var rollups = await ctx.KpiRollupHourly.ToListAsync();
|
|
var row = Assert.Single(rollups);
|
|
Assert.Equal(7, row.Value); // still 2+5, not 14
|
|
Assert.Equal(2, row.SampleCount);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetHourlySeriesAsync_ReturnsAscending_AndHonorsNullVsSiteScopeKey()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// Same metric, one Global (null key) hour and one Site-keyed hour, both foldable.
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("SiteCallAudit", "buffered", "Global", null, value: 3, capturedAtUtc: Base.AddMinutes(10)),
|
|
Sample("SiteCallAudit", "buffered", "Global", null, value: 8, capturedAtUtc: Base.AddHours(1).AddMinutes(10)),
|
|
Sample("SiteCallAudit", "buffered", "Site", "plant-a", value: 42, capturedAtUtc: Base.AddMinutes(10)),
|
|
});
|
|
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(2));
|
|
|
|
var global = await repo.GetHourlySeriesAsync(
|
|
"SiteCallAudit", "buffered", "Global", scopeKey: null,
|
|
fromUtc: Base, toUtc: Base.AddHours(24));
|
|
|
|
// Two Global hours, ascending; the Site-keyed row must not leak in.
|
|
Assert.Equal(2, global.Count);
|
|
Assert.Equal(Base, global[0].BucketStartUtc);
|
|
Assert.Equal(3, global[0].Value);
|
|
Assert.Equal(Base.AddHours(1), global[1].BucketStartUtc);
|
|
Assert.Equal(8, global[1].Value);
|
|
|
|
var site = await repo.GetHourlySeriesAsync(
|
|
"SiteCallAudit", "buffered", "Site", scopeKey: "plant-a",
|
|
fromUtc: Base, toUtc: Base.AddHours(24));
|
|
Assert.Equal(42, Assert.Single(site).Value);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PurgeRollupsOlderThanAsync_DeletesOnlyRollupsOlderThanCutoff()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
|
|
// Fold four distinct hours spanning several days.
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 1, capturedAtUtc: Base.AddDays(-10).AddMinutes(5)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 2, capturedAtUtc: Base.AddDays(-8).AddMinutes(5)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 3, capturedAtUtc: Base.AddDays(-7).AddMinutes(5)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, value: 4, capturedAtUtc: Base.AddDays(-1).AddMinutes(5)),
|
|
});
|
|
await repo.FoldHourlyRollupsAsync(Base.AddDays(-30), Base);
|
|
|
|
// Cutoff = the -7d hour start; strictly-older predicate keeps the ==cutoff row.
|
|
var cutoff = TruncateHour(Base.AddDays(-7));
|
|
await repo.PurgeRollupsOlderThanAsync(cutoff);
|
|
|
|
var remaining = await repo.GetHourlySeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Global", scopeKey: null,
|
|
fromUtc: Base.AddDays(-30), toUtc: Base.AddDays(30));
|
|
|
|
// The -7d (==cutoff) and -1d rollups survive; the -10d and -8d are purged.
|
|
Assert.Equal(new[] { 3d, 4d }, remaining.Select(p => p.Value).ToArray());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task FoldHourlyRollups_DoesNotTrack_KpiSampleEntities()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, 5, Base.AddMinutes(10)),
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, 7, Base.AddMinutes(50)),
|
|
});
|
|
ctx.ChangeTracker.Clear(); // isolate the fold's tracking behavior from the seed
|
|
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
|
|
|
|
// The fold READS samples and WRITES rollups; the read must not register
|
|
// thousands of read-only KpiSample entries for DetectChanges to re-scan
|
|
// (arch-review 04 round 2, R2).
|
|
Assert.Empty(ctx.ChangeTracker.Entries<KpiSample>());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetLatestRollupHour_ReturnsNull_WhenNoRollupsExist()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
Assert.Null(await repo.GetLatestRollupHourAsync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetLatestRollupHour_ReturnsNewestHourStart_AcrossAllSeries()
|
|
{
|
|
await using var ctx = NewContext();
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "queueDepth", "Global", null, 5, Base.AddMinutes(10)),
|
|
Sample("SiteCallAudit", "buffered", "Global", null, 2, Base.AddHours(3).AddMinutes(10)),
|
|
});
|
|
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(4));
|
|
|
|
Assert.Equal(Base.AddHours(3), await repo.GetLatestRollupHourAsync());
|
|
}
|
|
|
|
// ---- Task 9: failover fold-race failure grain is the pass, not the row (R5) ----
|
|
|
|
[Fact]
|
|
public async Task Fold_LosingFailoverUpsertRace_DoesNotThrow_AndNextFoldSelfHeals()
|
|
{
|
|
// A single shared in-memory SQLite connection so a second context (the interceptor's
|
|
// side-write, and the "next tick" fold) sees the same database.
|
|
await using var connection = new Microsoft.Data.Sqlite.SqliteConnection("DataSource=:memory:");
|
|
await connection.OpenAsync();
|
|
await using (var schema = ContextOn(connection))
|
|
{
|
|
await schema.Database.EnsureCreatedAsync();
|
|
}
|
|
|
|
// The old singleton incarnation's in-flight fold wins the (series, hour) upsert race:
|
|
// the interceptor inserts the conflicting rollup row on the same connection just before
|
|
// THIS fold's SaveChanges, so the unique IX_KpiRollupHourly_Series faults the loser's save.
|
|
// A non-null ScopeKey (site-scoped series) is used deliberately: SQLite treats NULLs as
|
|
// distinct in a UNIQUE index, so only a non-null key deterministically forces the conflict.
|
|
var interceptor = new InsertConflictingRollupOnFirstSave(connection, () => new KpiRollupHourly
|
|
{
|
|
Source = "NotificationOutbox", Metric = "queueDepth", Scope = "Site", ScopeKey = "plant-a",
|
|
HourStartUtc = Base, Value = -1, MinValue = -1, MaxValue = -1, SampleCount = 1,
|
|
});
|
|
|
|
await using var ctx = ContextOn(connection, interceptor);
|
|
var repo = new KpiHistoryRepository(ctx);
|
|
await repo.RecordSamplesAsync(new[]
|
|
{
|
|
Sample("NotificationOutbox", "queueDepth", "Site", "plant-a", 5, Base.AddMinutes(10)),
|
|
});
|
|
|
|
// Losing the race must NOT propagate DbUpdateException — the failure grain is the pass.
|
|
var ex = await Record.ExceptionAsync(() => repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1)));
|
|
Assert.Null(ex);
|
|
|
|
// Next tick's fold (a fresh scope/context over the same DB) self-heals: it finds the
|
|
// conflicting row and idempotently upserts it in place to the correct folded value.
|
|
await using var ctx2 = ContextOn(connection);
|
|
var repo2 = new KpiHistoryRepository(ctx2);
|
|
await repo2.FoldHourlyRollupsAsync(Base, Base.AddHours(1));
|
|
|
|
var series = await repo2.GetHourlySeriesAsync(
|
|
"NotificationOutbox", "queueDepth", "Site", "plant-a", Base, Base.AddHours(1));
|
|
Assert.Single(series);
|
|
Assert.Equal(5, series[0].Value); // the fold's own value overwrote the side-inserted -1
|
|
}
|
|
|
|
// Local mirror of the repo's private hour-truncation, for cutoff assertions.
|
|
private static DateTime TruncateHour(DateTime utc) =>
|
|
new(utc.Year, utc.Month, utc.Day, utc.Hour, 0, 0, DateTimeKind.Utc);
|
|
|
|
/// <summary>Builds a SQLite test context bound to an externally-owned open connection
|
|
/// (so multiple contexts share one in-memory database), with optional interceptors.</summary>
|
|
private static ScadaBridgeDbContext ContextOn(DbConnection connection, params IInterceptor[] interceptors)
|
|
{
|
|
var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
|
|
.UseSqlite(connection)
|
|
.ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning))
|
|
.AddInterceptors(interceptors)
|
|
.Options;
|
|
return new SqliteTestDbContext(options);
|
|
}
|
|
|
|
/// <summary>
|
|
/// SaveChanges interceptor that, exactly once and only when a fold is about to insert a
|
|
/// <see cref="KpiRollupHourly"/>, inserts a conflicting (series, hour) rollup row through a
|
|
/// second context on the same connection — deterministically reproducing the failover-overlap
|
|
/// upsert race the losing fold must survive (arch-review 04 round 2, R5).
|
|
/// </summary>
|
|
private sealed class InsertConflictingRollupOnFirstSave : Microsoft.EntityFrameworkCore.Diagnostics.SaveChangesInterceptor
|
|
{
|
|
private readonly DbConnection _connection;
|
|
private readonly Func<KpiRollupHourly> _rowFactory;
|
|
private bool _fired;
|
|
|
|
public InsertConflictingRollupOnFirstSave(DbConnection connection, Func<KpiRollupHourly> rowFactory)
|
|
{
|
|
_connection = connection;
|
|
_rowFactory = rowFactory;
|
|
}
|
|
|
|
public override async ValueTask<InterceptionResult<int>> SavingChangesAsync(
|
|
DbContextEventData eventData, InterceptionResult<int> result,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
if (!_fired && eventData.Context is not null &&
|
|
eventData.Context.ChangeTracker.Entries<KpiRollupHourly>()
|
|
.Any(e => e.State == EntityState.Added))
|
|
{
|
|
_fired = true;
|
|
var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
|
|
.UseSqlite(_connection)
|
|
.ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning))
|
|
.Options;
|
|
await using var side = new SqliteTestDbContext(options);
|
|
side.KpiRollupHourly.Add(_rowFactory());
|
|
await side.SaveChangesAsync(cancellationToken);
|
|
}
|
|
|
|
return await base.SavingChangesAsync(eventData, result, cancellationToken);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// EF command interceptor that counts the DELETE statements actually issued
|
|
/// to the store, so a test can prove the purge is sliced into multiple
|
|
/// windowed DELETEs rather than one mega-statement. Threads both the sync and
|
|
/// async non-query entry points (<c>ExecuteDeleteAsync</c> routes through the
|
|
/// async path).
|
|
/// </summary>
|
|
private sealed class DeleteCountingInterceptor : DbCommandInterceptor
|
|
{
|
|
public int DeleteCount { get; private set; }
|
|
|
|
public void Reset() => DeleteCount = 0;
|
|
|
|
private void CountIfDelete(DbCommand command)
|
|
{
|
|
if (command.CommandText.Contains("DELETE", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
DeleteCount++;
|
|
}
|
|
}
|
|
|
|
public override InterceptionResult<int> NonQueryExecuting(
|
|
DbCommand command, CommandEventData eventData, InterceptionResult<int> result)
|
|
{
|
|
CountIfDelete(command);
|
|
return base.NonQueryExecuting(command, eventData, result);
|
|
}
|
|
|
|
public override ValueTask<InterceptionResult<int>> NonQueryExecutingAsync(
|
|
DbCommand command, CommandEventData eventData, InterceptionResult<int> result,
|
|
CancellationToken cancellationToken = default)
|
|
{
|
|
CountIfDelete(command);
|
|
return base.NonQueryExecutingAsync(command, eventData, result, cancellationToken);
|
|
}
|
|
}
|
|
}
|