using System.Text.Json; using Microsoft.Data.SqlClient; using Microsoft.EntityFrameworkCore; using ZB.MOM.WW.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations; namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Maintenance; /// /// SQL Server integration tests for the AuditLog Parquet archive (ScadaBridge#42) against the /// real migrated, partitioned schema: archive-then-purge leaves the month's hash chain /// verifiable and the archive re-verifiable against the purge anchor; the purge refuses to /// drop a partition that changed after it was archived; tampering with an archive is caught. /// /// /// Every test owns a distinct month (partition) of the shared fixture database; the chain /// and the per-channel purge are table-wide, so each test drains the chain itself. /// public sealed class AuditLogPartitionArchiveTests : IClassFixture, IDisposable { private readonly MsSqlMigrationFixture _fixture; private readonly DirectoryInfo _root = Directory.CreateTempSubdirectory("auditlog-archive-"); /// Creates the test class with the shared migrated SQL Server fixture. /// The shared migration fixture. public AuditLogPartitionArchiveTests(MsSqlMigrationFixture fixture) => _fixture = fixture; /// public void Dispose() => _root.Delete(recursive: true); /// /// Archive, then purge: the archive holds exactly the partition's rows (chained, /// unchained, with their chain columns) and its tombstones; after the switch the month's /// chain still verifies from the anchor, and the archive verifies against that anchor. /// /// A task that represents the asynchronous operation. [SkippableFact] public async Task ArchiveThenPurge_KeepsTheChainVerifiable_AndTheArchiveMatchesTheAnchor() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var month = Month(2026, 2); await InsertAsync(Enumerable.Range(1, 10).Select(i => NewEvent(month.AddDays(i).AddTicks(i * 7_777)))); // Two rows of a short-retention channel, purged per channel mid-chain → tombstones. await InsertAsync([NewEvent(month.AddHours(1), AuditChannel.DbOutbound), NewEvent(month.AddHours(2), AuditChannel.DbOutbound)]); await DrainAsync(); await using (var ctx = CreateContext()) { Assert.Equal(2, await new AuditLogRepository(ctx).PurgeChannelOlderThanAsync("DbOutbound", month.AddDays(1), 100)); } await InsertAsync([NewEvent(month.AddDays(20))]); // written after the drain: unchained var before = await PartitionRowsAsync(month); Assert.Equal(11, before.Count); var manifest = await ArchiveAsync(month); Assert.Equal(AuditArchiveStatus.Archived, manifest.Status); Assert.Equal(11, manifest.RowCount); Assert.Equal(10, manifest.ChainedRowCount); Assert.Equal(1, manifest.UnchainedRowCount); Assert.Equal(2, manifest.TombstoneCount); var chain = Assert.Single(manifest.Chains); Assert.Equal("2026-02", chain.Partition); Assert.Equal(12, chain.LastSeq); // Row equality: the archive's data file holds exactly the partition's stored rows. var archived = await ReadArchiveRowsAsync(manifest); Assert.Equal(before.Select(Key).Order(), archived.Select(Key).Order()); await using (var ctx = CreateContext()) { Assert.Equal(11, await Archive(ctx).PurgeArchivedPartitionAsync(manifest)); } Assert.Empty(await PartitionRowsAsync(month)); var monthChain = await VerifyMonthAsync(month); Assert.Equal(AuditChainStatus.Intact, monthChain.Status); Assert.Equal(12, monthChain.PurgedThroughSeq); var verification = await VerifyArchiveAsync(manifest.Name); Assert.True(verification.Verified, verification.FileProblem); Assert.Equal(AuditArchiveStatus.Purged, verification.ManifestStatus); Assert.Equal(11, verification.RowsRead); var v = Assert.Single(verification.Chains); Assert.Equal(AuditChainStatus.Intact, v.Chain.Status); Assert.Equal(10, v.Chain.RowsVerified); Assert.Equal(2, v.Chain.TombstonesApplied); Assert.Equal(1, v.Chain.UnchainedRows); Assert.Equal(AuditArchiveAnchorStatus.Matches, v.Anchor); await using (var ctx = CreateContext()) { var listed = Assert.Single(await Archive(ctx).ListAsync(), m => m.Partition == "2026-02"); Assert.Equal(AuditArchiveStatus.Purged, listed.Status); Assert.NotNull(listed.PurgedAtUtc); } } /// /// A row that lands in the partition after it was archived makes the purge refuse /// (nothing dropped); a fresh archive then purges. The stale archive still verifies as a /// complete, un-purged copy. /// /// A task that represents the asynchronous operation. [SkippableFact] public async Task RowArrivingAfterTheArchive_StopsThePurge() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var month = Month(2026, 3); await InsertAsync(Enumerable.Range(1, 5).Select(i => NewEvent(month.AddDays(i)))); await DrainAsync(); var stale = await ArchiveAsync(month); await InsertAsync([NewEvent(month.AddDays(9))]); await using (var ctx = CreateContext()) { var ex = await Assert.ThrowsAsync(() => Archive(ctx).PurgeArchivedPartitionAsync(stale)); Assert.Equal(AuditPartitionSwitchGuard.RowCountChangedError, ((SqlException)ex.InnerException!).Number); } Assert.Equal(6, (await PartitionRowsAsync(month)).Count); Assert.Null((await VerifyMonthAsync(month)).PurgedThroughSeq); var staleVerification = await VerifyArchiveAsync(stale.Name); Assert.True(staleVerification.Verified); Assert.Equal(AuditArchiveStatus.Archived, staleVerification.ManifestStatus); Assert.Equal(AuditArchiveAnchorStatus.NotPurged, Assert.Single(staleVerification.Chains).Anchor); // Positive control: a fresh archive holds the new row and is allowed to purge. var fresh = await ArchiveAsync(month); Assert.Equal(6, fresh.RowCount); await using (var ctx = CreateContext()) { Assert.Equal(6, await Archive(ctx).PurgeArchivedPartitionAsync(fresh)); } Assert.Empty(await PartitionRowsAsync(month)); } /// /// A chaining pass between the archive and the purge (the row count is unchanged, the /// chain head moved) also stops the purge: otherwise the anchor would cover links the /// archive does not hold. /// /// A task that represents the asynchronous operation. [SkippableFact] public async Task ChainAdvancingAfterTheArchive_StopsThePurge() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var month = Month(2026, 4); await InsertAsync(Enumerable.Range(1, 4).Select(i => NewEvent(month.AddDays(i)))); await DrainAsync(); await InsertAsync([NewEvent(month.AddDays(8))]); // unchained at archive time var manifest = await ArchiveAsync(month); Assert.Equal(1, manifest.UnchainedRowCount); await DrainAsync(); // links the row: head 4 → 5, row count unchanged await using (var ctx = CreateContext()) { var ex = await Assert.ThrowsAsync(() => Archive(ctx).PurgeArchivedPartitionAsync(manifest)); Assert.Equal(AuditPartitionSwitchGuard.ChainAdvancedError, ((SqlException)ex.InnerException!).Number); } Assert.Equal(5, (await PartitionRowsAsync(month)).Count); } /// /// Tampering with a purged archive is caught three ways: a flipped byte (file hash), a /// re-written file with an edited row and a re-hashed manifest (chain), and a forger who /// also recomputes the whole chain (the purge anchor in the database no longer matches). /// /// A task that represents the asynchronous operation. [SkippableFact] public async Task TamperedArchive_FailsVerification() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var month = Month(2026, 5); await InsertAsync(Enumerable.Range(1, 8).Select(i => NewEvent(month.AddDays(i)))); await DrainAsync(); var manifest = await ArchiveAndPurgeAsync(month); Assert.True((await VerifyArchiveAsync(manifest.Name)).Verified); var dataPath = Path.Combine(_root.FullName, "2026", manifest.DataFile); var manifestPath = Path.Combine(_root.FullName, "2026", manifest.Name + ".manifest.json"); var original = await File.ReadAllBytesAsync(dataPath); var originalManifest = await File.ReadAllTextAsync(manifestPath); // 1. A flipped byte. var flipped = (byte[])original.Clone(); flipped[original.Length / 2] ^= 0xFF; await File.WriteAllBytesAsync(dataPath, flipped); var result = await VerifyArchiveAsync(manifest.Name); Assert.False(result.Verified); Assert.False(result.FilesIntact); Assert.Contains("SHA-256", result.FileProblem); // 2. An edited row in a re-written file, manifest hash/size updated to match. var rows = await ReadArchiveRowsAsync(manifest, original); var victim = rows[3]; rows[3] = victim with { Target = "forged" }; await RewriteAsync(rows, manifest, dataPath, manifestPath, lastHash: null); result = await VerifyArchiveAsync(manifest.Name); Assert.False(result.Verified); Assert.True(result.FilesIntact); var brk = Assert.Single(result.Chains).Chain.FirstBreak!; Assert.Equal(AuditChainBreakKind.HashMismatch, brk.Kind); Assert.Equal(victim.EventId, brk.EventId); // 3. The forger recomputes every link and the manifest's chain end: the archive is // self-consistent, but its end no longer matches the anchor the purge left. var previous = AuditChainHasher.GenesisHash.ToArray(); for (var i = 0; i < rows.Count; i++) { var hash = AuditChainHasher.ComputeRowHash(previous, month, rows[i].ChainSeq!.Value, rows[i]); rows[i] = rows[i] with { RowHash = hash }; previous = hash; } await RewriteAsync(rows, manifest, dataPath, manifestPath, lastHash: Convert.ToHexStringLower(previous)); result = await VerifyArchiveAsync(manifest.Name); Assert.False(result.Verified); var c = Assert.Single(result.Chains); Assert.Equal(AuditChainStatus.Intact, c.Chain.Status); Assert.Equal(AuditArchiveAnchorStatus.Mismatch, c.Anchor); // Positive control: the untouched archive still verifies. await File.WriteAllBytesAsync(dataPath, original); await File.WriteAllTextAsync(manifestPath, originalManifest); Assert.True((await VerifyArchiveAsync(manifest.Name)).Verified); } /// /// Rows that arrive for a month after it was purged chain on from the anchor; archiving /// and purging them again produces a second archive that resumes from the first one's end. /// The first archive then reports its anchor as superseded, and both verify. /// /// A task that represents the asynchronous operation. [SkippableFact] public async Task LateRows_AreArchivedAgain_FromTheAnchor() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); var month = Month(2026, 6); await InsertAsync(Enumerable.Range(1, 3).Select(i => NewEvent(month.AddDays(i)))); await DrainAsync(); var first = await ArchiveAndPurgeAsync(month); await InsertAsync([NewEvent(month.AddDays(15)), NewEvent(month.AddDays(16))]); await DrainAsync(); await Task.Delay(5); // distinct archive name (millisecond stamp) var second = await ArchiveAndPurgeAsync(month); var chain = Assert.Single(second.Chains); Assert.Equal(3, chain.PriorPurgedThroughSeq); Assert.Equal(5, chain.LastSeq); var secondResult = await VerifyArchiveAsync(second.Name); Assert.True(secondResult.Verified); Assert.Equal(AuditArchiveAnchorStatus.Matches, Assert.Single(secondResult.Chains).Anchor); Assert.Equal(2, Assert.Single(secondResult.Chains).Chain.RowsVerified); var firstResult = await VerifyArchiveAsync(first.Name); Assert.True(firstResult.Verified); Assert.Equal(AuditArchiveAnchorStatus.Superseded, Assert.Single(firstResult.Chains).Anchor); } /// A name that is not an archive name (e.g. a path) is rejected before any file access. /// A task that represents the asynchronous operation. [SkippableFact] public async Task VerifyRejectsPathLikeNames() { Skip.IfNot(_fixture.Available, _fixture.SkipReason); await using var ctx = CreateContext(); await Assert.ThrowsAsync(() => Archive(ctx).VerifyAsync("../../etc/passwd")); await Assert.ThrowsAsync(() => Archive(ctx).VerifyAsync("auditlog-2026-01-20260101T000000000Z")); } // ── helpers ────────────────────────────────────────────────────────────── private static DateTime Month(int year, int month) => new(year, month, 1, 0, 0, 0, DateTimeKind.Utc); private AuditLogPartitionArchive Archive(ScadaBridgeDbContext ctx) => new(ctx, new AuditLogArchiveOptions { Enabled = true, Directory = _root.FullName, RowGroupSize = 1_000, }); private async Task ArchiveAsync(DateTime month) { await using var ctx = CreateContext(); return await Archive(ctx).ArchivePartitionAsync(month); } private async Task ArchiveAndPurgeAsync(DateTime month) { var manifest = await ArchiveAsync(month); await using var ctx = CreateContext(); await Archive(ctx).PurgeArchivedPartitionAsync(manifest); return manifest; } private async Task VerifyArchiveAsync(string name) { await using var ctx = CreateContext(); return await Archive(ctx).VerifyAsync(name); } private async Task VerifyMonthAsync(DateTime month) { await using var ctx = CreateContext(); return await new AuditLogHashChain(ctx).VerifyMonthAsync(month); } private async Task> ReadArchiveRowsAsync(AuditArchiveManifest manifest, byte[]? bytes = null) { bytes ??= await File.ReadAllBytesAsync(Path.Combine(_root.FullName, manifest.Partition[..4], manifest.DataFile)); var rows = new List(); await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes))) { rows.Add(row); } return rows; } private static async Task RewriteAsync( List rows, AuditArchiveManifest manifest, string dataPath, string manifestPath, string? lastHash) { var stream = new MemoryStream(); await using (var writer = await AuditArchiveParquet.CreateDataWriterAsync(stream, manifest.Partition, 1_000)) { foreach (var row in rows) { await writer.AddAsync(row); } } var bytes = stream.ToArray(); await File.WriteAllBytesAsync(dataPath, bytes); var forged = manifest with { Status = AuditArchiveStatus.Purged, DataSha256 = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(bytes)), DataBytes = bytes.Length, Chains = lastHash is null ? manifest.Chains : [manifest.Chains[0] with { LastHash = lastHash }], }; await File.WriteAllTextAsync(manifestPath, JsonSerializer.Serialize(forged, new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.CamelCase, })); } private async Task> PartitionRowsAsync(DateTime month) { await using var conn = new SqlConnection(_fixture.ConnectionString); await conn.OpenAsync(); await using var cmd = conn.CreateCommand(); cmd.CommandText = $"SELECT {AuditLogHashChain.RowColumns}, ChainSeq, RowHash FROM dbo.AuditLog " + "WHERE OccurredAtUtc >= @s AND OccurredAtUtc < @e;"; cmd.Parameters.Add(new SqlParameter("@s", System.Data.SqlDbType.DateTime2) { Value = month }); cmd.Parameters.Add(new SqlParameter("@e", System.Data.SqlDbType.DateTime2) { Value = month.AddMonths(1) }); var rows = new List(); await using var reader = await cmd.ExecuteReaderAsync(); while (await reader.ReadAsync()) { rows.Add(AuditLogHashChain.ReadRow(reader, withLink: true)); } return rows; } private static string Key(AuditChainRow r) => string.Join('|', r.EventId, r.OccurredAtUtc.Ticks, r.Actor ?? "", r.Action, r.Outcome, r.Category, r.Target ?? "", r.SourceNode ?? "", r.CorrelationId?.ToString() ?? "", r.DetailsJson ?? "", r.ChainSeq?.ToString() ?? "", r.RowHash is null ? "" : Convert.ToHexString(r.RowHash)); private async Task InsertAsync(IEnumerable events) { await using var ctx = CreateContext(); var repo = new AuditLogRepository(ctx); foreach (var e in events) { await repo.InsertIfNotExistsAsync(e); } } private async Task DrainAsync() { await using var ctx = CreateContext(); var chain = new AuditLogHashChain(ctx); for (var i = 0; i < 1000; i++) { var r = await chain.ChainPendingAsync(500); if (r.RowsChained == 0 && !r.LockBusy) { return; } } throw new InvalidOperationException("Chain did not drain."); } private ScadaBridgeDbContext CreateContext() => new(new DbContextOptionsBuilder().UseSqlServer(_fixture.ConnectionString).Options); private static AuditEvent NewEvent(DateTime occurredAtUtc, AuditChannel channel = AuditChannel.ApiOutbound) => ScadaBridgeAuditEventFactory.Create( channel: channel, kind: channel == AuditChannel.DbOutbound ? AuditKind.DbWrite : AuditKind.ApiCall, status: AuditStatus.Delivered, occurredAtUtc: occurredAtUtc, actor: "archive-test", target: "erp", sourceNode: "central-a", sourceSiteId: "archive-test", correlationId: Guid.NewGuid()); }