lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayProbeHealthTests.cs

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;

/// <summary>
///     The Gateway historian's connection health is no longer derived from read outcomes alone: a slow
///     <c>ProbeAsync</c> timer (<c>ServerHistorian:Gateway:ProbeInterval</c>) 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.
/// </summary>
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<bool> 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<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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<HistorianSample>() };
        await using var ds = new GatewayHistorianDataSource(fake, NullLogger<GatewayHistorianDataSource>.Instance,
            probeInterval: TimeSpan.FromHours(1));
        await Eventually(() => ds.GetHealthSnapshot().ProcessConnectionOpen, "the immediate first probe opens");

        fake.ThrowOnRead = new InvalidOperationException("boom");
        await Should.ThrowAsync<InvalidOperationException>(() => 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<GatewayHistorianDataSource>.Instance);
        // Positive control in the same window: an enabled source does probe.
        var control = new FakeHistorianGatewayClient();
        await using var probed = new GatewayHistorianDataSource(control, NullLogger<GatewayHistorianDataSource>.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<GatewayHistorianDataSource>.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<string, string?>
        {
            ["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<ServerHistorianOptions>()!;
        using var services = new ServiceCollection().BuildServiceProvider();
        var ds = Assert.IsType<GatewayHistorianDataSource>(GatewayHistorian.CreateDataSource(opts, services));
        ds.ProbeInterval.ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
        ds.Dispose();
    }
}