lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs

using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Shouldly;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry;
using Xunit;

namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Telemetry;

/// <summary>
///     Unit tests for <see cref="TelemetryStreamClient"/> driving a fake generated client (the ctor
///     seam), so the pump/normal-shutdown/error-routing behaviour is provable without a live gRPC
///     server. The wire itself is covered by the Phase 5 live gate. The central assertion throughout is
///     the onError contract: only a transport fault reaches onError; programming faults throw
///     synchronously and consumer-callback faults are isolated.
/// </summary>
public sealed class TelemetryStreamClientTests
{
    [Fact]
    public async Task RunAsync_pumps_every_event_to_onEvent()
    {
        var events = new[]
        {
            new TelemetryEvent { ScriptLog = Script("one") },
            new TelemetryEvent { ScriptLog = Script("two") },
        };
        var fake = new FakeClient(new FakeStreamReader(events));
        using var sut = new TelemetryStreamClient(fake, "key");

        var received = new List<TelemetryEvent>();
        Exception? errored = null;

        await sut.RunAsync("corr-1", received.Add, ex => errored = ex, CancellationToken.None);

        received.Count.ShouldBe(2);
        received[0].ScriptLog.Message.ShouldBe("one");
        received[1].ScriptLog.Message.ShouldBe("two");
        errored.ShouldBeNull();
    }

    [Fact]
    public async Task RunAsync_swallows_rpc_cancelled_without_calling_onError()
    {
        var fake = new FakeClient(new FakeStreamReader(
            [new TelemetryEvent { ScriptLog = Script("one") }],
            throwAtEnd: new RpcException(new Status(StatusCode.Cancelled, "cancelled"))));
        using var sut = new TelemetryStreamClient(fake, "key");

        var received = new List<TelemetryEvent>();
        Exception? errored = null;

        await sut.RunAsync("corr-1", received.Add, ex => errored = ex, CancellationToken.None);

        received.Count.ShouldBe(1);
        errored.ShouldBeNull();
    }

    [Fact]
    public async Task RunAsync_cancelled_token_ends_without_calling_onError()
    {
        // The primary shutdown branch the supervisor's routine reconnect-teardown drives: the token
        // fires mid-stream, the transport surfaces an OperationCanceledException, and RunAsync returns
        // cleanly — never onError.
        using var cts = new CancellationTokenSource();
        var fake = new FakeClient(new FakeStreamReader(
            [new TelemetryEvent { ScriptLog = Script("one") }],
            cancelAfterFirst: cts));
        using var sut = new TelemetryStreamClient(fake, "key");

        var received = new List<TelemetryEvent>();
        Exception? errored = null;

        await sut.RunAsync("corr-1", received.Add, ex => errored = ex, cts.Token);

        received.Count.ShouldBe(1);
        errored.ShouldBeNull();
    }

    [Fact]
    public async Task RunAsync_routes_real_rpc_failure_to_onError()
    {
        var boom = new RpcException(new Status(StatusCode.Unavailable, "node down"));
        var fake = new FakeClient(new FakeStreamReader([], throwAtEnd: boom));
        using var sut = new TelemetryStreamClient(fake, "key");

        Exception? errored = null;

        await sut.RunAsync("corr-1", _ => { }, ex => errored = ex, CancellationToken.None);

        errored.ShouldBeSameAs(boom);
    }

    [Fact]
    public async Task RunAsync_onEvent_exception_is_isolated_pump_continues_and_onError_not_called()
    {
        var events = new[]
        {
            new TelemetryEvent { ScriptLog = Script("poison") },
            new TelemetryEvent { ScriptLog = Script("healthy") },
        };
        var fake = new FakeClient(new FakeStreamReader(events));
        using var sut = new TelemetryStreamClient(fake, "key");

        var received = new List<string>();
        Exception? errored = null;

        await sut.RunAsync(
            "corr-1",
            evt =>
            {
                if (evt.ScriptLog.Message == "poison")
                    throw new InvalidOperationException("unmappable event");
                received.Add(evt.ScriptLog.Message);
            },
            ex => errored = ex,
            CancellationToken.None);

        // The poison event was dropped, the pump continued to the next event, and onError never fired.
        received.ShouldBe(["healthy"]);
        errored.ShouldBeNull();
    }

    [Fact]
    public async Task RunAsync_empty_correlationId_throws_synchronously_and_never_calls_onError()
    {
        var fake = new FakeClient(new FakeStreamReader([]));
        using var sut = new TelemetryStreamClient(fake, "key");

        Exception? errored = null;

        await Should.ThrowAsync<ArgumentException>(() =>
            sut.RunAsync("", _ => { }, ex => errored = ex, CancellationToken.None));

        errored.ShouldBeNull();
    }

    [Fact]
    public async Task RunAsync_after_dispose_throws_ObjectDisposed_and_never_calls_onError()
    {
        var fake = new FakeClient(new FakeStreamReader([]));
        var sut = new TelemetryStreamClient(fake, "key");
        sut.Dispose();

        Exception? errored = null;

        await Should.ThrowAsync<ObjectDisposedException>(() =>
            sut.RunAsync("corr-1", _ => { }, ex => errored = ex, CancellationToken.None));

        errored.ShouldBeNull();
    }

    private static ScriptLog Script(string message) => new()
    {
        ScriptId = "s",
        Level = "Information",
        Message = message,
        TimestampUtc = Timestamp.FromDateTime(new DateTime(2026, 7, 23, 0, 0, 0, DateTimeKind.Utc)),
    };

    /// <summary>Fake generated client whose Subscribe returns a canned server-streaming call.</summary>
    private sealed class FakeClient : TelemetryStreamService.TelemetryStreamServiceClient
    {
        private readonly IAsyncStreamReader<TelemetryEvent> _reader;

        public FakeClient(IAsyncStreamReader<TelemetryEvent> reader) => _reader = reader;

        public override AsyncServerStreamingCall<TelemetryEvent> Subscribe(
            TelemetryStreamRequest request,
            Metadata? headers = null,
            DateTime? deadline = null,
            CancellationToken cancellationToken = default) =>
            new(
                _reader,
                Task.FromResult(new Metadata()),
                () => Status.DefaultSuccess,
                () => [],
                () => { });
    }

    /// <summary>
    ///     Replays a fixed sequence of events. Honors the pump's <see cref="CancellationToken"/> (so a
    ///     cancelled token surfaces as <see cref="OperationCanceledException"/> just as the real
    ///     transport does), can throw a supplied exception at the end of the sequence, and can cancel a
    ///     supplied source after delivering the first item (to drive the mid-stream cancel branch).
    /// </summary>
    private sealed class FakeStreamReader : IAsyncStreamReader<TelemetryEvent>
    {
        private readonly IReadOnlyList<TelemetryEvent> _items;
        private readonly Exception? _throwAtEnd;
        private readonly CancellationTokenSource? _cancelAfterFirst;
        private int _index = -1;

        public FakeStreamReader(
            IReadOnlyList<TelemetryEvent> items,
            Exception? throwAtEnd = null,
            CancellationTokenSource? cancelAfterFirst = null)
        {
            _items = items;
            _throwAtEnd = throwAtEnd;
            _cancelAfterFirst = cancelAfterFirst;
        }

        public TelemetryEvent Current => _items[_index];

        public Task<bool> MoveNext(CancellationToken cancellationToken)
        {
            cancellationToken.ThrowIfCancellationRequested();

            _index++;
            if (_index < _items.Count)
            {
                // After delivering the first item, request cancellation so the NEXT MoveNext throws
                // OperationCanceledException at the top — the transport-cancel shape.
                if (_index == 0)
                    _cancelAfterFirst?.Cancel();
                return Task.FromResult(true);
            }

            if (_throwAtEnd is not null)
                throw _throwAtEnd;

            return Task.FromResult(false);
        }
    }
}