lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorEndpointStateTests.cs

using System.Collections.Concurrent;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;

/// <summary>
///     lmxopcua#762: a node whose own OPC UA SDK failed to start must not advertise 250/240. Each test pairs the
///     failed endpoint with a positive control — the SAME actor wiring and snapshot with a started endpoint — so a
///     0 cannot come from some other demotion (Detached, a missing entry, a peer probe).
/// </summary>
public sealed class OpcUaPublishActorEndpointStateTests : RuntimeActorTestBase
{
    /// <summary>DB-less driver node (the survive-alone posture): Primary would be 250, a failed endpoint is 0.</summary>
    [Theory]
    [InlineData(true, 250)]
    [InlineData(false, 240)]
    public void DbLess_failed_endpoint_publishes_zero_instead_of_healthy_level(bool primary, byte healthy)
    {
        PublishedLevelFor(OpcUaEndpointStatus.Started, primary, dbLess: true).ShouldBe(healthy, "positive control");
        PublishedLevelFor(OpcUaEndpointStatus.Failed, primary, dbLess: true).ShouldBe((byte)0);
    }

    /// <summary>DB-backed node on the full calculator path (DB reachable, fresh): same rule.</summary>
    [Fact]
    public void DbBacked_failed_endpoint_publishes_zero_instead_of_250()
    {
        PublishedLevelFor(OpcUaEndpointStatus.Started, primary: true, dbLess: false, dbBacked: true).ShouldBe((byte)250, "positive control");
        PublishedLevelFor(OpcUaEndpointStatus.Failed, primary: true, dbLess: false, dbBacked: true).ShouldBe((byte)0);
    }

    /// <summary>The legacy role-only seam (no DB probe wired yet) must not hand a failed endpoint 240 either.</summary>
    [Fact]
    public void Legacy_role_only_seam_failed_endpoint_publishes_zero_instead_of_240()
    {
        PublishedLevelFor(OpcUaEndpointStatus.Started, primary: true, dbLess: false).ShouldBe((byte)240, "positive control");
        PublishedLevelFor(OpcUaEndpointStatus.Failed, primary: true, dbLess: false).ShouldBe((byte)0);
    }

    /// <summary>
    ///     Boot (<see cref="OpcUaEndpointStatus.NotStarted"/>) and shutdown (<see cref="OpcUaEndpointStatus.Stopped"/>)
    ///     are not faults: only a recorded start FAILURE demotes.
    /// </summary>
    [Theory]
    [InlineData(OpcUaEndpointStatus.NotStarted)]
    [InlineData(OpcUaEndpointStatus.Stopped)]
    public void Only_a_failed_start_demotes(OpcUaEndpointStatus status)
    {
        PublishedLevelFor(status, primary: true, dbLess: true).ShouldBe((byte)250);
    }

    /// <summary>
    ///     The fact is read on every recompute, so an endpoint that fails after the first healthy publish demotes on
    ///     the next health tick without any message being sent to the actor.
    /// </summary>
    [Fact]
    public void Endpoint_failing_later_demotes_on_the_next_tick()
    {
        var publisher = new RecordingPublisher();
        var state = new OpcUaEndpointState();
        state.MarkStarted();
        var local = NodeId.Parse("endpoint-later-node");
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
            serviceLevel: publisher, localNode: local, dbLess: true,
            healthTickInterval: TimeSpan.FromMilliseconds(25), endpointState: state));
        actor.Tell(Snapshot(local, primary: true));
        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 250 }), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);

        state.MarkFailed("port 4840 in use");
        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 250, 0 }), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
    }

    private byte PublishedLevelFor(OpcUaEndpointStatus status, bool primary, bool dbLess, bool dbBacked = false)
    {
        var publisher = new RecordingPublisher();
        var state = new OpcUaEndpointState();
        switch (status)
        {
            case OpcUaEndpointStatus.Started: state.MarkStarted(); break;
            case OpcUaEndpointStatus.Failed: state.MarkFailed("port 4840 in use"); break;
            case OpcUaEndpointStatus.Stopped: state.MarkStopped(); break;
        }

        var local = NodeId.Parse($"endpoint-{Guid.NewGuid():N}");
        // A DB-backed actor needs a probe ref to leave the legacy seam; its Ask is never answered here (the 5-min
        // tick keeps it to the one start-up Ask), and a healthy sample is Told directly BEFORE the snapshot so the
        // snapshot's recompute — the first publish — runs the full calculator path.
        var db = dbBacked ? CreateTestProbe() : null;
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
            serviceLevel: publisher, localNode: local, dbLess: dbLess, dbHealthProbe: db?.Ref,
            healthTickInterval: TimeSpan.FromMinutes(5), endpointState: state));
        if (db is not null)
            actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
        actor.Tell(Snapshot(local, primary));
        AwaitAssert(() => publisher.Levels.Length.ShouldBeGreaterThanOrEqualTo(1), PresenceBudget);

        Sys.Stop(actor);
        return publisher.Levels[0];
    }

    private static RedundancyStateChanged Snapshot(NodeId local, bool primary) =>
        new(new[] { new NodeRedundancyState(local,
            primary ? RedundancyRole.Primary : RedundancyRole.Secondary,
            IsClusterLeader: primary, IsDriverPrimary: primary, DateTime.UtcNow) }, CorrelationId.NewId());

    private sealed class RecordingPublisher : IServiceLevelPublisher
    {
        private readonly ConcurrentQueue<byte> _levels = new();
        public byte[] Levels => _levels.ToArray();
        public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel);
    }
}