Joseph Doherty
288 lines
13 KiB
C#
288 lines
13 KiB
C#
using System.Data.Common;
|
|
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using NSubstitute;
|
|
using ZB.MOM.WW.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests;
|
|
|
|
/// <summary>
|
|
/// Flow Debugger outbound calls (plan 2026-09-06, Task 4 / D7): every
|
|
/// <see cref="InboundDatabaseHelper"/> call made by a central method script writes ONE
|
|
/// <c>DbOutbound.DbWrite</c> row through <see cref="ICentralAuditWriter"/> — best-effort,
|
|
/// never into the script — using the site's <c>AuditingDbCommand</c> vocabulary
|
|
/// (<c>{"op":"write","rowsAffected":N}</c> / <c>{"op":"read","rowsReturned":N}</c>), and
|
|
/// hands the same instance to the flow-trace sink first when one is present.
|
|
/// </summary>
|
|
public class InboundDatabaseHelperAuditTests
|
|
{
|
|
// One shared-cache in-memory database per gateway instance. Such a database lives only
|
|
// while a connection to it is open, so the gateway ROOTS its keep-alive connection for
|
|
// its own lifetime — an unrooted local would be GC-finalized mid-test under load (the
|
|
// executor test's Roslyn compile is long enough) and the table would silently vanish.
|
|
private sealed class SqliteGateway : IDatabaseGateway
|
|
{
|
|
private readonly string _cs = $"DataSource=file:dbaudit-{Guid.NewGuid():N}?mode=memory&cache=shared";
|
|
private readonly SqliteConnection _keepAlive;
|
|
|
|
public SqliteGateway()
|
|
{
|
|
_keepAlive = new SqliteConnection(_cs);
|
|
_keepAlive.Open();
|
|
using var cmd = _keepAlive.CreateCommand();
|
|
cmd.CommandText = "CREATE TABLE Machine(Code TEXT, SAPID TEXT); INSERT INTO Machine VALUES('Z28061A','131453');";
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
|
|
public async Task<DbConnection> GetConnectionAsync(string name, CancellationToken ct = default)
|
|
{ var c = new SqliteConnection(_cs); await c.OpenAsync(ct); return c; }
|
|
public Task<ExternalCallResult> CachedWriteAsync(string c, string s,
|
|
IReadOnlyDictionary<string, object?>? p = null, string? o = null, CancellationToken ct = default,
|
|
TrackedOperationId? t = null, Guid? e = null, string? src = null, Guid? pe = null)
|
|
=> throw new NotImplementedException();
|
|
}
|
|
|
|
private sealed class ThrowingGateway : IDatabaseGateway
|
|
{
|
|
public Task<DbConnection> GetConnectionAsync(string name, CancellationToken ct = default)
|
|
=> throw new InvalidOperationException("no such connection: " + name);
|
|
public Task<ExternalCallResult> CachedWriteAsync(string c, string s,
|
|
IReadOnlyDictionary<string, object?>? p = null, string? o = null, CancellationToken ct = default,
|
|
TrackedOperationId? t = null, Guid? e = null, string? src = null, Guid? pe = null)
|
|
=> throw new NotImplementedException();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Arch-review F8: a minimal <see cref="ILogger"/> double that records every
|
|
/// Warning-level message, mirroring the <c>CountingLogger<T></c> pattern in
|
|
/// <c>InboundScriptExecutorTests</c>.
|
|
/// </summary>
|
|
private sealed class RecordingLogger : ILogger
|
|
{
|
|
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = new();
|
|
|
|
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
|
public bool IsEnabled(LogLevel logLevel) => true;
|
|
|
|
public void Log<TState>(
|
|
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
|
|
Func<TState, Exception?, string> formatter)
|
|
{
|
|
Entries.Add((logLevel, formatter(state, exception), exception));
|
|
}
|
|
}
|
|
|
|
private static SqliteGateway SeededGateway() => new();
|
|
|
|
private readonly List<AuditEvent> _written = new();
|
|
private readonly ICentralAuditWriter _writer = Substitute.For<ICentralAuditWriter>();
|
|
private readonly INodeIdentityProvider _node = Substitute.For<INodeIdentityProvider>();
|
|
private readonly Guid _exec = Guid.NewGuid();
|
|
private readonly Guid _parent = Guid.NewGuid();
|
|
|
|
public InboundDatabaseHelperAuditTests()
|
|
{
|
|
_writer.WriteAsync(Arg.Do<AuditEvent>(e => _written.Add(e)), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
|
|
_node.NodeName.Returns("central-a");
|
|
}
|
|
|
|
private InboundDatabaseHelper Helper(IDatabaseGateway? gateway = null, Action<AuditEvent>? sink = null,
|
|
ICentralAuditWriter? writer = null) =>
|
|
new(gateway ?? SeededGateway(), CancellationToken.None, TimeSpan.Zero,
|
|
auditWriter: writer ?? _writer, nodeIdentity: _node,
|
|
executionId: _exec, parentExecutionId: _parent, methodName: "movein", traceSink: sink);
|
|
|
|
[Fact]
|
|
public async Task ExecuteAsync_WritesDbWriteDeliveredRow_WithRowsAffected()
|
|
{
|
|
var helper = Helper();
|
|
|
|
var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=@c WHERE SAPID=@s", new { c = "Z28061A", s = "131453" });
|
|
|
|
Assert.Equal(1, n);
|
|
var evt = Assert.Single(_written);
|
|
var row = evt.AsRow();
|
|
Assert.Equal(AuditChannel.DbOutbound, row.Channel);
|
|
Assert.Equal(AuditKind.DbWrite, row.Kind);
|
|
Assert.Equal(AuditStatus.Delivered, row.Status);
|
|
Assert.Equal("MesDb", evt.Target);
|
|
Assert.Equal("api:movein", evt.Actor);
|
|
Assert.Equal("central-a", evt.SourceNode);
|
|
Assert.Equal(_exec, row.ExecutionId);
|
|
Assert.Equal(_parent, row.ParentExecutionId);
|
|
Assert.Null(row.CorrelationId);
|
|
Assert.NotNull(row.DurationMs);
|
|
Assert.Null(row.ErrorMessage);
|
|
Assert.Equal("{\"op\":\"write\",\"rowsAffected\":1}", row.Extra);
|
|
Assert.Contains("\"sql\":\"UPDATE Machine SET Code=@c WHERE SAPID=@s\"", row.RequestSummary);
|
|
Assert.Contains("\"@s\":\"131453\"", row.RequestSummary);
|
|
Assert.Contains("\"@c\":\"Z28061A\"", row.RequestSummary);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task QueryAsync_WritesReadRow_WithRowsReturned()
|
|
{
|
|
var helper = Helper();
|
|
|
|
var rows = await helper.QueryAsync("MesDb", "SELECT Code, SAPID FROM Machine WHERE SAPID=@s", new { s = "131453" });
|
|
|
|
Assert.Single(rows);
|
|
var row = Assert.Single(_written).AsRow();
|
|
Assert.Equal(AuditKind.DbWrite, row.Kind);
|
|
Assert.Equal(AuditStatus.Delivered, row.Status);
|
|
Assert.Equal("{\"op\":\"read\",\"rowsReturned\":1}", row.Extra);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task QuerySingleAsync_WritesReadRow_ZeroRowsWhenNoMatch()
|
|
{
|
|
var helper = Helper();
|
|
|
|
var code = await helper.QuerySingleAsync<string>("MesDb", "SELECT Code FROM Machine WHERE SAPID=@s", new { s = "nope" });
|
|
|
|
Assert.Null(code);
|
|
var row = Assert.Single(_written).AsRow();
|
|
Assert.Equal("{\"op\":\"read\",\"rowsReturned\":0}", row.Extra);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ThrowingGateway_WritesFailedRow_AndRethrowsToScript()
|
|
{
|
|
var helper = Helper(new ThrowingGateway());
|
|
|
|
var ex = await Assert.ThrowsAsync<InvalidOperationException>(
|
|
() => helper.ExecuteAsync("Missing", "DELETE FROM Machine", null));
|
|
|
|
Assert.Contains("no such connection", ex.Message);
|
|
var evt = Assert.Single(_written);
|
|
var row = evt.AsRow();
|
|
Assert.Equal(AuditStatus.Failed, row.Status);
|
|
Assert.Equal("Missing", evt.Target);
|
|
Assert.Equal(ex.Message, row.ErrorMessage);
|
|
Assert.Equal("{\"op\":\"write\",\"rowsAffected\":0}", row.Extra);
|
|
Assert.Contains("\"sql\":\"DELETE FROM Machine\"", row.RequestSummary);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AuditWriterFaulting_DoesNotFailTheCall()
|
|
{
|
|
var syncThrow = Substitute.For<ICentralAuditWriter>();
|
|
syncThrow.WriteAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>())
|
|
.Returns<Task>(_ => throw new InvalidOperationException("audit down"));
|
|
var faultedTask = Substitute.For<ICentralAuditWriter>();
|
|
faultedTask.WriteAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>())
|
|
.Returns(Task.FromException(new InvalidOperationException("audit down")));
|
|
|
|
var n1 = await Helper(writer: syncThrow).ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
|
|
var n2 = await Helper(writer: faultedTask).ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
|
|
|
|
Assert.Equal(1, n1);
|
|
Assert.Equal(1, n2);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AuditWriterFaulting_LogsWarning_AndCallStillSucceeds()
|
|
{
|
|
// Arch-review F8: the best-effort continuation used to be `static _ => { }` — a
|
|
// faulting write vanished with no trace anywhere. It now logs a Warning (and runs
|
|
// on TaskScheduler.Default rather than riding whatever context completed the
|
|
// audit-writer task), while the call itself still succeeds either way.
|
|
var faultedTask = Substitute.For<ICentralAuditWriter>();
|
|
faultedTask.WriteAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>())
|
|
.Returns(Task.FromException(new InvalidOperationException("audit down")));
|
|
var logger = new RecordingLogger();
|
|
var helper = new InboundDatabaseHelper(
|
|
SeededGateway(), CancellationToken.None, TimeSpan.Zero,
|
|
auditWriter: faultedTask, nodeIdentity: _node,
|
|
executionId: _exec, parentExecutionId: _parent, methodName: "movein",
|
|
traceSink: null, logger: logger);
|
|
|
|
var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
|
|
|
|
Assert.Equal(1, n);
|
|
var entry = Assert.Single(logger.Entries);
|
|
Assert.Equal(LogLevel.Warning, entry.Level);
|
|
Assert.NotNull(entry.Exception);
|
|
Assert.Contains("MesDb", entry.Message);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task TraceSink_ReceivesTheSameInstance_BeforeTheWriter()
|
|
{
|
|
var order = new List<(string Who, AuditEvent Evt)>();
|
|
_writer.WriteAsync(Arg.Do<AuditEvent>(e => order.Add(("writer", e))), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
|
|
var helper = Helper(sink: e => order.Add(("sink", e)));
|
|
|
|
await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
|
|
|
|
Assert.Equal(new[] { "sink", "writer" }, order.Select(o => o.Who));
|
|
Assert.Same(order[0].Evt, order[1].Evt);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ThrowingSink_DoesNotFailTheCall_AndWriterStillGetsTheRow()
|
|
{
|
|
var helper = Helper(sink: _ => throw new InvalidOperationException("bad sink"));
|
|
|
|
var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
|
|
|
|
Assert.Equal(1, n);
|
|
Assert.Single(_written);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task NoWriterNoSink_EmitsNothing_AndStillWorks()
|
|
{
|
|
var helper = new InboundDatabaseHelper(SeededGateway(), CancellationToken.None);
|
|
|
|
var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
|
|
|
|
Assert.Equal(1, n);
|
|
Assert.Empty(_written);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Executor_WiresWriterNodeIdentityExecutionIdAndRouteSink_IntoTheHelper()
|
|
{
|
|
// End-to-end through InboundScriptExecutor: the DbOutbound row carries the inbound
|
|
// request's ExecutionId, "api:{method}" as actor, SourceNode from DI, and the same
|
|
// instance reaches the RouteHelper's trace sink (the Flow Debugger's Live path).
|
|
var services = new ServiceCollection();
|
|
services.AddSingleton<IDatabaseGateway>(SeededGateway());
|
|
services.AddSingleton(_writer);
|
|
services.AddSingleton(_node);
|
|
using var provider = services.BuildServiceProvider();
|
|
var executor = new InboundScriptExecutor(NullLogger<InboundScriptExecutor>.Instance, provider);
|
|
|
|
var sunk = new List<AuditEvent>();
|
|
var route = new RouteHelper(Substitute.For<IInstanceLocator>(), Substitute.For<IInstanceRouter>())
|
|
.WithTraceSink(sunk.Add);
|
|
var method = new ApiMethod(
|
|
"movein",
|
|
"return await Database.ExecuteAsync(\"MesDb\", \"UPDATE Machine SET Code=@c WHERE SAPID=@s\", new { c = \"Z28061A\", s = \"131453\" });")
|
|
{ Id = 1, TimeoutSeconds = 10, ReturnDefinition = null };
|
|
var inboundExecution = Guid.NewGuid();
|
|
|
|
var result = await executor.ExecuteAsync(
|
|
method, new Dictionary<string, object?>(), route, TimeSpan.FromSeconds(10),
|
|
parentExecutionId: inboundExecution, trace: true);
|
|
|
|
Assert.True(result.Success, result.ErrorMessage);
|
|
var evt = Assert.Single(_written);
|
|
var row = evt.AsRow();
|
|
Assert.Equal(AuditChannel.DbOutbound, row.Channel);
|
|
Assert.Equal(inboundExecution, row.ExecutionId);
|
|
Assert.Null(row.ParentExecutionId);
|
|
Assert.Equal("api:movein", evt.Actor);
|
|
Assert.Equal("central-a", evt.SourceNode);
|
|
Assert.Same(evt, Assert.Single(sunk));
|
|
}
|
|
}
|