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

using System.Runtime.CompilerServices;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;

/// <summary>
/// Reusable in-memory test double for <see cref="IHistorianGatewayClient"/>. Every method returns
/// from a public settable result field and records its call arguments into public fields, so the
/// later driver tasks (T7/T8/T11/T12/T14) can drive behaviour and assert on what the driver sent
/// without a live gateway. Throw fields let a test simulate transport faults (e.g. an
/// <c>RpcException</c>) per operation; reads share <see cref="ThrowOnRead"/>.
/// </summary>
public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
{
    public bool TagExistsResult = true;
    public int TagExistsCallCount;
    public Exception? TagExistsThrows;
    /// <summary>Optional asynchronous catalog response for overlapping presence-cache tests.</summary>
    public Func<string, CancellationToken, Task<bool>>? TagExistsSource;
    public Task<bool> TagExistsAsync(string tag, CancellationToken ct)
    {
        ct.ThrowIfCancellationRequested();
        Interlocked.Increment(ref TagExistsCallCount);
        if (TagExistsSource is { } source) return source(tag, ct);
        if (TagExistsThrows is { } error) throw error;
        return Task.FromResult(TagExistsResult);
    }

    // ---- ReadRaw -------------------------------------------------------------------------------
    public IReadOnlyList<HistorianSample> RawSamples = Array.Empty<HistorianSample>();
    public string? LastReadRawTag;
    public DateTime LastReadRawStartUtc;
    public DateTime LastReadRawEndUtc;
    public int LastReadRawMaxValues;
    public int ReadRawCallCount;
    /// <summary>
    ///     When set, ReadRaw serves <c>RawSource(start, end, maxValues)</c> instead of <see cref="RawSamples"/> (a
    ///     store-like fake honouring the window, #681).
    /// </summary>
    public Func<DateTime, DateTime, int, IEnumerable<HistorianSample>>? RawSource;
    /// <summary>Raw samples actually pulled by the consumer across every ReadRaw stream.</summary>
    public int RawYieldedCount;
    /// <summary>ReadRaw streams the consumer enumerated to the END (never broke out of — #681 review).</summary>
    public int RawStreamsCompleted;
    /// <summary>Every ReadRaw call's (start, end, maxValues), in order.</summary>
    public readonly List<(DateTime Start, DateTime End, int MaxValues)> RawCalls = new();

    // ---- ReadAggregate -------------------------------------------------------------------------
    public IReadOnlyList<HistorianAggregateSample> AggregateSamples = Array.Empty<HistorianAggregateSample>();
    public string? LastAggregateTag;
    public DateTime LastAggregateStartUtc;
    public DateTime LastAggregateEndUtc;
    public RetrievalMode LastAggregateMode;
    public TimeSpan LastAggregateInterval;
    public int ReadAggregateCallCount;

    // ---- ReadAtTime ----------------------------------------------------------------------------
    public IReadOnlyList<HistorianSample> AtTimeSamples = Array.Empty<HistorianSample>();
    public string? LastReadAtTimeTag;
    public IReadOnlyList<DateTime>? LastReadAtTimeTimestamps;
    public int ReadAtTimeCallCount;

    // ---- ReadEvents ----------------------------------------------------------------------------
    public IReadOnlyList<HistorianEvent> Events = Array.Empty<HistorianEvent>();
    public string? LastReadEventsSourceName;
    public DateTime LastReadEventsStartUtc;
    public DateTime LastReadEventsEndUtc;
    public int LastReadEventsMaxEvents;
    public int ReadEventsCallCount;

    /// <summary>When set, ReadEvents serves <c>EventsSource(start, end)</c> instead of <see cref="Events"/>.</summary>
    public Func<DateTime, DateTime, IEnumerable<HistorianEvent>>? EventsSource;

    /// <summary>Events actually pulled by the consumer across every ReadEvents stream.</summary>
    public int EventsYieldedCount;
    /// <summary>ReadEvents streams the consumer enumerated to the END (never broke out of — R08:1.2).</summary>
    public int EventStreamsCompleted;
    /// <summary>ReadEvents streams the consumer disposed before their end (an abandoned gateway stream).</summary>
    public int EventStreamsAbandoned;

    /// <summary>Thrown (deferred to first enumeration) by every read method when set.</summary>
    public Exception? ThrowOnRead;

    // ---- WriteLiveValues -----------------------------------------------------------------------
    public WriteAck WriteLiveValuesResult = new() { Success = true };
    public string? LastWriteLiveTag;
    public IReadOnlyList<HistorianLiveValue>? LastWriteLiveValues;
    public int WriteLiveValuesCallCount;
    public Exception? WriteLiveValuesThrows;

    // ---- SendEvent -----------------------------------------------------------------------------
    public WriteAck SendEventResult = new() { Success = true };
    public HistorianEvent? LastSendEvent;
    public int SendEventCallCount;
    public Exception? SendEventThrows;

    // ---- EnsureTags ----------------------------------------------------------------------------
    public TagOperationResults EnsureTagsResult = new();
    public IReadOnlyList<HistorianTagDefinition>? LastEnsureDefinitions;
    public int EnsureTagsCallCount;
    public Exception? EnsureTagsThrows;

    // ---- Probe ---------------------------------------------------------------------------------
    public bool ProbeResult = true;
    public int ProbeCallCount;
    public Exception? ProbeThrows;

    // ---- GetConnectionStatus -------------------------------------------------------------------
    public ConnectionStatus ConnectionStatus = new();
    public int GetConnectionStatusCallCount;
    public Exception? GetConnectionStatusThrows;

    // ---- Dispose -------------------------------------------------------------------------------
    public int DisposeCallCount;

    public IAsyncEnumerable<HistorianSample> ReadRawAsync(
        string tag,
        DateTime startUtc,
        DateTime endUtc,
        int maxValues,
        CancellationToken ct)
    {
        LastReadRawTag = tag;
        LastReadRawStartUtc = startUtc;
        LastReadRawEndUtc = endUtc;
        LastReadRawMaxValues = maxValues;
        lock (RawCalls)
        {
            ReadRawCallCount++;
            RawCalls.Add((startUtc, endUtc, maxValues));
        }
        if (MaxValuesCeiling is { } ceiling && maxValues > ceiling)
        {
            // The real gateway's refusal (HistorianReadService.ReadRaw), raised on the first MoveNext before any row, as
            // the client package surfaces it: a HistorianGatewayException wrapping the RpcException.
            var rpc = new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument,
                CeilingRejectionDetail ?? $"max_values exceeds the server ceiling of {ceiling}; read large ranges by time-slicing " +
                "(split the range and read the slices sequentially), or enable " +
                "Historian:ReadLimits:AllowStreamingBeyondCeiling to page through server-side."));
            return ToAsyncStream(Array.Empty<HistorianSample>(),
                new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException(rpc.Status.Detail, rpc), ct);
        }
        var items = RawSource is null ? RawSamples : RawSource(startUtc, endUtc, maxValues).ToList();
        return CountingStream(ToAsyncStream(items, ThrowOnRead, ct));
    }

    /// <summary>
    ///     A strict gateway (<c>Historian:ReadLimits:AllowStreamingBeyondCeiling=false</c>) with this
    ///     <c>MaxValuesCeiling</c>: a ReadRaw asking for more is refused <c>InvalidArgument</c> (#691.3).
    /// </summary>
    public int? MaxValuesCeiling;

    /// <summary>Overrides the refusal's detail text (an older/other gateway wording).</summary>
    public string? CeilingRejectionDetail;

    public IAsyncEnumerable<HistorianAggregateSample> ReadAggregateAsync(
        string tag,
        DateTime startUtc,
        DateTime endUtc,
        RetrievalMode mode,
        TimeSpan interval,
        CancellationToken ct)
    {
        LastAggregateTag = tag;
        LastAggregateStartUtc = startUtc;
        LastAggregateEndUtc = endUtc;
        LastAggregateMode = mode;
        LastAggregateInterval = interval;
        ReadAggregateCallCount++;
        AggregateCalls.Add((startUtc, endUtc, mode, interval));
        if (AggregateSource is null) return ToAsyncStream(AggregateSamples, ThrowOnRead, ct);
        if (startUtc > endUtc || interval <= TimeSpan.Zero)
        {
            // The real gateway's request validation (ValidateRange; interval must be > 0), raised before any row.
            var rpc = new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument,
                startUtc > endUtc ? "start_time must be less than or equal to end_time." : "interval must be greater than zero."));
            return ToAsyncStream(Array.Empty<HistorianAggregateSample>(),
                new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException(rpc.Status.Detail, rpc), ct);
        }
        return ToAsyncStream(AggregateSource(startUtc, endUtc, mode, interval).ToList(), ThrowOnRead, ct);
    }

    /// <summary>When set, ReadAggregate serves <c>AggregateSource(start, end, mode, interval)</c> (and refuses start &gt; end).</summary>
    public Func<DateTime, DateTime, RetrievalMode, TimeSpan, IEnumerable<HistorianAggregateSample>>? AggregateSource;

    /// <summary>Every ReadAggregate call's (start, end, mode, interval), in order.</summary>
    public readonly List<(DateTime Start, DateTime End, RetrievalMode Mode, TimeSpan Interval)> AggregateCalls = new();

    public Task<IReadOnlyList<HistorianSample>> ReadAtTimeAsync(
        string tag,
        IReadOnlyList<DateTime> timestampsUtc,
        CancellationToken ct)
    {
        LastReadAtTimeTag = tag;
        LastReadAtTimeTimestamps = timestampsUtc;
        ReadAtTimeCallCount++;
        return ThrowOnRead is not null
            ? Task.FromException<IReadOnlyList<HistorianSample>>(ThrowOnRead)
            : Task.FromResult(AtTimeSamples);
    }

    public IAsyncEnumerable<HistorianEvent> ReadEventsAsync(
        string? sourceName,
        DateTime startUtc,
        DateTime endUtc,
        int maxEvents,
        CancellationToken ct)
    {
        LastReadEventsSourceName = sourceName;
        LastReadEventsStartUtc = startUtc;
        LastReadEventsEndUtc = endUtc;
        LastReadEventsMaxEvents = maxEvents;
        ReadEventsCallCount++;
        lock (EventReadCaps) EventReadCaps.Add(maxEvents);
        if (MaxEventsCeiling is { } ceiling && maxEvents > ceiling)
        {
            // The real gateway's refusal (HistorianReadService.ReadEvents: max_events over MaxEventsCeiling), raised before
            // any row, as the client package surfaces it.
            var rpc = new Grpc.Core.RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.InvalidArgument,
                $"max_events exceeds the server ceiling of {ceiling}."));
            return CountingEventStream(ToAsyncStream(Array.Empty<HistorianEvent>(),
                new ZB.MOM.WW.HistorianGateway.Client.HistorianGatewayException(rpc.Status.Detail, rpc), ct));
        }
        IReadOnlyList<HistorianEvent> items = EventsSource is null ? Events : EventsSource(startUtc, endUtc).ToList();
        // A current gateway stops at max_events and completes the stream normally (historiangw#53); IgnoreMaxEvents
        // simulates an older gateway that streams everything.
        if (maxEvents > 0 && !IgnoreMaxEvents && items.Count > maxEvents) items = items.Take(maxEvents).ToList();
        return CountingEventStream(ToAsyncStream(items, ThrowOnRead, ct));
    }

    /// <summary>Simulates a gateway older than <c>max_events</c>: every event in the window is streamed.</summary>
    public bool IgnoreMaxEvents;

    /// <summary>When set, a ReadEvents whose max_events exceeds it is refused <c>InvalidArgument</c> (MaxEventsCeiling).</summary>
    public int? MaxEventsCeiling;

    /// <summary>Every ReadEvents call's maxEvents (the server-side max_events sent), in order.</summary>
    public readonly List<int> EventReadCaps = new();

    // ---- SendEvents ----------------------------------------------------------------------------
    /// <summary>
    ///     When set, SendEvents is served by it. Unset, SendEvents answers <c>Unimplemented</c> (the seam's default — an
    ///     older gateway), so the writer falls back to SendEvent.
    /// </summary>
    public Func<IReadOnlyList<HistorianEvent>, CancellationToken, Task<SendEventsReply>>? SendEventsSource;

    /// <summary>Every SendEvents call's events, in order.</summary>
    public readonly List<IReadOnlyList<HistorianEvent>> SendEventsCalls = new();

    public Task<SendEventsReply> SendEventsAsync(IReadOnlyList<HistorianEvent> events, CancellationToken ct)
    {
        lock (SendEventsCalls) SendEventsCalls.Add(events);
        if (SendEventsSource is { } source) return source(events, ct);
        return Task.FromException<SendEventsReply>(new Grpc.Core.RpcException(
            new Grpc.Core.Status(Grpc.Core.StatusCode.Unimplemented, "Method is unimplemented.")));
    }

    public Task<WriteAck> WriteLiveValuesAsync(
        string tag,
        IReadOnlyList<HistorianLiveValue> values,
        CancellationToken ct)
    {
        LastWriteLiveTag = tag;
        LastWriteLiveValues = values;
        WriteLiveValuesCallCount++;
        return WriteLiveValuesThrows is not null
            ? Task.FromException<WriteAck>(WriteLiveValuesThrows)
            : Task.FromResult(WriteLiveValuesResult);
    }

    /// <summary>When set, SendEvent is served by it (an asynchronous, observable gateway — concurrency tests).</summary>
    public Func<HistorianEvent, CancellationToken, Task<WriteAck>>? SendEventSource;

    public Task<WriteAck> SendEventAsync(HistorianEvent evt, CancellationToken ct)
    {
        if (SendEventSource is { } source)
        {
            Interlocked.Increment(ref SendEventCallCount);
            return source(evt, ct);
        }
        LastSendEvent = evt;
        SendEventCallCount++;
        return SendEventThrows is not null
            ? Task.FromException<WriteAck>(SendEventThrows)
            : Task.FromResult(SendEventResult);
    }

    public Task<TagOperationResults> EnsureTagsAsync(
        IReadOnlyList<HistorianTagDefinition> definitions,
        CancellationToken ct)
    {
        LastEnsureDefinitions = definitions;
        EnsureTagsCallCount++;
        return EnsureTagsThrows is not null
            ? Task.FromException<TagOperationResults>(EnsureTagsThrows)
            : Task.FromResult(EnsureTagsResult);
    }

    public Task<bool> ProbeAsync(CancellationToken ct)
    {
        ProbeCallCount++;
        return ProbeThrows is not null
            ? Task.FromException<bool>(ProbeThrows)
            : Task.FromResult(ProbeResult);
    }

    public Task<ConnectionStatus> GetConnectionStatusAsync(CancellationToken ct)
    {
        GetConnectionStatusCallCount++;
        return GetConnectionStatusThrows is not null
            ? Task.FromException<ConnectionStatus>(GetConnectionStatusThrows)
            : Task.FromResult(ConnectionStatus);
    }

    public ValueTask DisposeAsync()
    {
        DisposeCallCount++;
        return ValueTask.CompletedTask;
    }

    private async IAsyncEnumerable<HistorianSample> CountingStream(IAsyncEnumerable<HistorianSample> inner)
    {
        await foreach (var item in inner.ConfigureAwait(false))
        {
            RawYieldedCount++;
            yield return item;
        }
        RawStreamsCompleted++;
    }

    private async IAsyncEnumerable<HistorianEvent> CountingEventStream(IAsyncEnumerable<HistorianEvent> inner)
    {
        var completed = false;
        try
        {
            await foreach (var item in inner.ConfigureAwait(false))
            {
                Interlocked.Increment(ref EventsYieldedCount);
                yield return item;
            }
            completed = true;
            Interlocked.Increment(ref EventStreamsCompleted);
        }
        finally
        {
            // A consumer that breaks out (or a fault) disposes the stream early: on the real gateway that cancels the RPC.
            if (!completed) Interlocked.Increment(ref EventStreamsAbandoned);
        }
    }

    private static async IAsyncEnumerable<T> ToAsyncStream<T>(
        IReadOnlyList<T> items,
        Exception? error,
        [EnumeratorCancellation] CancellationToken ct)
    {
        if (error is not null)
        {
            throw error;
        }

        foreach (var item in items)
        {
            ct.ThrowIfCancellationRequested();
            yield return item;
        }

        await ValueTask.CompletedTask.ConfigureAwait(false);
    }
}