mxaccessgw/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs

using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Alarms;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Grpc;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;

namespace ZB.MOM.WW.MxGateway.Tests.Alarms;

/// <summary>
///     Carries the worker's truncated-fetch verdict across the gateway: worker
///     reply payload → <see cref="GatewayAlarmMonitor"/> → the public
///     <c>QueryActiveAlarms</c> stream.
/// </summary>
/// <remarks>
///     <para>
///         The truncation guard itself (the worker merging rather than replacing
///         a capped snapshot) is already covered in the worker suite. What was
///         missing is that the guard is <em>silent</em>: a capped fetch suppresses
///         absence-implies-Clear inference and says so only in a rate-limited
///         stderr warning, so a consumer of the alarm surface could not tell a
///         complete active set from a capped one. These tests pin the structural
///         signal that replaces the guesswork.
///     </para>
///     <para>
///         The load-bearing assertion is the <em>false</em> one
///         (<see cref="QueryActiveAlarms_WithCompleteWorkerReply_LeavesFlagUnset"/>).
///         "Truncated reply sets the flag" would also pass against a field
///         hard-wired to true; only the complete-reply case proves the flag is
///         actually derived from the worker's verdict.
///     </para>
/// </remarks>
public sealed class AlarmTruncationSignalTests
{
    private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(15);

    /// <summary>
    ///     A capped worker reply sets the monitor's completeness caveat and
    ///     stamps every cached snapshot, so both the dashboard (which reads the
    ///     service flag) and the RPC (which reads the records) can surface it.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_WithTruncatedWorkerReply_SurfacesTheFlagOnMonitorAndSnapshots()
    {
        using GatewayMetrics metrics = new();
        StubSessionManager sessions = new()
        {
            SnapshotTruncated = true,
            Snapshots = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: true)],
        };
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForReconcileAsync(WaitTimeout);
        await WaitUntilAsync(() => monitor.CurrentAlarms.Count == 1, WaitTimeout);

        Assert.True(monitor.SnapshotTruncated);
        ActiveAlarmSnapshot cached = Assert.Single(monitor.CurrentAlarms);
        Assert.True(cached.FromTruncatedSnapshot);
        // Truncation is about the completeness of the SET, not the fidelity of
        // the record — the subtag-fallback flag must stay independent of it.
        Assert.False(cached.Degraded);

        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    /// <summary>
    ///     The public <c>QueryActiveAlarms</c> stream carries the per-record flag
    ///     through untouched. That RPC returns a bare
    ///     <c>stream ActiveAlarmSnapshot</c> with no envelope message, so the
    ///     per-record boolean is the only place set-level degraded status can
    ///     ride — if the service ever starts re-projecting records instead of
    ///     forwarding them, this is what catches the dropped field.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task QueryActiveAlarms_WithTruncatedSnapshot_StreamsTheFlagToTheClient()
    {
        FakeGatewayAlarmService alarms = new()
        {
            SnapshotTruncated = true,
            CurrentAlarms = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: true)],
        };
        MxAccessGatewayService service = CreateService(alarms);
        RecordingServerStreamWriter<ActiveAlarmSnapshot> sink = new();

        await service.QueryActiveAlarms(
            new QueryActiveAlarmsRequest(),
            sink,
            new TestServerCallContext());

        ActiveAlarmSnapshot streamed = Assert.Single(sink.Messages);
        Assert.True(streamed.FromTruncatedSnapshot);
        Assert.Equal("Galaxy!Area.Tank01.Level.HiHi", streamed.AlarmFullReference);
    }

    /// <summary>
    ///     The control. A complete worker reply must leave both the monitor flag
    ///     and the streamed records unset — otherwise every snapshot would read
    ///     as possibly-incomplete and the signal would carry no information.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task QueryActiveAlarms_WithCompleteWorkerReply_LeavesFlagUnset()
    {
        using GatewayMetrics metrics = new();
        StubSessionManager sessions = new()
        {
            SnapshotTruncated = false,
            Snapshots = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: false)],
        };
        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);

        using CancellationTokenSource cts = new();
        await monitor.StartAsync(cts.Token);
        await sessions.WaitForReconcileAsync(WaitTimeout);
        await WaitUntilAsync(() => monitor.CurrentAlarms.Count == 1, WaitTimeout);

        Assert.False(monitor.SnapshotTruncated);

        MxAccessGatewayService service = CreateService(new FakeGatewayAlarmService
        {
            SnapshotTruncated = monitor.SnapshotTruncated,
            CurrentAlarms = monitor.CurrentAlarms,
        });
        RecordingServerStreamWriter<ActiveAlarmSnapshot> sink = new();

        await service.QueryActiveAlarms(
            new QueryActiveAlarmsRequest(),
            sink,
            new TestServerCallContext());

        Assert.False(Assert.Single(sink.Messages).FromTruncatedSnapshot);

        await cts.CancelAsync();
        await monitor.StopAsync(CancellationToken.None);
    }

    private static ActiveAlarmSnapshot NewSnapshot(string reference, bool fromTruncatedSnapshot)
    {
        return new ActiveAlarmSnapshot
        {
            AlarmFullReference = reference,
            SourceObjectReference = "Tank01.Level",
            AlarmTypeName = "HiHi",
            Category = "Area",
            Severity = 500,
            CurrentState = AlarmConditionState.Active,
            SourceProvider = AlarmProviderMode.Alarmmgr,
            FromTruncatedSnapshot = fromTruncatedSnapshot,
        };
    }

    private static GatewayAlarmMonitor CreateMonitor(StubSessionManager sessions, GatewayMetrics metrics)
    {
        AlarmsOptions options = new()
        {
            Enabled = true,
            SubscriptionExpression = @"\\NODE\Galaxy!Area",
        };
        return new GatewayAlarmMonitor(
            sessions,
            new StubWatchListResolver(),
            metrics,
            Microsoft.Extensions.Options.Options.Create(new GatewayOptions { Alarms = options }),
            NullLogger<GatewayAlarmMonitor>.Instance);
    }

    private static MxAccessGatewayService CreateService(FakeGatewayAlarmService alarms)
    {
        StubSessionManager sessions = new();
        return new MxAccessGatewayService(
            sessions,
            new GatewayRequestIdentityAccessor(),
            new AllowAllConstraintEnforcer(),
            new MxAccessGrpcRequestValidator(),
            new MxAccessGrpcMapper(),
            new StubEventStreamService(),
            new GatewayMetrics(),
            NullLogger<MxAccessGatewayService>.Instance,
            alarms);
    }

    private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
    {
        DateTime deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (condition())
            {
                return;
            }

            await Task.Delay(25);
        }

        throw new TimeoutException("Condition was not met in time.");
    }

    /// <summary><see cref="IAlarmWatchListResolver"/> that resolves an empty watch-list.</summary>
    private sealed class StubWatchListResolver : IAlarmWatchListResolver
    {
        /// <inheritdoc />
        public Task<IReadOnlyList<AlarmSubtagTarget>> ResolveAsync(
            AlarmsOptions options,
            CancellationToken cancellationToken = default) =>
            Task.FromResult<IReadOnlyList<AlarmSubtagTarget>>([]);
    }

    /// <summary>
    ///     Minimal <see cref="ISessionManager"/> that answers the monitor's
    ///     QueryActiveAlarms with a scripted reply payload — the seam this suite
    ///     drives the truncation verdict through.
    /// </summary>
    private sealed class StubSessionManager : ISessionManager
    {
        private readonly Channel<WorkerEvent> _events = Channel.CreateUnbounded<WorkerEvent>();
        private readonly TaskCompletionSource _reconciled =
            new(TaskCreationOptions.RunContinuationsAsynchronously);

        /// <summary>Gets or sets the truncation verdict the scripted reply carries.</summary>
        public bool SnapshotTruncated { get; init; }

        /// <summary>Gets or sets the snapshots the scripted reply carries.</summary>
        public IReadOnlyList<ActiveAlarmSnapshot> Snapshots { get; init; } = [];

        /// <summary>Completes once the monitor has issued its first QueryActiveAlarms.</summary>
        /// <param name="timeout">The maximum time to wait.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public Task WaitForReconcileAsync(TimeSpan timeout) => _reconciled.Task.WaitAsync(timeout);

        /// <inheritdoc />
        public Task<GatewaySession> OpenSessionAsync(
            SessionOpenRequest request,
            string? clientIdentity,
            string? ownerKeyId,
            CancellationToken cancellationToken)
        {
            GatewaySession session = new(
                Guid.NewGuid().ToString("N"),
                "Galaxy",
                "pipe-test",
                "nonce-test",
                clientIdentity,
                null,
                null,
                TimeSpan.FromSeconds(30),
                TimeSpan.FromSeconds(30),
                TimeSpan.FromSeconds(30),
                DateTimeOffset.UtcNow);
            session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
            session.MarkReady();
            return Task.FromResult(session);
        }

        /// <inheritdoc />
        public Task<WorkerCommandReply> InvokeAsync(
            string sessionId,
            WorkerCommand command,
            CancellationToken cancellationToken)
        {
            MxCommandReply reply = new()
            {
                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
            };

            if (command.Command?.Kind == MxCommandKind.QueryActiveAlarms)
            {
                QueryActiveAlarmsReplyPayload payload = new() { SnapshotTruncated = SnapshotTruncated };
                payload.Snapshots.AddRange(Snapshots.Select(snapshot => snapshot.Clone()));
                reply.QueryActiveAlarms = payload;
                _reconciled.TrySetResult();
            }

            return Task.FromResult(new WorkerCommandReply { Reply = reply });
        }

        /// <inheritdoc />
        public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
        {
            session = null;
            return false;
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(string sessionId, CancellationToken cancellationToken)
        {
            _events.Writer.TryComplete();
            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> KillWorkerAsync(string sessionId, string reason, CancellationToken cancellationToken) =>
            Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));

        /// <inheritdoc />
        public Task<int> CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) =>
            Task.FromResult(0);

        /// <inheritdoc />
        public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
    }

    /// <summary>
    ///     <see cref="IEventStreamService"/> stub — QueryActiveAlarms never
    ///     touches the event path, but the service constructor requires one.
    /// </summary>
    private sealed class StubEventStreamService : IEventStreamService
    {
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
            StreamEventsRequest request,
            string? callerKeyId,
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await Task.CompletedTask.ConfigureAwait(false);
            yield break;
        }
    }
}