using Parquet;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Maintenance;
///
/// Storage-free tests for the AuditLog archive Parquet layout (ScadaBridge#42): rows and
/// tombstones round-trip exactly (tick-precise timestamps, NULL vs empty strings, unicode,
/// the chain columns), the archived rows re-verify against the archived chain head, and an
/// edit made inside the Parquet file breaks that verification.
///
public class AuditArchiveParquetTests
{
private static readonly DateTime Month = new(2025, 10, 1, 0, 0, 0, DateTimeKind.Utc);
/// Rows written to the archive layout read back equal, column for column, across row groups.
/// A task that represents the asynchronous operation.
[Fact]
public async Task Rows_RoundTrip_Exactly_AcrossRowGroups()
{
var (rows, _) = BuildChain(2_503);
rows.Add(Row(9_999) with { ChainSeq = null, RowHash = null }); // an unchained row
var bytes = await WriteAsync(rows, rowGroupSize: 1_000);
var back = await ReadAsync(bytes);
Assert.Equal(rows.Select(Key), back.Select(Key));
await using var reader = await ParquetReader.CreateAsync(new MemoryStream(bytes));
Assert.Equal(3, reader.RowGroupCount);
Assert.Equal("1", reader.CustomMetadata[AuditArchiveParquet.FormatVersionKey]);
Assert.Equal("2025-10", reader.CustomMetadata[AuditArchiveParquet.PartitionKey]);
}
/// Tombstones round-trip exactly.
/// A task that represents the asynchronous operation.
[Fact]
public async Task Tombstones_RoundTrip_Exactly()
{
var tombstones = Enumerable.Range(1, 5).Select(i => new AuditArchiveTombstone(
Month, i * 3, Enumerable.Repeat((byte)i, 32).ToArray(), Guid.NewGuid(),
new DateTime(2026, 1, 2, 3, 4, 5, DateTimeKind.Utc).AddMilliseconds(i))).ToList();
var stream = new MemoryStream();
await using (var writer = await AuditArchiveParquet.CreateTombstoneWriterAsync(stream, "2025-10", 2))
{
foreach (var t in tombstones)
{
await writer.AddAsync(t);
}
}
var back = new List();
await foreach (var t in AuditArchiveParquet.ReadTombstonesAsync(new MemoryStream(stream.ToArray())))
{
back.Add(t);
}
Assert.Equal(
tombstones.Select(t => (t.ChainMonth, t.ChainSeq, Convert.ToHexString(t.RowHash), t.EventId, t.PurgedAtUtc)),
back.Select(t => (t.ChainMonth, t.ChainSeq, Convert.ToHexString(t.RowHash), t.EventId, t.PurgedAtUtc)));
}
///
/// The rows read back from an archive re-verify against the archived head — the same
/// check audit archive verify makes long after the database dropped them.
///
/// A task that represents the asynchronous operation.
[Fact]
public async Task ArchivedRows_ReverifyAgainstTheArchivedHead()
{
var (rows, head) = BuildChain(50);
var verifier = new AuditChainVerifier(Month, head, new Dictionary());
foreach (var row in await ReadAsync(await WriteAsync(rows, 16)))
{
verifier.Accept(row);
}
var result = verifier.Complete(0);
Assert.Equal(AuditChainStatus.Intact, result.Status);
Assert.Equal(50, result.RowsVerified);
}
///
/// Positive control for the test above: a row edited inside the archive (re-written
/// file, as a forger would) fails re-verification at that row.
///
/// A task that represents the asynchronous operation.
[Fact]
public async Task RowEditedInsideTheArchive_FailsReverification()
{
var (rows, head) = BuildChain(50);
rows[17] = rows[17] with { DetailsJson = "{\"kind\":\"ApiCall\",\"forged\":true}" };
var verifier = new AuditChainVerifier(Month, head, new Dictionary());
foreach (var row in await ReadAsync(await WriteAsync(rows, 16)))
{
verifier.Accept(row);
}
var result = verifier.Complete(0);
Assert.Equal(AuditChainStatus.Broken, result.Status);
Assert.Equal(AuditChainBreakKind.HashMismatch, result.FirstBreak!.Kind);
Assert.Equal(18, result.FirstBreak.ChainSeq);
}
internal static (List Rows, AuditChainHeadState Head) BuildChain(int count)
{
var rows = new List();
var previous = AuditChainHasher.GenesisHash.ToArray();
for (var i = 1; i <= count; i++)
{
var row = Row(i);
var hash = AuditChainHasher.ComputeRowHash(previous, Month, i, row);
rows.Add(row with { ChainSeq = i, RowHash = hash });
previous = hash;
}
return (rows, new AuditChainHeadState(count, previous, null, null));
}
private static AuditChainRow Row(int i) => new(
Guid.NewGuid(),
// Odd ticks on purpose: the archive must keep datetime2(7) precision the hash covers.
Month.AddMinutes(i).AddTicks(1_234_567 + i),
i % 5 == 0 ? null : i % 7 == 0 ? string.Empty : "api-key-" + i,
"ApiOutbound.ApiCall",
i % 3 == 0 ? "Failure" : "Success",
"ApiOutbound",
i % 4 == 0 ? null : "erp-ü-日本",
i % 6 == 0 ? null : "central-a",
i % 2 == 0 ? Guid.NewGuid() : null,
i % 8 == 0 ? null : $"{{\"kind\":\"ApiCall\",\"status\":\"Delivered\",\"n\":{i}}}");
private static string Key(AuditChainRow r) => string.Join('|',
r.EventId, r.OccurredAtUtc.Ticks, r.OccurredAtUtc.Kind, 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 static async Task WriteAsync(IEnumerable rows, int rowGroupSize)
{
var stream = new MemoryStream();
await using (var writer = await AuditArchiveParquet.CreateDataWriterAsync(stream, "2025-10", rowGroupSize))
{
foreach (var row in rows)
{
await writer.AddAsync(row);
}
}
return stream.ToArray();
}
private static async Task> ReadAsync(byte[] bytes)
{
var rows = new List();
await foreach (var row in AuditArchiveParquet.ReadRowsAsync(new MemoryStream(bytes)))
{
rows.Add(row);
}
return rows;
}
}