using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
///
/// The Gateway historian's connection health is no longer derived from read outcomes alone: a slow
/// ProbeAsync timer (ServerHistorian:Gateway:ProbeInterval) reports a healthy gateway as open on a node
/// that has served no HistoryRead yet (or only writes), and a dead one as closed. The latest observation — probe or
/// read — wins.
///
public sealed class GatewayProbeHealthTests
{
private static readonly TimeSpan Fast = TimeSpan.FromMilliseconds(30);
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static async Task Eventually(Func condition, string because)
{
var deadline = DateTime.UtcNow.AddSeconds(5);
while (!condition())
{
if (DateTime.UtcNow > deadline) throw new ShouldAssertException($"timed out waiting: {because}");
await Task.Delay(10, Ct);
}
}
[Fact]
public async Task A_healthy_gateway_reads_open_before_any_HistoryRead()
{
var fake = new FakeHistorianGatewayClient { ProbeResult = true };
await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, probeInterval: Fast);
await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "the probe should open the connection flag");
var h = ds.GetHealthSnapshot();
h.EventConnectionOpen.ShouldBeTrue();
h.TotalQueries.ShouldBe(0, "a probe is not a history query");
fake.ProbeCallCount.ShouldBeGreaterThan(0);
}
[Fact]
public async Task A_failing_or_throwing_probe_reads_closed_and_the_latest_observation_wins()
{
var fake = new FakeHistorianGatewayClient { ProbeResult = false };
await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, probeInterval: Fast);
await Eventually(() => fake.ProbeCallCount >= 2, "probes ran");
ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse();
// A thrown probe (transport down) is closed too, and never kills the loop.
fake.ProbeThrows = new InvalidOperationException("down");
var before = fake.ProbeCallCount;
await Eventually(() => fake.ProbeCallCount >= before + 2, "the loop survives a throwing probe");
ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse();
// The probe recovers ⇒ open, with no read at all.
fake.ProbeThrows = null;
fake.ProbeResult = true;
await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "a recovered probe reopens");
}
[Fact]
public async Task A_failed_read_after_a_good_probe_closes_until_the_next_good_observation()
{
var fake = new FakeHistorianGatewayClient { ProbeResult = true, RawSamples = Array.Empty() };
await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance,
probeInterval: TimeSpan.FromHours(1));
await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "the immediate first probe opens");
fake.ThrowOnRead = new InvalidOperationException("boom");
await Should.ThrowAsync(() => ds.ReadRawAsync("T", default, default, 1, Ct));
ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse("the read failed after the probe");
fake.ThrowOnRead = null;
await ds.ReadRawAsync("T", default, default, 1, Ct);
ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeTrue();
}
[Fact]
public async Task Without_a_probe_interval_nothing_probes_and_health_stays_read_derived()
{
var fake = new FakeHistorianGatewayClient();
await using var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance);
// Positive control in the same window: an enabled source does probe.
var control = new FakeHistorianGatewayClient();
await using var probed = new GatewayHistorianDataSource(control, NullLogger.Instance, probeInterval: Fast);
await Eventually(() => control.ProbeCallCount >= 2, "the control probes");
fake.ProbeCallCount.ShouldBe(0);
ds.GetHealthSnapshot().ProcessConnectionOpen.ShouldBeFalse();
}
[Fact]
public async Task Disposing_stops_the_probe_loop()
{
var fake = new FakeHistorianGatewayClient();
var ds = new GatewayHistorianDataSource(fake, NullLogger.Instance, probeInterval: Fast);
await Eventually(() => fake.ProbeCallCount >= 2, "probes ran");
await ds.DisposeAsync();
var after = fake.ProbeCallCount;
await Task.Delay(Fast * 5, Ct);
fake.ProbeCallCount.ShouldBe(after);
fake.DisposeCallCount.ShouldBe(1);
}
[Theory]
[InlineData(null, 30)]
[InlineData("00:00:05", 5)]
[InlineData("00:00:00", 0)]
public void The_probe_interval_binds_from_configuration_and_reaches_the_data_source(string? configured, int expectedSeconds)
{
var values = new Dictionary
{
["ServerHistorian:Enabled"] = "true",
["ServerHistorian:Endpoint"] = "https://localhost:5222",
["ServerHistorian:ApiKey"] = "histgw_x_y",
};
if (configured is not null) values["ServerHistorian:Gateway:ProbeInterval"] = configured;
var opts = new ConfigurationBuilder().AddInMemoryCollection(values).Build()
.GetSection(ServerHistorianOptions.SectionName).Get()!;
using var services = new ServiceCollection().BuildServiceProvider();
var ds = Assert.IsType(GatewayHistorian.CreateDataSource(opts, services));
ds.ProbeInterval.ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
ds.Dispose();
}
}