fix(external-system-gateway): resolve ExternalSystemGateway-004..010 — honour retry settings, dispose HTTP messages, fix URL building, truncate error bodies, fix connection leak
This commit is contained in:
Joseph Doherty
@@ -1,3 +1,5 @@
|
||||
using System.Data;
|
||||
using System.Data.Common;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using NSubstitute;
|
||||
using ScadaLink.Commons.Entities.ExternalSystems;
|
||||
@@ -75,4 +77,66 @@ public class DatabaseGatewayTests
|
||||
|
||||
Assert.False(delivered); // permanent — the S&F engine parks the message
|
||||
}
|
||||
|
||||
// ── ExternalSystemGateway-010: SqlConnection must not leak when OpenAsync fails ──
|
||||
|
||||
[Fact]
|
||||
public async Task GetConnection_OpenFails_DisposesConnectionBeforeRethrowing()
|
||||
{
|
||||
var conn = new DatabaseConnectionDefinition("testDb", "Server=localhost;Database=test") { Id = 1 };
|
||||
_repository.GetAllDatabaseConnectionsAsync().Returns(new List<DatabaseConnectionDefinition> { conn });
|
||||
|
||||
var fake = new ThrowingDbConnection();
|
||||
var gateway = new ConnectionFactoryStubGateway(_repository, fake);
|
||||
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(
|
||||
() => gateway.GetConnectionAsync("testDb"));
|
||||
|
||||
Assert.True(fake.WasDisposed, "The SqlConnection was leaked — it must be disposed when OpenAsync fails");
|
||||
}
|
||||
|
||||
/// <summary>Test gateway that substitutes the connection factory with a stub.</summary>
|
||||
private sealed class ConnectionFactoryStubGateway : DatabaseGateway
|
||||
{
|
||||
private readonly DbConnection _connection;
|
||||
|
||||
public ConnectionFactoryStubGateway(IExternalSystemRepository repository, DbConnection connection)
|
||||
: base(repository, NullLogger<DatabaseGateway>.Instance) => _connection = connection;
|
||||
|
||||
internal override DbConnection CreateConnection(string connectionString) => _connection;
|
||||
}
|
||||
|
||||
/// <summary>A DbConnection whose OpenAsync always fails, tracking whether it was disposed.</summary>
|
||||
private sealed class ThrowingDbConnection : DbConnection
|
||||
{
|
||||
public bool WasDisposed { get; private set; }
|
||||
|
||||
public override Task OpenAsync(CancellationToken cancellationToken) =>
|
||||
throw new InvalidOperationException("simulated open failure");
|
||||
public override void Open() => throw new InvalidOperationException("simulated open failure");
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing) WasDisposed = true;
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
public override ValueTask DisposeAsync()
|
||||
{
|
||||
WasDisposed = true;
|
||||
return base.DisposeAsync();
|
||||
}
|
||||
|
||||
// Unused abstract members.
|
||||
[System.Diagnostics.CodeAnalysis.AllowNull]
|
||||
public override string ConnectionString { get; set; } = string.Empty;
|
||||
public override string Database => string.Empty;
|
||||
public override string DataSource => string.Empty;
|
||||
public override string ServerVersion => string.Empty;
|
||||
public override ConnectionState State => ConnectionState.Closed;
|
||||
public override void ChangeDatabase(string databaseName) => throw new NotSupportedException();
|
||||
public override void Close() { }
|
||||
protected override DbTransaction BeginDbTransaction(IsolationLevel il) => throw new NotSupportedException();
|
||||
protected override DbCommand CreateDbCommand() => throw new NotSupportedException();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -328,6 +328,281 @@ public class ExternalSystemClientTests
|
||||
() => client.CallAsync("TestAPI", "getData", cancellationToken: cts.Token));
|
||||
}
|
||||
|
||||
// ── ExternalSystemGateway-004: per-system retry settings honoured for cached calls ──
|
||||
|
||||
[Fact]
|
||||
public async Task CachedCall_TransientFailure_BuffersWithSystemRetrySettings()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none")
|
||||
{
|
||||
Id = 1,
|
||||
MaxRetries = 7,
|
||||
RetryDelay = TimeSpan.FromSeconds(42),
|
||||
};
|
||||
var method = new ExternalSystemMethod("postData", "POST", "/post") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var httpClient = new HttpClient(new MockHttpMessageHandler(HttpStatusCode.InternalServerError, "boom"));
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var dbName = $"EsgRetry_{Guid.NewGuid():N}";
|
||||
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
|
||||
using var keepAlive = new SqliteConnection(connStr);
|
||||
keepAlive.Open();
|
||||
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
// S&F defaults deliberately different from the system's settings.
|
||||
var sfOptions = new StoreAndForwardOptions
|
||||
{
|
||||
DefaultMaxRetries = 3,
|
||||
DefaultRetryInterval = TimeSpan.FromMinutes(10),
|
||||
RetryTimerInterval = TimeSpan.FromMinutes(10),
|
||||
};
|
||||
var sf = new StoreAndForwardService(storage, sfOptions, NullLogger<StoreAndForwardService>.Instance);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance,
|
||||
storeAndForward: sf);
|
||||
|
||||
var result = await client.CachedCallAsync("TestAPI", "postData");
|
||||
Assert.True(result.WasBuffered);
|
||||
|
||||
var depth = await storage.GetBufferDepthByCategoryAsync();
|
||||
Assert.Equal(1, depth[ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.ExternalSystem]);
|
||||
|
||||
var (maxRetries, retryIntervalMs) = ReadBufferedRetrySettings(connStr);
|
||||
Assert.Equal(7, maxRetries);
|
||||
Assert.Equal((long)TimeSpan.FromSeconds(42).TotalMilliseconds, retryIntervalMs);
|
||||
}
|
||||
|
||||
private static (int MaxRetries, long RetryIntervalMs) ReadBufferedRetrySettings(string connStr)
|
||||
{
|
||||
using var conn = new SqliteConnection(connStr);
|
||||
conn.Open();
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = "SELECT max_retries, retry_interval_ms FROM sf_messages";
|
||||
using var reader = cmd.ExecuteReader();
|
||||
Assert.True(reader.Read(), "expected exactly one buffered message");
|
||||
var result = (reader.GetInt32(0), reader.GetInt64(1));
|
||||
Assert.False(reader.Read(), "expected exactly one buffered message");
|
||||
return result;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CachedCall_TransientFailure_ZeroMaxRetriesIsHonouredNotTreatedAsUnset()
|
||||
{
|
||||
// MaxRetries == 0 must mean "never retry", not "fall back to the S&F default".
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none")
|
||||
{
|
||||
Id = 1,
|
||||
MaxRetries = 0,
|
||||
RetryDelay = TimeSpan.FromSeconds(5),
|
||||
};
|
||||
var method = new ExternalSystemMethod("postData", "POST", "/post") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var httpClient = new HttpClient(new MockHttpMessageHandler(HttpStatusCode.InternalServerError, "boom"));
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var dbName = $"EsgRetryZero_{Guid.NewGuid():N}";
|
||||
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
|
||||
using var keepAlive = new SqliteConnection(connStr);
|
||||
keepAlive.Open();
|
||||
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
var sfOptions = new StoreAndForwardOptions
|
||||
{
|
||||
DefaultMaxRetries = 99,
|
||||
DefaultRetryInterval = TimeSpan.FromMinutes(10),
|
||||
RetryTimerInterval = TimeSpan.FromMinutes(10),
|
||||
};
|
||||
var sf = new StoreAndForwardService(storage, sfOptions, NullLogger<StoreAndForwardService>.Instance);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance,
|
||||
storeAndForward: sf);
|
||||
|
||||
await client.CachedCallAsync("TestAPI", "postData");
|
||||
|
||||
var (maxRetries, _) = ReadBufferedRetrySettings(connStr);
|
||||
Assert.Equal(0, maxRetries); // honoured — not the default of 99
|
||||
}
|
||||
|
||||
// ── ExternalSystemGateway-005: HttpRequestMessage / HttpResponseMessage disposal ──
|
||||
|
||||
[Fact]
|
||||
public async Task Call_SuccessfulHttp_DisposesRequestAndResponse()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none") { Id = 1 };
|
||||
var method = new ExternalSystemMethod("getData", "GET", "/data") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var handler = new DisposalTrackingHandler(HttpStatusCode.OK, "{\"ok\":true}");
|
||||
var httpClient = new HttpClient(handler);
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
|
||||
|
||||
await client.CallAsync("TestAPI", "getData");
|
||||
|
||||
Assert.True(handler.RequestDisposed, "HttpRequestMessage was not disposed");
|
||||
Assert.True(handler.ResponseContentDisposed, "HttpResponseMessage content was not disposed");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Call_PermanentFailure_StillDisposesRequestAndResponse()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none") { Id = 1 };
|
||||
var method = new ExternalSystemMethod("badMethod", "POST", "/bad") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var handler = new DisposalTrackingHandler(HttpStatusCode.BadRequest, "bad request");
|
||||
var httpClient = new HttpClient(handler);
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
|
||||
|
||||
await client.CallAsync("TestAPI", "badMethod");
|
||||
|
||||
Assert.True(handler.RequestDisposed, "HttpRequestMessage was not disposed on the error path");
|
||||
Assert.True(handler.ResponseContentDisposed, "HttpResponseMessage content was not disposed on the error path");
|
||||
}
|
||||
|
||||
// ── ExternalSystemGateway-006: BuildUrl — empty path must not append a trailing slash ──
|
||||
|
||||
[Fact]
|
||||
public async Task Call_MethodWithEmptyPath_DoesNotAppendTrailingSlash()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com/api", "none") { Id = 1 };
|
||||
var method = new ExternalSystemMethod("root", "GET", "") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var handler = new RequestCapturingHandler(HttpStatusCode.OK, "{}");
|
||||
var httpClient = new HttpClient(handler);
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
|
||||
|
||||
await client.CallAsync("TestAPI", "root");
|
||||
|
||||
Assert.Equal("https://api.example.com/api", handler.LastUri!.ToString());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Call_MethodWithPath_BuildsExpectedUrl()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com/api", "none") { Id = 1 };
|
||||
var method = new ExternalSystemMethod("getData", "GET", "/data") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var handler = new RequestCapturingHandler(HttpStatusCode.OK, "{}");
|
||||
var httpClient = new HttpClient(handler);
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
|
||||
|
||||
await client.CallAsync("TestAPI", "getData");
|
||||
|
||||
Assert.Equal("https://api.example.com/api/data", handler.LastUri!.ToString());
|
||||
}
|
||||
|
||||
// ── ExternalSystemGateway-007: external error body must be truncated, not echoed verbatim ──
|
||||
|
||||
[Fact]
|
||||
public async Task Call_PermanentFailureWithHugeErrorBody_TruncatesErrorMessage()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none") { Id = 1 };
|
||||
var method = new ExternalSystemMethod("badMethod", "POST", "/bad") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var hugeBody = new string('X', 500_000);
|
||||
var handler = new MockHttpMessageHandler(HttpStatusCode.BadRequest, hugeBody);
|
||||
var httpClient = new HttpClient(handler);
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
|
||||
|
||||
var result = await client.CallAsync("TestAPI", "badMethod");
|
||||
|
||||
Assert.False(result.Success);
|
||||
// The error message must be bounded — a misbehaving endpoint cannot inflate
|
||||
// every script-visible error string / event-log entry.
|
||||
Assert.True(result.ErrorMessage!.Length < 4096,
|
||||
$"Error message was {result.ErrorMessage.Length} chars — expected it to be truncated");
|
||||
}
|
||||
|
||||
// ── ExternalSystemGateway-008: cancellation of a CachedCall must not be buffered ──
|
||||
|
||||
[Fact]
|
||||
public async Task CachedCall_CallerCancellation_IsNotBufferedAsTransient()
|
||||
{
|
||||
var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none") { Id = 1 };
|
||||
var method = new ExternalSystemMethod("postData", "POST", "/post") { Id = 1, ExternalSystemDefinitionId = 1 };
|
||||
_repository.GetAllExternalSystemsAsync(Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemDefinition> { system });
|
||||
_repository.GetMethodsByExternalSystemIdAsync(1, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<ExternalSystemMethod> { method });
|
||||
|
||||
var httpClient = new HttpClient(new HangingHttpMessageHandler(TimeSpan.FromMinutes(10)));
|
||||
_httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
|
||||
|
||||
var dbName = $"EsgCancel_{Guid.NewGuid():N}";
|
||||
var connStr = $"Data Source={dbName};Mode=Memory;Cache=Shared";
|
||||
using var keepAlive = new SqliteConnection(connStr);
|
||||
keepAlive.Open();
|
||||
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
|
||||
await storage.InitializeAsync();
|
||||
var sfOptions = new StoreAndForwardOptions
|
||||
{
|
||||
DefaultRetryInterval = TimeSpan.FromMinutes(10),
|
||||
RetryTimerInterval = TimeSpan.FromMinutes(10),
|
||||
};
|
||||
var sf = new StoreAndForwardService(storage, sfOptions, NullLogger<StoreAndForwardService>.Instance);
|
||||
|
||||
var options = new ExternalSystemGatewayOptions { DefaultHttpTimeout = TimeSpan.FromMinutes(5) };
|
||||
var client = new ExternalSystemClient(
|
||||
_httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance,
|
||||
storeAndForward: sf,
|
||||
options: Microsoft.Extensions.Options.Options.Create(options));
|
||||
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(200));
|
||||
|
||||
// Caller asked to abandon the work — it must NOT be buffered for retry.
|
||||
await Assert.ThrowsAnyAsync<OperationCanceledException>(
|
||||
() => client.CachedCallAsync("TestAPI", "postData", cancellationToken: cts.Token));
|
||||
|
||||
var depth = await storage.GetBufferDepthByCategoryAsync();
|
||||
Assert.False(
|
||||
depth.TryGetValue(ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.ExternalSystem, out var n) && n > 0,
|
||||
"A caller-cancelled CachedCall must not be buffered for retry");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Test helper: mock HTTP message handler.
|
||||
/// </summary>
|
||||
@@ -351,6 +626,72 @@ public class ExternalSystemClientTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Test helper: tracks disposal of the request and the response content.
|
||||
/// </summary>
|
||||
private class DisposalTrackingHandler : HttpMessageHandler
|
||||
{
|
||||
private readonly HttpStatusCode _statusCode;
|
||||
private readonly string _body;
|
||||
|
||||
public DisposalTrackingHandler(HttpStatusCode statusCode, string body)
|
||||
{
|
||||
_statusCode = statusCode;
|
||||
_body = body;
|
||||
}
|
||||
|
||||
public bool RequestDisposed { get; private set; }
|
||||
public bool ResponseContentDisposed { get; private set; }
|
||||
|
||||
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
|
||||
{
|
||||
request.Content = new TrackingContent(string.Empty, () => RequestDisposed = true);
|
||||
var response = new HttpResponseMessage(_statusCode)
|
||||
{
|
||||
Content = new TrackingContent(_body, () => ResponseContentDisposed = true)
|
||||
};
|
||||
return Task.FromResult(response);
|
||||
}
|
||||
|
||||
private sealed class TrackingContent : StringContent
|
||||
{
|
||||
private readonly Action _onDispose;
|
||||
public TrackingContent(string content, Action onDispose) : base(content) => _onDispose = onDispose;
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing) _onDispose();
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Test helper: captures the request URI of the last request.
|
||||
/// </summary>
|
||||
private class RequestCapturingHandler : HttpMessageHandler
|
||||
{
|
||||
private readonly HttpStatusCode _statusCode;
|
||||
private readonly string _body;
|
||||
|
||||
public RequestCapturingHandler(HttpStatusCode statusCode, string body)
|
||||
{
|
||||
_statusCode = statusCode;
|
||||
_body = body;
|
||||
}
|
||||
|
||||
public Uri? LastUri { get; private set; }
|
||||
|
||||
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
|
||||
{
|
||||
LastUri = request.RequestUri;
|
||||
return Task.FromResult(new HttpResponseMessage(_statusCode)
|
||||
{
|
||||
Content = new StringContent(_body)
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Test helper: an HTTP handler that hangs until cancelled (simulates a slow/hung system).
|
||||
/// </summary>
|
||||
|
||||
Reference in New Issue
Block a user