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(); } }