lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientGateSplitTests.cs

using System.Text.Json;
using Moq;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;

/// <summary>
///     Review 2026-10-21 R06:O-6 — one <c>SemaphoreSlim</c> serialised reads, writes, subscribes, acks AND the
///     whole recursive discovery browse, so a stalled write (up to the operation timeout) blocked every read and
///     subscription create, and a discovery browse blocked the data plane for its full duration. Reads (and acks)
///     keep their gate; writes and the discovery browse take none (the SDK session is thread-safe for concurrent
///     service calls); subscription management has its own gate.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientGateSplitTests
{
    private const string SpeedRef = "Plant/dev/Speed";
    private static readonly TimeSpan Bound = TimeSpan.FromSeconds(10);

    private static readonly IReadOnlyList<RawTagEntry> Tags =
        [new(SpeedRef, JsonSerializer.Serialize(new { nodeId = "ns=0;s=Speed" }), WriteIdempotent: true)];

    private static (OpcUaClientDriver Driver, Mock<ISession> Session) NewConnectedDriver()
    {
        var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-o6");
        var session = new Mock<ISession>(MockBehavior.Loose);
        session.SetupGet(s => s.MessageContext).Returns(new ServiceMessageContext(SdkTelemetry.Context));
        session.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable());
        session.Setup(s => s.ReadAsync(
                It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
                It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
            .ReturnsAsync((RequestHeader _, double _, TimestampsToReturn _, ReadValueIdCollection nodes, CancellationToken _) =>
                new ReadResponse
                {
                    ResponseHeader = new ResponseHeader(),
                    Results = [.. nodes.Select(_ => new DataValue(new Variant(42), StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow))],
                    DiagnosticInfos = [],
                });
        drv.SetSessionForTest(session.Object);
        drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags));
        drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
        return (drv, session);
    }

    private static async Task ReadCompletes(OpcUaClientDriver drv, string because)
    {
        var result = await drv.ReadAsync([SpeedRef], TestContext.Current.CancellationToken).WaitAsync(Bound, TestContext.Current.CancellationToken);
        result.Single().Value.ShouldBe(42, because);
    }

    /// <summary>A write stalled on the wire does not hold up a read.</summary>
    [Fact]
    public async Task A_stalled_write_does_not_block_a_read()
    {
        var (drv, session) = NewConnectedDriver();
        using var _ = drv;
        var writeOnWire = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var releaseWrite = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        session.Setup(s => s.WriteAsync(It.IsAny<RequestHeader>(), It.IsAny<WriteValueCollection>(), It.IsAny<CancellationToken>()))
            .Returns(async (RequestHeader _, WriteValueCollection nodes, CancellationToken _) =>
            {
                writeOnWire.TrySetResult();
                await releaseWrite.Task;
                return new WriteResponse
                {
                    ResponseHeader = new ResponseHeader(),
                    Results = [.. nodes.Select(_ => (StatusCode)StatusCodes.Good)],
                    DiagnosticInfos = [],
                };
            });

        var write = drv.WriteAsync([new ZB.MOM.WW.OtOpcUa.Core.Abstractions.WriteRequest(SpeedRef, 7)], TestContext.Current.CancellationToken);
        await writeOnWire.Task.WaitAsync(Bound, TestContext.Current.CancellationToken);

        await ReadCompletes(drv, "a read must not queue behind a stalled write");

        releaseWrite.SetResult();
        (await write).Single().StatusCode.ShouldBe(0u);
    }

    /// <summary>Subscription management queues on its own gate, not the read gate.</summary>
    [Fact]
    public async Task Subscription_management_does_not_share_the_read_gate()
    {
        var (drv, _) = NewConnectedDriver();
        using var __ = drv;

        await drv.SubscriptionGate.WaitAsync(TestContext.Current.CancellationToken);
        var subscribe = drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);

        // Positive control: the subscribe really is parked on the subscription gate.
        subscribe.IsCompleted.ShouldBeFalse();
        await ReadCompletes(drv, "a read must not queue behind subscription management");

        drv.SubscriptionGate.Release();
        // The mock never attaches the subscription, so the create itself throws — what matters is that it ran.
        await Should.ThrowAsync<Exception>(() => subscribe.WaitAsync(Bound, TestContext.Current.CancellationToken));
    }

    /// <summary>A discovery browse in progress does not hold the read gate.</summary>
    [Fact]
    public async Task A_discovery_browse_in_progress_does_not_block_a_read()
    {
        var (drv, session) = NewConnectedDriver();
        using var _ = drv;
        var browseOnWire = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var releaseBrowse = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        session.Setup(s => s.BrowseAsync(
                It.IsAny<RequestHeader>(), It.IsAny<ViewDescription>(), It.IsAny<uint>(),
                It.IsAny<BrowseDescriptionCollection>(), It.IsAny<CancellationToken>()))
            .Returns(async (RequestHeader _, ViewDescription _, uint _, BrowseDescriptionCollection _, CancellationToken _) =>
            {
                browseOnWire.TrySetResult();
                await releaseBrowse.Task;
                return new BrowseResponse
                {
                    ResponseHeader = new ResponseHeader(),
                    Results = [new BrowseResult { References = [], ContinuationPoint = [] }],
                    DiagnosticInfos = [],
                };
            });

        var discover = drv.DiscoverAsync(new NullAddressSpaceBuilder(), TestContext.Current.CancellationToken);
        await browseOnWire.Task.WaitAsync(Bound, TestContext.Current.CancellationToken);

        await ReadCompletes(drv, "a read must not queue behind a recursive discovery browse");

        releaseBrowse.SetResult();
        await discover.WaitAsync(Bound, TestContext.Current.CancellationToken);
    }

    private sealed class NullAddressSpaceBuilder : IAddressSpaceBuilder
    {
        public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;

        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
            => new StubHandle();

        public void AddProperty(string browseName, DriverDataType dataType, object? value) { }

        public void AttachAlarmCondition(IVariableHandle sourceVariable, string alarmName, DriverAttributeInfo alarmInfo) { }

        private sealed class StubHandle : IVariableHandle
        {
            public string FullReference => "stub";
            public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => throw new NotSupportedException();
        }
    }
}