lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherTests.cs

using ZB.MOM.WW.MxGateway.Contracts.Proto;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Health;

/// <summary>
///     Tests for <see cref="PerPlatformProbeWatcher"/> — the per-platform probe state
///     machine. Uses a fake <see cref="IGalaxySubscriber"/> to control SubscribeBulk
///     results and assert the watcher subscribes the right addresses + decodes ScanState
///     values correctly.
/// </summary>
public sealed class PerPlatformProbeWatcherTests
{
    private sealed class FakeSubscriber : IGalaxySubscriber
    {
        /// <summary>Gets a list of all subscribe requests made to the subscriber.</summary>
        public List<List<string>> Subscribes { get; } = [];
        /// <summary>Gets the buffered update intervals used in each subscribe request.</summary>
        public List<int> SubscribeIntervalsMs { get; } = [];
        /// <summary>Gets a list of all unsubscribe requests made to the subscriber.</summary>
        public List<List<int>> Unsubscribes { get; } = [];
        private int _nextHandle = 1;
        /// <summary>Gets a mapping of tag addresses to their assigned item handles.</summary>
        public Dictionary<string, int> HandleByAddress { get; } = new(StringComparer.OrdinalIgnoreCase);

        /// <inheritdoc />
        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
        {
            Subscribes.Add([.. fullReferences]);
            SubscribeIntervalsMs.Add(bufferedUpdateIntervalMs);
            var results = new List<SubscribeResult>(fullReferences.Count);
            foreach (var addr in fullReferences)
            {
                var handle = Interlocked.Increment(ref _nextHandle);
                HandleByAddress[addr] = handle;
                results.Add(new SubscribeResult
                {
                    TagAddress = addr,
                    ItemHandle = handle,
                    WasSuccessful = true,
                });
            }
            return Task.FromResult<IReadOnlyList<SubscribeResult>>(results);
        }

        /// <inheritdoc />
        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
        {
            Unsubscribes.Add([.. itemHandles]);
            return Task.CompletedTask;
        }

        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> StreamEventsAsync(CancellationToken cancellationToken)
            => Empty();

        private static async IAsyncEnumerable<MxEvent> Empty()
        {
            await Task.CompletedTask;
            yield break;
        }
    }

    /// <summary>Verifies that syncing platforms subscribes to the ScanState address for each platform.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SyncPlatformsAsync_SubscribesScanStateAddressForEachPlatform()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        using var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["PlatformA", "PlatformB"], CancellationToken.None);

        subscriber.Subscribes.Count.ShouldBe(1);
        subscriber.Subscribes[0].ShouldBe(new[] { "PlatformA.ScanState", "PlatformB.ScanState" });
        watcher.WatchedPlatforms.OrderBy(x => x).ShouldBe(new[] { "PlatformA", "PlatformB" });
    }

    /// <summary>Verifies that the default buffered interval is zero, matching gateway cadence.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SyncPlatformsAsync_DefaultBufferedIntervalIsZero_GwCadence()
    {
        // PR 6.3 — without an override, the watcher passes 0 (gw default cadence) so
        // existing deployments don't see a behavior change.
        var subscriber = new FakeSubscriber();
        using var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator());
        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);
        subscriber.SubscribeIntervalsMs.ShouldHaveSingleItem().ShouldBe(0);
    }

    /// <summary>Verifies that a configured buffered interval is forwarded to the gateway.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SyncPlatformsAsync_ConfiguredBufferedInterval_IsForwardedToGw()
    {
        // PR 6.3 — when a deployment dials down MxAccess.PublishingIntervalMs for
        // tighter health visibility, the probe watcher must forward that interval
        // through SubscribeBulk so the gw publishes ScanState changes at the
        // configured cadence.
        var subscriber = new FakeSubscriber();
        using var watcher = new PerPlatformProbeWatcher(
            subscriber, new HostStatusAggregator(),
            bufferedUpdateIntervalMs: 250);
        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);
        subscriber.SubscribeIntervalsMs.ShouldHaveSingleItem().ShouldBe(250);
    }

    /// <summary>Verifies that the constructor rejects negative buffered intervals.</summary>
    [Fact]
    public void Constructor_RejectsNegativeBufferedInterval()
    {
        var subscriber = new FakeSubscriber();
        Should.Throw<ArgumentOutOfRangeException>(() =>
            new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), bufferedUpdateIntervalMs: -1));
    }

    /// <summary>Verifies that syncing the same platform set twice does not resubscribe.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SyncPlatformsAsync_SameSetTwice_DoesNotResubscribe()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        using var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);
        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);

        subscriber.Subscribes.Count.ShouldBe(1);
    }

    /// <summary>Verifies that removed platforms are unsubscribed and dropped from the aggregator.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SyncPlatformsAsync_RemovedPlatforms_AreUnsubscribed_AndDroppedFromAggregator()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        using var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["A", "B"], CancellationToken.None);
        var bHandle = subscriber.HandleByAddress["B.ScanState"];

        // Push a value so B is in the aggregator before we remove it.
        watcher.OnProbeValueChanged("B.ScanState", true, qualityByte: 192);
        aggregator.Snapshot().Any(s => s.HostName == "B").ShouldBeTrue();

        await watcher.SyncPlatformsAsync(["A"], CancellationToken.None);

        subscriber.Unsubscribes.Count.ShouldBe(1);
        subscriber.Unsubscribes[0].ShouldBe(new[] { bHandle });
        watcher.WatchedPlatforms.ShouldBe(new[] { "A" });
        aggregator.Snapshot().Any(s => s.HostName == "B").ShouldBeFalse();
    }

    /// <summary>Verifies that DecodeState correctly decodes ScanState values and quality bytes across multiple pin configurations.</summary>
    /// <param name="value">The probe value to decode.</param>
    /// <param name="qualityByte">The OPC UA quality byte indicating data validity.</param>
    /// <param name="expected">The expected decoded host state.</param>
    [Theory]
    [InlineData(true, (byte)192, HostState.Running)]
    [InlineData(false, (byte)192, HostState.Stopped)]
    [InlineData(1, (byte)192, HostState.Running)]
    [InlineData(0, (byte)192, HostState.Stopped)]
    [InlineData("Running", (byte)192, HostState.Running)]
    [InlineData("Stopped", (byte)192, HostState.Stopped)]
    [InlineData("running", (byte)192, HostState.Running)]
    [InlineData(2, (byte)192, HostState.Faulted)]                  // unknown int
    [InlineData("Whatever", (byte)192, HostState.Faulted)]         // unknown string
    [InlineData(true, (byte)64, HostState.Unknown)]                // bad quality wins
    [InlineData(true, (byte)0, HostState.Unknown)]
    public void DecodeState_TablePins(object? value, byte qualityByte, HostState expected)
    {
        PerPlatformProbeWatcher.DecodeState(value, qualityByte).ShouldBe(expected);
    }

    /// <summary>Verifies that a running probe value is routed to the aggregator.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task OnProbeValueChanged_Running_RoutesToAggregator()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        using var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);
        watcher.OnProbeValueChanged("PlatformA.ScanState", true, qualityByte: 192);

        var snap = aggregator.Snapshot().Single(s => s.HostName == "PlatformA");
        snap.State.ShouldBe(HostState.Running);
    }

    /// <summary>Verifies that a probe value with bad quality routes as unknown state.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task OnProbeValueChanged_BadQuality_RoutesUnknown()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        using var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);
        watcher.OnProbeValueChanged("PlatformA.ScanState", true, qualityByte: 0);

        aggregator.Snapshot().Single(s => s.HostName == "PlatformA").State.ShouldBe(HostState.Unknown);
    }

    /// <summary>Verifies that foreign probe references are silently dropped.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task OnProbeValueChanged_ForeignReference_IsSilentlyDropped()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        using var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["PlatformA"], CancellationToken.None);

        // Reference doesn't end with .ScanState — silently dropped.
        watcher.OnProbeValueChanged("PlatformA.SomethingElse", true, qualityByte: 192);
        aggregator.Snapshot().Any(s => s.HostName == "PlatformA").ShouldBeFalse();

        // Unknown platform — silently dropped.
        watcher.OnProbeValueChanged("Stranger.ScanState", true, qualityByte: 192);
        aggregator.Snapshot().Any(s => s.HostName == "Stranger").ShouldBeFalse();
    }

    /// <summary>Verifies that dispose unsubscribes all tracked platforms.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Dispose_UnsubscribesAllTrackedPlatforms()
    {
        var subscriber = new FakeSubscriber();
        var aggregator = new HostStatusAggregator();
        var watcher = new PerPlatformProbeWatcher(subscriber, aggregator);

        await watcher.SyncPlatformsAsync(["A", "B", "C"], CancellationToken.None);
        var expectedHandles = new[]
        {
            subscriber.HandleByAddress["A.ScanState"],
            subscriber.HandleByAddress["B.ScanState"],
            subscriber.HandleByAddress["C.ScanState"],
        };

        watcher.Dispose();

        subscriber.Unsubscribes.Count.ShouldBe(1);
        subscriber.Unsubscribes[0].OrderBy(x => x).ShouldBe(expectedHandles.OrderBy(x => x));
    }
}