357 lines
14 KiB
C#
357 lines
14 KiB
C#
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using ScadaLink.AuditLog.Site;
|
|
using ScadaLink.Commons.Entities.Audit;
|
|
using ScadaLink.Commons.Types.Enums;
|
|
|
|
namespace ScadaLink.AuditLog.Tests.Site;
|
|
|
|
/// <summary>
|
|
/// Bundle B (M2-T2) hot-path tests for <see cref="SqliteAuditWriter"/>. Exercise
|
|
/// the Channel-based enqueue, the background writer's batch INSERTs, duplicate-
|
|
/// EventId swallowing, ForwardState defaulting, and the
|
|
/// <see cref="SqliteAuditWriter.ReadPendingAsync"/> /
|
|
/// <see cref="SqliteAuditWriter.MarkForwardedAsync"/> support surface that
|
|
/// Bundle D's telemetry actor will call.
|
|
/// </summary>
|
|
public class SqliteAuditWriterWriteTests
|
|
{
|
|
private static (SqliteAuditWriter writer, string dataSource) CreateWriter(
|
|
string testName,
|
|
int? channelCapacity = null)
|
|
{
|
|
var dataSource = $"file:{testName}-{Guid.NewGuid():N}?mode=memory&cache=shared";
|
|
var opts = new SqliteAuditWriterOptions { DatabasePath = dataSource };
|
|
if (channelCapacity is int cap)
|
|
{
|
|
opts.ChannelCapacity = cap;
|
|
}
|
|
|
|
var writer = new SqliteAuditWriter(
|
|
Options.Create(opts),
|
|
NullLogger<SqliteAuditWriter>.Instance,
|
|
connectionStringOverride: $"Data Source={dataSource};Cache=Shared");
|
|
return (writer, dataSource);
|
|
}
|
|
|
|
private static SqliteConnection OpenVerifierConnection(string dataSource)
|
|
{
|
|
var connection = new SqliteConnection($"Data Source={dataSource};Cache=Shared");
|
|
connection.Open();
|
|
return connection;
|
|
}
|
|
|
|
private static AuditEvent NewEvent(Guid? id = null, DateTime? occurredAtUtc = null)
|
|
{
|
|
return new AuditEvent
|
|
{
|
|
EventId = id ?? Guid.NewGuid(),
|
|
OccurredAtUtc = occurredAtUtc ?? DateTime.UtcNow,
|
|
Channel = AuditChannel.ApiOutbound,
|
|
Kind = AuditKind.ApiCall,
|
|
Status = AuditStatus.Delivered,
|
|
PayloadTruncated = false,
|
|
};
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WriteAsync_FreshEvent_PersistsWithForwardStatePending()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(WriteAsync_FreshEvent_PersistsWithForwardStatePending));
|
|
await using var _ = writer;
|
|
|
|
var evt = NewEvent();
|
|
await writer.WriteAsync(evt);
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT ForwardState FROM AuditLog WHERE EventId = $id;";
|
|
cmd.Parameters.AddWithValue("$id", evt.EventId.ToString());
|
|
var actual = cmd.ExecuteScalar() as string;
|
|
|
|
Assert.Equal(AuditForwardState.Pending.ToString(), actual);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WriteAsync_Concurrent_1000Calls_All_Persist_NoExceptions()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(WriteAsync_Concurrent_1000Calls_All_Persist_NoExceptions));
|
|
await using var _ = writer;
|
|
|
|
var events = Enumerable.Range(0, 1000).Select(_ => NewEvent()).ToList();
|
|
|
|
await Parallel.ForEachAsync(events, new ParallelOptions { MaxDegreeOfParallelism = 16 },
|
|
async (evt, ct) => await writer.WriteAsync(evt, ct));
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT COUNT(*) FROM AuditLog;";
|
|
var count = Convert.ToInt64(cmd.ExecuteScalar());
|
|
|
|
Assert.Equal(1000, count);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WriteAsync_DuplicateEventId_FirstWriteWins_NoException()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(WriteAsync_DuplicateEventId_FirstWriteWins_NoException));
|
|
await using var _ = writer;
|
|
|
|
var sharedId = Guid.NewGuid();
|
|
var first = NewEvent(sharedId) with { Target = "first" };
|
|
var second = NewEvent(sharedId) with { Target = "second" };
|
|
|
|
await writer.WriteAsync(first);
|
|
await writer.WriteAsync(second);
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var countCmd = connection.CreateCommand();
|
|
countCmd.CommandText = "SELECT COUNT(*) FROM AuditLog WHERE EventId = $id;";
|
|
countCmd.Parameters.AddWithValue("$id", sharedId.ToString());
|
|
var count = Convert.ToInt64(countCmd.ExecuteScalar());
|
|
|
|
Assert.Equal(1, count);
|
|
|
|
using var targetCmd = connection.CreateCommand();
|
|
targetCmd.CommandText = "SELECT Target FROM AuditLog WHERE EventId = $id;";
|
|
targetCmd.Parameters.AddWithValue("$id", sharedId.ToString());
|
|
Assert.Equal("first", targetCmd.ExecuteScalar() as string);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task WriteAsync_ForcesForwardStatePending_IfNull()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(WriteAsync_ForcesForwardStatePending_IfNull));
|
|
await using var _ = writer;
|
|
|
|
var evt = NewEvent() with { ForwardState = null };
|
|
await writer.WriteAsync(evt);
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT ForwardState FROM AuditLog WHERE EventId = $id;";
|
|
cmd.Parameters.AddWithValue("$id", evt.EventId.ToString());
|
|
|
|
Assert.Equal(AuditForwardState.Pending.ToString(), cmd.ExecuteScalar() as string);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadPendingAsync_Returns_OldestFirst_LimitedToN()
|
|
{
|
|
var (writer, _) = CreateWriter(nameof(ReadPendingAsync_Returns_OldestFirst_LimitedToN));
|
|
await using var _writer = writer;
|
|
|
|
var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
|
|
var evts = new[]
|
|
{
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(5)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(1)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(3)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(2)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(4)),
|
|
};
|
|
|
|
foreach (var e in evts)
|
|
{
|
|
await writer.WriteAsync(e);
|
|
}
|
|
|
|
var rows = await writer.ReadPendingAsync(limit: 3);
|
|
|
|
Assert.Equal(3, rows.Count);
|
|
Assert.Equal(baseTime.AddSeconds(1), rows[0].OccurredAtUtc);
|
|
Assert.Equal(baseTime.AddSeconds(2), rows[1].OccurredAtUtc);
|
|
Assert.Equal(baseTime.AddSeconds(3), rows[2].OccurredAtUtc);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MarkForwardedAsync_FlipsRowsToForwarded()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(MarkForwardedAsync_FlipsRowsToForwarded));
|
|
await using var _ = writer;
|
|
|
|
var ids = new[] { Guid.NewGuid(), Guid.NewGuid(), Guid.NewGuid() };
|
|
foreach (var id in ids)
|
|
{
|
|
await writer.WriteAsync(NewEvent(id));
|
|
}
|
|
|
|
await writer.MarkForwardedAsync(ids);
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT ForwardState, COUNT(*) FROM AuditLog GROUP BY ForwardState;";
|
|
using var reader = cmd.ExecuteReader();
|
|
var byState = new Dictionary<string, long>();
|
|
while (reader.Read())
|
|
{
|
|
byState[reader.GetString(0)] = reader.GetInt64(1);
|
|
}
|
|
|
|
Assert.Equal(3, byState[AuditForwardState.Forwarded.ToString()]);
|
|
Assert.False(byState.ContainsKey(AuditForwardState.Pending.ToString()));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MarkForwardedAsync_NonExistentId_NoThrow()
|
|
{
|
|
var (writer, _) = CreateWriter(nameof(MarkForwardedAsync_NonExistentId_NoThrow));
|
|
await using var _writer = writer;
|
|
|
|
var phantomIds = new[] { Guid.NewGuid(), Guid.NewGuid() };
|
|
|
|
await writer.MarkForwardedAsync(phantomIds);
|
|
// No assertion needed: the call must complete without throwing.
|
|
}
|
|
|
|
// ----- M6 reconciliation pull surface ----- //
|
|
|
|
[Fact]
|
|
public async Task ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN));
|
|
await using var _ = writer;
|
|
|
|
var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
|
|
var evts = new[]
|
|
{
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(5)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(1)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(3)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(2)),
|
|
NewEvent(occurredAtUtc: baseTime.AddSeconds(4)),
|
|
};
|
|
foreach (var e in evts) await writer.WriteAsync(e);
|
|
|
|
// Flip half to Forwarded — they must still surface in the reconciliation pull
|
|
// because central hasn't confirmed they were ingested yet.
|
|
await writer.MarkForwardedAsync(new[] { evts[0].EventId, evts[2].EventId });
|
|
|
|
var rows = await writer.ReadPendingSinceAsync(sinceUtc: DateTime.MinValue, batchSize: 3);
|
|
|
|
Assert.Equal(3, rows.Count);
|
|
Assert.Equal(baseTime.AddSeconds(1), rows[0].OccurredAtUtc);
|
|
Assert.Equal(baseTime.AddSeconds(2), rows[1].OccurredAtUtc);
|
|
Assert.Equal(baseTime.AddSeconds(3), rows[2].OccurredAtUtc);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadPendingSinceAsync_ExcludesRowsOlderThanSinceUtc()
|
|
{
|
|
var (writer, _) = CreateWriter(nameof(ReadPendingSinceAsync_ExcludesRowsOlderThanSinceUtc));
|
|
await using var _w = writer;
|
|
|
|
var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
|
|
var old = NewEvent(occurredAtUtc: baseTime.AddSeconds(-30));
|
|
var newer1 = NewEvent(occurredAtUtc: baseTime.AddSeconds(10));
|
|
var newer2 = NewEvent(occurredAtUtc: baseTime.AddSeconds(20));
|
|
|
|
await writer.WriteAsync(old);
|
|
await writer.WriteAsync(newer1);
|
|
await writer.WriteAsync(newer2);
|
|
|
|
var rows = await writer.ReadPendingSinceAsync(sinceUtc: baseTime, batchSize: 10);
|
|
|
|
Assert.Equal(2, rows.Count);
|
|
Assert.Contains(rows, r => r.EventId == newer1.EventId);
|
|
Assert.Contains(rows, r => r.EventId == newer2.EventId);
|
|
Assert.DoesNotContain(rows, r => r.EventId == old.EventId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadPendingSinceAsync_ExcludesReconciledRows()
|
|
{
|
|
var (writer, _) = CreateWriter(nameof(ReadPendingSinceAsync_ExcludesReconciledRows));
|
|
await using var _w = writer;
|
|
|
|
var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
|
|
var pending = NewEvent(occurredAtUtc: baseTime);
|
|
var reconciled = NewEvent(occurredAtUtc: baseTime.AddSeconds(1));
|
|
|
|
await writer.WriteAsync(pending);
|
|
await writer.WriteAsync(reconciled);
|
|
await writer.MarkReconciledAsync(new[] { reconciled.EventId });
|
|
|
|
var rows = await writer.ReadPendingSinceAsync(sinceUtc: DateTime.MinValue, batchSize: 10);
|
|
|
|
Assert.Single(rows);
|
|
Assert.Equal(pending.EventId, rows[0].EventId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadPendingSinceAsync_InvalidBatchSize_Throws()
|
|
{
|
|
var (writer, _) = CreateWriter(nameof(ReadPendingSinceAsync_InvalidBatchSize_Throws));
|
|
await using var _w = writer;
|
|
|
|
await Assert.ThrowsAsync<ArgumentOutOfRangeException>(
|
|
() => writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 0));
|
|
await Assert.ThrowsAsync<ArgumentOutOfRangeException>(
|
|
() => writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: -3));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MarkReconciledAsync_FlipsPendingAndForwarded_To_Reconciled()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(MarkReconciledAsync_FlipsPendingAndForwarded_To_Reconciled));
|
|
await using var _ = writer;
|
|
|
|
var a = NewEvent();
|
|
var b = NewEvent();
|
|
var c = NewEvent();
|
|
await writer.WriteAsync(a);
|
|
await writer.WriteAsync(b);
|
|
await writer.WriteAsync(c);
|
|
|
|
// b is currently Forwarded; a and c are Pending.
|
|
await writer.MarkForwardedAsync(new[] { b.EventId });
|
|
|
|
await writer.MarkReconciledAsync(new[] { a.EventId, b.EventId, c.EventId });
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT ForwardState, COUNT(*) FROM AuditLog GROUP BY ForwardState;";
|
|
using var reader = cmd.ExecuteReader();
|
|
var byState = new Dictionary<string, long>();
|
|
while (reader.Read())
|
|
{
|
|
byState[reader.GetString(0)] = reader.GetInt64(1);
|
|
}
|
|
|
|
Assert.Equal(3, byState[AuditForwardState.Reconciled.ToString()]);
|
|
Assert.False(byState.ContainsKey(AuditForwardState.Pending.ToString()));
|
|
Assert.False(byState.ContainsKey(AuditForwardState.Forwarded.ToString()));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MarkReconciledAsync_Idempotent_LeavesAlreadyReconciledRowsUntouched()
|
|
{
|
|
var (writer, dataSource) = CreateWriter(nameof(MarkReconciledAsync_Idempotent_LeavesAlreadyReconciledRowsUntouched));
|
|
await using var _ = writer;
|
|
|
|
var a = NewEvent();
|
|
await writer.WriteAsync(a);
|
|
await writer.MarkReconciledAsync(new[] { a.EventId });
|
|
// Re-call must not throw and must leave the single row Reconciled.
|
|
await writer.MarkReconciledAsync(new[] { a.EventId });
|
|
|
|
using var connection = OpenVerifierConnection(dataSource);
|
|
using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT ForwardState FROM AuditLog WHERE EventId = $id;";
|
|
cmd.Parameters.AddWithValue("$id", a.EventId.ToString());
|
|
|
|
Assert.Equal(AuditForwardState.Reconciled.ToString(), cmd.ExecuteScalar() as string);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MarkReconciledAsync_NonExistentId_NoThrow()
|
|
{
|
|
var (writer, _) = CreateWriter(nameof(MarkReconciledAsync_NonExistentId_NoThrow));
|
|
await using var _w = writer;
|
|
|
|
await writer.MarkReconciledAsync(new[] { Guid.NewGuid(), Guid.NewGuid() });
|
|
// Completes without throwing.
|
|
}
|
|
}
|