ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/AuditExportParquetTests.cs

using Parquet;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;

/// <summary>
/// <see cref="AuditExportParquet"/> (ScadaBridge#42): the Parquet export round-trips every
/// one of the 24 export columns, splits into row groups, and never writes synchronously to
/// the output (Kestrel rejects synchronous response writes).
/// </summary>
public class AuditExportParquetTests
{
    /// <summary>Rows written and read back are equal, column for column, across row groups.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task RoundTrip_EveryColumn_AcrossRowGroups()
    {
        var rows = Enumerable.Range(0, AuditExportParquet.RowGroupRows + 7).Select(Row).ToList();
        var output = new AsyncOnlyStream();

        var written = await AuditExportParquet.WriteAsync(Pages(rows, 1000), output);

        Assert.Equal(rows.Count, written);
        await using var reader = await ParquetReader.CreateAsync(new MemoryStream(output.ToArray()));
        Assert.Equal(2, reader.RowGroupCount);
        var back = new List<AuditExportRow>();
        var fields = reader.Schema.GetDataFields();
        for (var g = 0; g < reader.RowGroupCount; g++)
        {
            using var group = reader.OpenRowGroupReader(g);
            var n = (int)group.RowCount;
            async Task<string?[]> S(int i) { var a = new string?[n]; await group.ReadAsync(fields[i], a!); return a; }
            async Task<T?[]> V<T>(int i) where T : struct { var a = new T?[n]; await group.ReadAsync<T>(fields[i], a); return a; }
            var id = await S(0); var occurred = await V<DateTime>(1); var ingested = await V<DateTime>(2);
            var channel = await S(3); var kind = await S(4); var corr = await S(5); var exec = await S(6); var parent = await S(7);
            var site = await S(8); var node = await S(9); var inst = await S(10); var script = await S(11); var actor = await S(12);
            var target = await S(13); var status = await S(14); var http = await V<int>(15); var dur = await V<int>(16);
            var errMsg = await S(17); var errDetail = await S(18); var req = await S(19); var resp = await S(20);
            var trunc = await V<bool>(21); var extra = await S(22); var fwd = await S(23);
            for (var i = 0; i < n; i++)
            {
                back.Add(new AuditExportRow
                {
                    EventId = Guid.Parse(id[i]!),
                    OccurredAtUtc = occurred[i]!.Value,
                    IngestedAtUtc = ingested[i],
                    Channel = Enum.Parse<AuditChannel>(channel[i]!),
                    Kind = Enum.Parse<AuditKind>(kind[i]!),
                    CorrelationId = corr[i] is { } c ? Guid.Parse(c) : null,
                    ExecutionId = exec[i] is { } e ? Guid.Parse(e) : null,
                    ParentExecutionId = parent[i] is { } p ? Guid.Parse(p) : null,
                    SourceSiteId = site[i],
                    SourceNode = node[i],
                    SourceInstanceId = inst[i],
                    SourceScript = script[i],
                    Actor = actor[i],
                    Target = target[i],
                    Status = Enum.Parse<AuditStatus>(status[i]!),
                    HttpStatus = http[i],
                    DurationMs = dur[i],
                    ErrorMessage = errMsg[i],
                    ErrorDetail = errDetail[i],
                    RequestSummary = req[i],
                    ResponseSummary = resp[i],
                    PayloadTruncated = trunc[i]!.Value,
                    Extra = extra[i],
                    ForwardState = fwd[i] is { } f ? Enum.Parse<AuditForwardState>(f) : null,
                });
            }
        }

        Assert.Equal(rows, back);
    }

    /// <summary>An export with no rows is still a valid (empty) Parquet file.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task NoRows_IsAValidEmptyFile()
    {
        var output = new AsyncOnlyStream();
        Assert.Equal(0, await AuditExportParquet.WriteAsync(Pages([], 10), output));
        await using var reader = await ParquetReader.CreateAsync(new MemoryStream(output.ToArray()));
        Assert.Equal(0, reader.RowGroupCount);
        Assert.Equal(24, reader.Schema.GetDataFields().Length);
    }

    private static AuditExportRow Row(int i) => new()
    {
        EventId = Guid.NewGuid(),
        // Microsecond precision: the export timestamp column is TIMESTAMP(MICROS).
        OccurredAtUtc = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc).AddTicks(i * 10L),
        IngestedAtUtc = i % 3 == 0 ? null : new DateTime(2026, 5, 20, 12, 0, 1, DateTimeKind.Utc),
        Channel = (AuditChannel)(i % Enum.GetValues<AuditChannel>().Length),
        Kind = AuditKind.ApiCall,
        CorrelationId = i % 2 == 0 ? Guid.NewGuid() : null,
        ExecutionId = Guid.NewGuid(),
        ParentExecutionId = i % 5 == 0 ? Guid.NewGuid() : null,
        SourceSiteId = i % 4 == 0 ? null : "plant-a",
        SourceNode = "node-a",
        SourceInstanceId = "Plant1.Boiler",
        SourceScript = i % 2 == 0 ? "OnTick" : null,
        Actor = i % 7 == 0 ? string.Empty : "api-key-1",
        Target = "erp",
        Status = AuditStatus.Delivered,
        HttpStatus = i % 2 == 0 ? 200 : null,
        DurationMs = i,
        ErrorMessage = i % 9 == 0 ? "boom, \"quoted\"\nline" : null,
        ErrorDetail = null,
        RequestSummary = "{\"ü\":\"日本\"}",
        ResponseSummary = null,
        PayloadTruncated = i % 2 == 1,
        Extra = null,
        ForwardState = i % 11 == 0 ? AuditForwardState.Forwarded : null,
    };

    private static async IAsyncEnumerable<IReadOnlyList<AuditExportRow>> Pages(List<AuditExportRow> rows, int size)
    {
        for (var i = 0; i < rows.Count; i += size)
        {
            await Task.Yield();
            yield return rows.Skip(i).Take(size).ToList();
        }
    }

    /// <summary>A response-body stand-in that fails any synchronous write, like Kestrel.</summary>
    private sealed class AsyncOnlyStream : Stream
    {
        private readonly MemoryStream _inner = new();

        public byte[] ToArray() => _inner.ToArray();

        public override bool CanRead => false;
        public override bool CanSeek => false;
        public override bool CanWrite => true;
        public override long Length => throw new NotSupportedException();
        public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
        public override void Flush() => throw new InvalidOperationException("Synchronous operations are disallowed.");
        public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
        public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
        public override void SetLength(long value) => throw new NotSupportedException();
        public override void Write(byte[] buffer, int offset, int count) => throw new InvalidOperationException("Synchronous operations are disallowed.");
        public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
        {
            _inner.Write(buffer.Span);
            return ValueTask.CompletedTask;
        }
    }
}