lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs

using System.Text.Json;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
///     Wire-level coverage of the real <see cref="OpcUaClientDriver"/> against the in-process OtOpcUa server
///     (<see cref="ServerEventRig"/>), replacing the "lands in a follow-up PR once we have an in-process server
///     fixture" scaffolds in <c>Driver.OpcUaClient.Tests</c> (Gitea #622). No opc-plc and no Docker: every test
///     here runs a real secure-channel + session + subscription exchange over loopback. The driver authenticates
///     with the rig's API-key credential so writes carry the <c>WriteOperate</c> role.
///     HistoryRead lives in <see cref="OpcUaClientDriverHistoryWireTests"/>; reconnect/resume in
///     <see cref="OpcUaClientDriverReconnectWireTests"/>.
/// </summary>
[Trait("Category", "Integration")]
public sealed class OpcUaClientDriverWireTests
{
    internal const string SpeedRef = "Sim/dev1/Speed";
    internal const string RoRef = "Sim/dev1/Counter";
    private const string MissingRef = "Sim/dev1/Missing";
    internal const string SpeedId = ServerEventRig.RawDeviceFolder + "/Speed";
    internal const string CounterId = ServerEventRig.RawDeviceFolder + "/Counter";

    private sealed class RecordingWriteGateway(bool succeed) : IOpcUaNodeWriteGateway
    {
        public List<(string NodeId, object? Value)> Writes { get; } = [];

        public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct)
        {
            lock (Writes) Writes.Add((nodeId, value));
            return Task.FromResult(new NodeWriteOutcome(succeed, succeed ? null : "device refused"));
        }
    }

    internal static RawTagEntry Tag(string rawPath, string nodeId) =>
        new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true);

    internal static OpcUaClientDriver NewDriver(
        ServerEventRig rig, string id, string? browseRoot = null, IReadOnlyList<RawTagEntry>? extraTags = null,
        string? endpointUrl = null, TimeSpan? timeout = null, TimeSpan? sessionTimeout = null,
        uint? subscriptionLifetimeCount = null) =>
        new(new OpcUaClientDriverOptions
        {
            EndpointUrl = endpointUrl ?? rig.EndpointUrl,
            RawTags =
            [
                Tag(SpeedRef, ServerEventRig.RawNsu(SpeedId)),
                Tag(RoRef, ServerEventRig.RawNsu(CounterId)),
                Tag(MissingRef, ServerEventRig.RawNsu($"{ServerEventRig.RawDeviceFolder}/NoSuchNode")),
                .. extraTags ?? [],
            ],
            SecurityPolicy = OpcUaSecurityPolicy.None,
            SecurityMode = OpcUaSecurityMode.None,
            AuthType = OpcUaAuthType.Username,
            Username = "apikey",
            Password = ServerEventRig.KeyPassword,
            AutoAcceptCertificates = true,
            Timeout = timeout ?? TimeSpan.FromSeconds(10),
            SessionTimeout = sessionTimeout ?? TimeSpan.FromSeconds(30),
            SubscriptionLifetimeCount = subscriptionLifetimeCount ?? 1000,
            KeepAliveInterval = TimeSpan.FromSeconds(1),
            BrowseRoot = browseRoot,
        }, id);

    internal static async Task<ServerEventRig> StartRigAsync(string name, CancellationToken ct)
    {
        var rig = await ServerEventRig.StartAsync(name, encrypted: false, ct);
        rig.EnsureRawVariable("Speed", "Float", writable: true);
        rig.EnsureRawVariable("Counter", "Int32", writable: false);
        return rig;
    }

    internal static void Publish(ServerEventRig rig, string id, object value) =>
        rig.Sink.WriteValue(id, value, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);

    /// <summary>Reads round-trip through the real wire; a node the server lacks does not read Good.</summary>
    [Fact]
    public async Task Read_returns_the_published_value_and_a_bad_status_for_a_missing_node()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-read", ct);
        Publish(rig, SpeedId, 42.5f);

        await using var drv = NewDriver(rig, "wire-read");
        await drv.InitializeAsync("{}", ct);

        var snaps = await drv.ReadAsync([SpeedRef, MissingRef], ct);

        snaps.Count.ShouldBe(2);
        snaps[0].StatusCode.ShouldBe(0u);
        Convert.ToSingle(snaps[0].Value).ShouldBe(42.5f);
        snaps[1].StatusCode.ShouldNotBe(0u, "a node the server does not have must not read Good");
    }

    /// <summary>A write the device accepts lands on the node and reads back.</summary>
    [Fact]
    public async Task Write_roundtrips_through_the_node_write_gateway_and_reads_back()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-write", ct);
        var gateway = new RecordingWriteGateway(succeed: true);
        rig.Nm.NodeWriteGateway = gateway;
        Publish(rig, SpeedId, 1.0f);

        await using var drv = NewDriver(rig, "wire-write");
        await drv.InitializeAsync("{}", ct);

        var results = await drv.WriteAsync([new WriteRequest(SpeedRef, 7.25f)], ct);

        results.Single().StatusCode.ShouldBe(0u);
        lock (gateway.Writes) Convert.ToSingle(gateway.Writes.Single().Value).ShouldBe(7.25f);
        var back = await drv.ReadAsync([SpeedRef], ct);
        Convert.ToSingle(back[0].Value).ShouldBe(7.25f);
    }

    /// <summary>A write to a read-only node is refused by the server and never reaches the gateway.</summary>
    [Fact]
    public async Task Write_to_a_read_only_node_is_refused_and_never_reaches_the_gateway()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-write-ro", ct);
        var gateway = new RecordingWriteGateway(succeed: true);
        rig.Nm.NodeWriteGateway = gateway;
        Publish(rig, CounterId, 5);

        await using var drv = NewDriver(rig, "wire-write-ro");
        await drv.InitializeAsync("{}", ct);

        var results = await drv.WriteAsync([new WriteRequest(RoRef, 9)], ct);

        results.Single().StatusCode.ShouldNotBe(0u);
        lock (gateway.Writes) gateway.Writes.ShouldBeEmpty();
        // Positive control: the same session CAN write the writable node, so the refusal above is the node's.
        (await drv.WriteAsync([new WriteRequest(SpeedRef, 2.0f)], ct)).Single().StatusCode.ShouldBe(0u);
        lock (gateway.Writes) gateway.Writes.Count.ShouldBe(1);
    }

    /// <summary>The node manager's write-outcome self-correction is visible to the driver as a revert.</summary>
    [Fact]
    public async Task A_write_the_device_refuses_is_reverted_and_the_driver_reads_the_original_value()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-write-fail", ct);
        rig.Nm.NodeWriteGateway = new RecordingWriteGateway(succeed: false);
        Publish(rig, SpeedId, 3.0f);

        await using var drv = NewDriver(rig, "wire-write-fail");
        await drv.InitializeAsync("{}", ct);

        await drv.WriteAsync([new WriteRequest(SpeedRef, 99.0f)], ct);

        var reverted = false;
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
        while (DateTime.UtcNow < deadline && !reverted)
        {
            var snap = (await drv.ReadAsync([SpeedRef], ct))[0];
            reverted = Math.Abs(Convert.ToSingle(snap.Value) - 3.0f) < 0.001f;
            if (!reverted) await Task.Delay(50, ct);
        }
        reverted.ShouldBeTrue("the refused write must revert the node to its pre-write value");
    }

    /// <summary>A subscription delivers the initial value and then later server-side changes, and stops after unsubscribe.</summary>
    [Fact]
    public async Task Subscribe_delivers_the_initial_value_and_subsequent_changes_until_unsubscribed()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-sub", ct);
        Publish(rig, SpeedId, 10.0f);

        await using var drv = NewDriver(rig, "wire-sub");
        await drv.InitializeAsync("{}", ct);

        var seen = new List<float>();
        drv.OnDataChange += (_, e) =>
        {
            if (e.FullReference == SpeedRef && e.Snapshot.StatusCode == 0u)
                lock (seen) seen.Add(Convert.ToSingle(e.Snapshot.Value));
        };

        var handle = await drv.SubscribeAsync([SpeedRef], TimeSpan.FromMilliseconds(100), ct);
        await RecordedEvents.WaitUntilAsync(() => { lock (seen) return seen.Contains(10.0f); });

        Publish(rig, SpeedId, 20.0f);
        await RecordedEvents.WaitUntilAsync(() => { lock (seen) return seen.Contains(20.0f); });

        await drv.UnsubscribeAsync(handle, ct);

        // Absence needs a positive control: after unsubscribing, publish 30.0 and prove a FRESH subscription
        // delivers it exactly once, i.e. the dead first subscription contributed nothing.
        Publish(rig, SpeedId, 30.0f);
        var control = await drv.SubscribeAsync([SpeedRef], TimeSpan.FromMilliseconds(100), ct);
        await RecordedEvents.WaitUntilAsync(() => { lock (seen) return seen.Contains(30.0f); });
        await Task.Delay(500, ct);
        await drv.UnsubscribeAsync(control, ct);
        lock (seen) seen.Count(v => v == 30.0f).ShouldBe(1, "only the control subscription delivers 30.0");
    }

    /// <summary>The host probe reports Running while the server is up and leaves Running when it goes away.</summary>
    [Fact]
    public async Task Host_status_is_Running_then_leaves_Running_when_the_server_stops()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await StartRigAsync("drv-probe", ct);
        await using var drv = NewDriver(rig, "wire-probe");
        try
        {
            await drv.InitializeAsync("{}", ct);
            drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running);

            var transitioned = new TaskCompletionSource<HostState>(TaskCreationOptions.RunContinuationsAsynchronously);
            drv.OnHostStatusChanged += (_, e) =>
            {
                if (e.NewState != HostState.Running) transitioned.TrySetResult(e.NewState);
            };

            await rig.DisposeAsync();
            rig = null;

            var state = await transitioned.Task.WaitAsync(TimeSpan.FromSeconds(30), ct);
            state.ShouldNotBe(HostState.Running);
            drv.GetHostStatuses().Single().State.ShouldNotBe(HostState.Running);
        }
        finally
        {
            if (rig is not null) await rig.DisposeAsync();
        }
    }

    /// <summary>Discovery browses the live server's variables into the builder with their real data types.</summary>
    [Fact]
    public async Task Discovery_browses_the_servers_variables_into_the_builder()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-disco", ct);
        Publish(rig, SpeedId, 1.0f);

        await using var drv = NewDriver(rig, "wire-disco", browseRoot: ServerEventRig.RawNsu(ServerEventRig.RawDeviceFolder));
        await drv.InitializeAsync("{}", ct);

        var builder = new RecordingBuilder();
        await drv.DiscoverAsync(builder, ct);

        // The server's browse name for a raw variable is its full RawPath, so key on the stable nsu= reference.
        var byRef = builder.Variables.ToDictionary(v => v.Info.FullName);
        var speed = byRef[ServerEventRig.RawNsu(SpeedId)].Info;
        var counter = byRef[ServerEventRig.RawNsu(CounterId)].Info;
        speed.DriverDataType.ShouldBe(DriverDataType.Float32);
        counter.DriverDataType.ShouldBe(DriverDataType.Int32);
        speed.SecurityClass.ShouldNotBe(counter.SecurityClass,
            "the writable variable and the read-only one must classify differently");
    }

    /// <summary>
    ///     #619 — ConditionRefresh over the wire: a condition already active+unacked BEFORE the driver subscribes is
    ///     reported (Retrigger since #647, keyed by the RawPath) right after subscribe, not only on its next transition. A second
    ///     subscribe of the same set refreshes again and must emit nothing (no re-raise); a positive control then
    ///     proves the feed is still live by acknowledging upstream and seeing exactly the Acknowledge arrive.
    /// </summary>
    [Fact]
    public async Task A_condition_already_active_before_subscribe_is_reported_by_ConditionRefresh()
    {
        const string AlarmRef = "Sim/dev1/TempHi";
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await StartRigAsync("drv-alarm-refresh", ct);
        rig.MaterialiseNative();
        rig.Project(native: true, new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true,
            Enabled: true, AlarmShelvingKind.Unshelved, 800, "already hot"));

        await using var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions
        {
            EndpointUrl = rig.EndpointUrl,
            RawTags = [Tag(AlarmRef, ServerEventRig.RawNsu(ServerEventRig.RawAlarmPath))],
            SecurityPolicy = OpcUaSecurityPolicy.None,
            SecurityMode = OpcUaSecurityMode.None,
            AuthType = OpcUaAuthType.Username,
            Username = "apikey",
            Password = ServerEventRig.KeyPassword,
            AutoAcceptCertificates = true,
            Timeout = TimeSpan.FromSeconds(10),
            SessionTimeout = TimeSpan.FromSeconds(30),
            KeepAliveInterval = TimeSpan.FromSeconds(1),
        }, "wire-alarm-refresh");
        await drv.InitializeAsync("{}", ct);

        var seen = new List<AlarmEventArgs>();
        drv.OnAlarmEvent += (_, e) => { lock (seen) seen.Add(e); };
        AlarmEventArgs[] Seen() { lock (seen) return [.. seen]; }

        var first = await drv.SubscribeAlarmsAsync([AlarmRef], ct);
        // First sight of an active+UNACKED upstream condition is a Retrigger (#647 review): the host keeps an ack
        // across a Raise on an already-active condition, so only a Retrigger forces the upstream's unack.
        await RecordedEvents.WaitUntilAsync(() => Seen().Any(e => e.Kind == AlarmTransitionKind.Retrigger));

        var raise = Seen().Single(e => e.Kind == AlarmTransitionKind.Retrigger);
        raise.ConditionId.ShouldBe(AlarmRef);
        raise.Message.ShouldBe("already hot");
        raise.OpcUaSeverity.ShouldBe(800);
        Seen().ShouldAllBe(e => e.ConditionId == AlarmRef, "refresh markers must never surface as alarms");

        // Same set again, released first exactly as DriverInstanceActor does → a fresh subscription and another
        // ConditionRefresh; the identical state must not re-raise.
        var before = Seen().Length;
        await drv.UnsubscribeAlarmsAsync(first, ct);
        await drv.SubscribeAlarmsAsync([AlarmRef], ct);
        await Task.Delay(1500, ct); // let the second refresh's events land before the control transition

        // Positive control: an upstream acknowledge arrives as exactly one Acknowledge, after which nothing else.
        rig.Project(native: true, new AlarmConditionSnapshot(Active: true, Acknowledged: true, Confirmed: true,
            Enabled: true, AlarmShelvingKind.Unshelved, 800, "already hot"));
        await RecordedEvents.WaitUntilAsync(() => Seen().Any(e => e.Kind == AlarmTransitionKind.Acknowledge));
        await Task.Delay(500, ct);
        // #728: record each event's full shape on failure, not only its Kind — the extra Unspecified that failed CI
        // could be a changed message/severity, a non-condition event or a quality-only annotation, and the Kind alone
        // cannot tell them apart.
        var after = Seen().Skip(before).ToArray();
        after.Select(e => e.Kind).ShouldBe([AlarmTransitionKind.Acknowledge],
            "the second refresh re-delivered an unchanged state and must have emitted nothing; events after the " +
            "resubscribe: " + string.Join(" | ", after.Select(Describe)));
    }

    /// <summary>One alarm event's distinguishing fields, for a failure message.</summary>
    private static string Describe(AlarmEventArgs e) =>
        $"Kind={e.Kind} AlarmType={e.AlarmType} Message='{e.Message}' Severity={e.Severity}/{e.OpcUaSeverity?.ToString() ?? "-"} " +
        $"Quality={(e.QualityStatusCode is { } q ? $"0x{q:X8}" : "-")} Restatement={e.IsRestatement} " +
        $"Condition={e.ConditionId} Source={e.SourceTimestampUtc:HH:mm:ss.fff}";

    private sealed class RecordingBuilder : IAddressSpaceBuilder
    {
        public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = [];

        public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;

        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
        {
            Variables.Add((browseName, attributeInfo));
            return new Handle(attributeInfo.FullName);
        }

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

        private sealed class Handle(string fullReference) : IVariableHandle
        {
            public string FullReference { get; } = fullReference;

            public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new Sink();
        }

        private sealed class Sink : IAlarmConditionSink
        {
            public void OnTransition(AlarmEventArgs args) { }
        }
    }
}