using System.Data;
using System.Data.Common;
using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.ExternalSystems;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
namespace ZB.MOM.WW.ScadaBridge.ExternalSystemGateway.Tests;
///
/// WP-9: Tests for Database access — connection resolution, cached writes.
///
public class DatabaseGatewayTests
{
private readonly IExternalSystemRepository _repository = Substitute.For();
///
/// Configures the repository substitute for the name-keyed connection-resolution
/// path used by DatabaseGateway (ExternalSystemGateway-011). A null
/// connection models a "not found" — the substitute returns null by default,
/// so no stub is needed for the absent entity.
///
private void StubConnection(DatabaseConnectionDefinition? connection)
{
if (connection != null)
{
_repository.GetDatabaseConnectionByNameAsync(connection.Name, Arg.Any())
.Returns(connection);
}
}
[Fact]
public async Task GetConnection_NotFound_Throws()
{
StubConnection(connection: null);
var gateway = new DatabaseGateway(
_repository,
NullLogger.Instance);
await Assert.ThrowsAsync(
() => gateway.GetConnectionAsync("nonexistent"));
}
[Fact]
public async Task CachedWrite_NoStoreAndForward_Throws()
{
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test") { Id = 1 };
StubConnection(conn);
var gateway = new DatabaseGateway(
_repository,
NullLogger.Instance,
storeAndForward: null);
await Assert.ThrowsAsync(
() => gateway.CachedWriteAsync("testDb", "INSERT INTO t VALUES (1)"));
}
[Fact]
public async Task CachedWrite_ConnectionNotFound_Throws()
{
StubConnection(connection: null);
var gateway = new DatabaseGateway(
_repository,
NullLogger.Instance);
await Assert.ThrowsAsync(
() => gateway.CachedWriteAsync("nonexistent", "INSERT INTO t VALUES (1)"));
}
// ── ExternalSystemGateway-014: CachedWrite happy-path buffering ──
[Fact]
public async Task CachedWrite_BuffersTheWriteWithConnectionRetrySettings()
{
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test")
{
Id = 1,
MaxRetries = 5,
RetryDelay = TimeSpan.FromSeconds(12),
};
StubConnection(conn);
var dbName = $"EsgCachedWrite_{Guid.NewGuid():N}";
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
using var keepAlive = new Microsoft.Data.Sqlite.SqliteConnection(connStr);
keepAlive.Open();
var storage = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardStorage(
connStr, NullLogger.Instance);
await storage.InitializeAsync();
var sfOptions = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardOptions
{
DefaultMaxRetries = 99,
DefaultRetryInterval = TimeSpan.FromMinutes(10),
RetryTimerInterval = TimeSpan.FromMinutes(10),
};
var sf = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardService(
storage, sfOptions, NullLogger.Instance);
// M2.3 (#7): CachedWriteAsync now attempts the write immediately and
// only buffers on a TRANSIENT failure. The stub forces a transient
// outcome so this test exercises the buffering path deterministically
// without a real SQL Server.
var gateway = new ExecuteStubGateway(
_repository,
sf,
onExecute: () => throw new TransientDatabaseException("deadlock", errorNumber: 1205));
// Audit Log #23 (ExecutionId Task 4): a known execution id / source
// script so the gateway -> EnqueueAsync hop can be asserted below.
var executionId = Guid.NewGuid();
const string sourceScript = "ScriptActor:WriteAudit";
await gateway.CachedWriteAsync("testDb", "INSERT INTO t VALUES (@v)",
new Dictionary { ["v"] = 1 },
executionId: executionId, sourceScript: sourceScript);
var depth = await storage.GetBufferDepthByCategoryAsync();
Assert.Equal(1, depth[ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.StoreAndForwardCategory.CachedDbWrite]);
var buffered = ReadBufferedRetrySettings(connStr);
Assert.Equal(5, buffered.MaxRetries);
Assert.Equal((long)TimeSpan.FromSeconds(12).TotalMilliseconds, buffered.RetryIntervalMs);
// ExecutionId Task 4: the gateway must forward executionId / sourceScript
// into EnqueueAsync, and the S&F layer must persist them on the
// sf_messages row so the retry loop can stamp the right provenance.
Assert.Equal(executionId, buffered.ExecutionId);
Assert.Equal(sourceScript, buffered.SourceScript);
}
[Fact]
public async Task CachedWrite_ZeroMaxRetriesIsTreatedAsUnsetNotRetryForever()
{
// ExternalSystemGateway-015: a stored MaxRetries of 0 is interpreted by the
// Store-and-Forward retry sweep as "no limit" (retry forever). The entity's
// non-nullable int default is also 0, so the gateway must treat the
// connection's MaxRetries == 0 as "unset" and pass null — the bounded S&F
// default must apply, never 0.
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test")
{
Id = 1,
MaxRetries = 0,
RetryDelay = TimeSpan.FromSeconds(3),
};
StubConnection(conn);
var dbName = $"EsgCachedWriteZero_{Guid.NewGuid():N}";
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
using var keepAlive = new Microsoft.Data.Sqlite.SqliteConnection(connStr);
keepAlive.Open();
var storage = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardStorage(
connStr, NullLogger.Instance);
await storage.InitializeAsync();
var sfOptions = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardOptions
{
DefaultMaxRetries = 99,
DefaultRetryInterval = TimeSpan.FromMinutes(10),
RetryTimerInterval = TimeSpan.FromMinutes(10),
};
var sf = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardService(
storage, sfOptions, NullLogger.Instance);
// M2.3 (#7): force a transient outcome so the write reaches S&F.
var gateway = new ExecuteStubGateway(
_repository,
sf,
onExecute: () => throw new TransientDatabaseException("deadlock", errorNumber: 1205));
await gateway.CachedWriteAsync("testDb", "INSERT INTO t VALUES (1)");
var (maxRetries, _, _, _) = ReadBufferedRetrySettings(connStr);
// Must be the bounded S&F default, never 0 — a stored 0 would mean retry-forever.
Assert.Equal(99, maxRetries);
Assert.NotEqual(0, maxRetries);
}
// ── M2.3 (#7): transient-vs-permanent SQL classification on the immediate
// cached-write attempt + the buffered retry path ──
///
/// Builds a real, initialised in-memory store-and-forward service plus a
/// keep-alive connection (the SQLite shared-cache DB lives only while a
/// connection is open). The caller disposes .
///
private static (ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardService Sf, string ConnStr, Microsoft.Data.Sqlite.SqliteConnection KeepAlive)
NewStoreAndForward()
{
var dbName = $"EsgCachedWriteClassify_{Guid.NewGuid():N}";
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
var keepAlive = new Microsoft.Data.Sqlite.SqliteConnection(connStr);
keepAlive.Open();
var storage = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardStorage(
connStr, NullLogger.Instance);
storage.InitializeAsync().GetAwaiter().GetResult();
var sfOptions = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardOptions
{
DefaultMaxRetries = 99,
DefaultRetryInterval = TimeSpan.FromMinutes(10),
RetryTimerInterval = TimeSpan.FromMinutes(10),
};
var sf = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardService(
storage, sfOptions, NullLogger.Instance);
return (sf, connStr, keepAlive);
}
[Fact]
public async Task CachedWrite_PermanentSqlError_ReturnsFailedSynchronously_NotBuffered()
{
// A constraint/syntax/permission failure on the IMMEDIATE attempt must
// be returned to the script as Failed and must NOT be buffered — mirrors
// ExternalSystemClient.CachedCallAsync's PermanentExternalSystemException
// path.
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test") { Id = 1 };
StubConnection(conn);
var (sf, connStr, keepAlive) = NewStoreAndForward();
using var _ = keepAlive;
var gateway = new ExecuteStubGateway(
_repository,
sf,
onExecute: () => throw new PermanentDatabaseException(
"Violation of PRIMARY KEY constraint", errorNumber: 2627));
var result = await gateway.CachedWriteAsync("testDb", "INSERT INTO t VALUES (1)");
Assert.False(result.Success);
Assert.False(result.WasBuffered);
Assert.NotNull(result.ErrorMessage);
// Nothing buffered — the permanent failure short-circuited S&F.
Assert.Equal(0, ReadBufferDepth(connStr));
}
[Fact]
public async Task CachedWrite_TransientSqlError_BuffersToStoreAndForward()
{
// A deadlock / timeout on the IMMEDIATE attempt is transient — the write
// is handed to S&F (WasBuffered=true), not returned as Failed.
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test")
{
Id = 1,
MaxRetries = 5,
RetryDelay = TimeSpan.FromSeconds(12),
};
StubConnection(conn);
var (sf, connStr, keepAlive) = NewStoreAndForward();
using var _ = keepAlive;
var gateway = new ExecuteStubGateway(
_repository,
sf,
onExecute: () => throw new TransientDatabaseException(
"Transaction was deadlocked", errorNumber: 1205));
var result = await gateway.CachedWriteAsync(
"testDb", "UPDATE t SET v = 1", new Dictionary { ["x"] = 1 });
Assert.True(result.Success); // accepted for delivery
Assert.True(result.WasBuffered); // handed to S&F, not synchronously failed
Assert.Null(result.ErrorMessage);
Assert.Equal(1, ReadBufferDepth(connStr));
}
[Fact]
public async Task CachedWrite_ImmediateSuccess_NotBuffered_ReturnsDelivered()
{
// A write that succeeds immediately is done — it must NOT be buffered,
// and the result reports success (WasBuffered=false), mirroring the API
// path's immediate-success behaviour.
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test") { Id = 1 };
StubConnection(conn);
var (sf, connStr, keepAlive) = NewStoreAndForward();
using var _ = keepAlive;
var gateway = new ExecuteStubGateway(_repository, sf, onExecute: () => { /* succeeds */ });
var result = await gateway.CachedWriteAsync("testDb", "INSERT INTO t VALUES (1)");
Assert.True(result.Success);
Assert.False(result.WasBuffered);
Assert.Null(result.ErrorMessage);
Assert.Equal(0, ReadBufferDepth(connStr));
}
[Fact]
public async Task DeliverBuffered_TransientSqlError_RethrowsSoEngineRetries()
{
// On the retry path a transient failure must propagate so the S&F engine
// schedules another retry — mirrors ExternalSystemClient.DeliverBuffered
// letting TransientExternalSystemException escape.
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test") { Id = 1 };
StubConnection(conn);
var gateway = new ExecuteStubGateway(
_repository,
storeAndForward: null,
onExecute: () => throw new TransientDatabaseException("timeout", errorNumber: -2));
var message = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardMessage
{
Id = Guid.NewGuid().ToString("N"),
Category = ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.StoreAndForwardCategory.CachedDbWrite,
Target = "testDb",
PayloadJson =
"""{"ConnectionName":"testDb","Sql":"INSERT INTO t VALUES (1)","Parameters":null}""",
};
await Assert.ThrowsAsync(
() => gateway.DeliverBufferedAsync(message));
}
[Fact]
public async Task DeliverBuffered_PermanentSqlError_ReturnsFalseSoMessageParks()
{
// On the retry path a permanent failure must park the message (return
// false) rather than retry forever — mirrors ExternalSystemClient.
// DeliverBuffered returning false on PermanentExternalSystemException.
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test") { Id = 1 };
StubConnection(conn);
var gateway = new ExecuteStubGateway(
_repository,
storeAndForward: null,
onExecute: () => throw new PermanentDatabaseException(
"Invalid column name", errorNumber: 207));
var message = new ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardMessage
{
Id = Guid.NewGuid().ToString("N"),
Category = ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.StoreAndForwardCategory.CachedDbWrite,
Target = "testDb",
PayloadJson =
"""{"ConnectionName":"testDb","Sql":"INSERT INTO t VALUES (1)","Parameters":null}""",
};
var delivered = await gateway.DeliverBufferedAsync(message);
Assert.False(delivered); // permanent — the S&F engine parks the message
}
// ── M2.3 (#7) code-review fix: ExecuteWriteAsync must classify NON-SqlException
// DB outages as transient (buffer+retry) and propagate cancellation —
// mirroring the HTTP path's ordered catches in InvokeHttpAsync. The pre-fix
// code only caught SqlException, so a live outage surfacing as
// InvalidOperationException / SocketException / IOException / TimeoutException
// escaped unclassified and crashed the Script Execution Actor instead of
// buffering. These tests drive the RAW execution seam (RunSqlAsync) so the
// PRODUCTION classification in ExecuteWriteAsync runs end-to-end. ──
public static IEnumerable