ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/HighReviewRegressionTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;

public class HighReviewRegressionTests : TestKit
{
    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
    private static IDataConnection Adapter()
    {
        var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
        adapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>()).Returns("handle");
        adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()).Returns(new ReadResult(false, null, null));
        return adapter;
    }
    private IActorRef Actor(IDataConnection adapter) => Sys.ActorOf(Props.Create(() => new DataConnectionActor(
        "conn", adapter, new DataConnectionOptions { ReconnectInterval = TimeSpan.FromMilliseconds(50), TagResolutionRetryInterval = TimeSpan.FromHours(1), SeedReadMaxAttempts = 1 },
        _health, Substitute.For<IDataConnectionFactory>(), "OpcUa")));
    private void Connected() => AwaitCondition(() => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateConnectionHealth" && c.GetArguments()[1] is ConnectionHealth.Connected));
    private static SubscribeTagsRequest Subscribe(string correlation = "s") => new(correlation, "instance", "conn", ["tag"], DateTimeOffset.UtcNow);

    private DataConnectionHealthReport Health(IActorRef actor)
    {
        actor.Tell(new DataConnectionActor.GetHealthReport());
        return ExpectMsg<DataConnectionHealthReport>();
    }

    [Fact]
    public void ResolutionTick_DoesNotDuplicateAnInitialSubscribeAlreadyInFlight()
    {
        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag");
        var actor = Actor(adapter);
        var first = CreateTestProbe();
        actor.Tell(Subscribe(), first.Ref);
        first.ExpectMsg<TagValueUpdate>();
        first.ExpectMsg<SubscribeTagsResponse>();
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.SubscribeGate = gate.Task;
        adapter.FailingTags.Clear();
        var second = CreateTestProbe();
        actor.Tell(Subscribe("second") with { InstanceUniqueName = "second-instance" }, second.Ref);
        AwaitCondition(() => adapter.SubscribeBatches.Count == 2);
        var unexpected = CreateTestProbe();
        adapter.BatchSubscribeThrows = () => { unexpected.Ref.Tell("duplicate subscribe"); return null; };
        actor.Tell(new DataConnectionActor.RetryTagResolution());
        try { unexpected.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); }
        finally { gate.SetResult(); }
    }

    [Fact]
    public void OriginalOwnerDies_InFlightHandleStillServesDifferentInstance()
    {
        var adapter = Adapter();
        var gate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.SubscribeAsync("tag", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
        var actor = Actor(adapter);
        var a = CreateTestProbe(); var b = CreateTestProbe();
        actor.Tell(Subscribe(), a.Ref);
        AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync"));
        actor.Tell(Subscribe("b") with { InstanceUniqueName = "other-instance" }, b.Ref);
        b.ExpectMsg<SubscribeTagsResponse>();
        Health(actor); // mailbox barrier after B's registration
        var beforeDeath = _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagResolution");
        Sys.Stop(a.Ref);
        AwaitCondition(() => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagResolution") > beforeDeath);
        // The remaining instance still needs the tag, even though A started the I/O.
        Assert.Equal(1, Health(actor).TotalSubscribedTags);
        gate.SetResult("shared-handle");
        AwaitAssert(() => Assert.Equal(1, Health(actor).ResolvedTags), TimeSpan.FromSeconds(1));
        Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "UnsubscribeAsync");
    }

    [Fact]
    public void SubscriberTermination_DuringSubscribe_ReleasesLateHandleWithoutResurrection()
    {
        var adapter = Adapter();
        var gate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.SubscribeAsync("tag", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
        var actor = Actor(adapter);
        var subscriber = CreateTestProbe();
        actor.Tell(Subscribe(), subscriber.Ref);
        AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync"));
        Assert.Equal(1, Health(actor).TotalSubscribedTags);
        Sys.Stop(subscriber.Ref);
        AwaitAssert(() => Assert.Equal(0, Health(actor).TotalSubscribedTags));
        gate.SetResult("late-handle");
        AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UnsubscribeAsync" && Equals(c.GetArguments()[0], "late-handle")));
        Assert.Equal(0, Health(actor).ResolvedTags);
    }

    [Fact]
    public void Replacement_DuringSubscribe_AdoptsOnlyStillRequestedTags()
    {
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var adapter = new FakeBatchDataConnection { SubscribeGate = gate.Task };
        var actor = Actor(adapter);
        var old = CreateTestProbe(); var current = CreateTestProbe();
        actor.Tell(new SubscribeTagsRequest("old", "instance", "conn", ["keep", "drop"], DateTimeOffset.UtcNow), old.Ref);
        AwaitCondition(() => adapter.SubscribeBatches.Count == 1);
        actor.Tell(new SubscribeTagsRequest("new", "instance", "conn", ["keep"], DateTimeOffset.UtcNow), current.Ref);
        current.ExpectMsg<SubscribeTagsResponse>();
        Assert.Equal(1, Health(actor).TotalSubscribedTags);
        gate.SetResult();
        AwaitCondition(() => adapter.UnsubscribeBatches.SelectMany(b => b).Contains("sub-2"));
        AwaitAssert(() => Assert.Equal(1, Health(actor).ResolvedTags));
        Assert.DoesNotContain("sub-1", adapter.UnsubscribeBatches.SelectMany(b => b));
        Sys.Stop(old.Ref);
        Assert.Equal(1, Health(actor).TotalSubscribedTags);
    }

    [Fact]
    public void SubscriberTermination_DuringResolution_ReleasesLateHandle()
    {
        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag");
        var options = new DataConnectionOptions { TagResolutionRetryInterval = TimeSpan.FromMilliseconds(500), SeedReadMaxAttempts = 1 };
        var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor("conn", adapter, options, _health, Substitute.For<IDataConnectionFactory>(), "OpcUa")));
        var subscriber = CreateTestProbe();
        actor.Tell(Subscribe(), subscriber.Ref);
        subscriber.ExpectMsg<TagValueUpdate>();
        subscriber.ExpectMsg<SubscribeTagsResponse>();
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.SubscribeGate = gate.Task;
        adapter.FailingTags.Clear();
        AwaitCondition(() => adapter.SubscribeBatches.Count == 2);
        Sys.Stop(subscriber.Ref);
        AwaitAssert(() => Assert.Equal(0, Health(actor).TotalSubscribedTags));
        gate.SetResult();
        AwaitCondition(() => adapter.UnsubscribeBatches.SelectMany(b => b).Any());
        Assert.Equal(0, Health(actor).ResolvedTags);
    }

    [Fact]
    public void DisconnectedSubscribeRetries_KeepOnlyLatestRequest()
    {
        var adapter = Adapter();
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
        var actor = Actor(adapter);
        for (var i = 0; i < 100; i++) actor.Tell(Subscribe(i.ToString()));
        Health(actor); // all retries have reached the disconnected-state queue
        gate.SetResult();
        ExpectMsg<SubscribeTagsResponse>(r => r.CorrelationId == "99");
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        Assert.Single(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "SubscribeAsync");
    }

    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public void DisconnectedWrites_AreRejectedAndNeverReplayed(bool reconnect)
    {
        var adapter = Adapter();
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        if (!reconnect) adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
        var actor = Actor(adapter);
        if (reconnect)
        {
            Connected();
            adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(gate.Task);
            adapter.Disconnected += Raise.Event<Action>();
            AwaitCondition(() => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateConnectionHealth" && c.GetArguments()[1] is ConnectionHealth.Disconnected));
        }
        actor.Tell(new WriteTagRequest("single", "conn", "tag", 42, DateTimeOffset.UtcNow));
        ExpectMsg<WriteTagResponse>(r => !r.Success && r.CorrelationId == "single", TimeSpan.FromSeconds(1));
        actor.Tell(new WriteTagBatchRequest("batch", "conn", new Dictionary<string, object?> { ["tag"] = 42 }, null, null, DateTimeOffset.UtcNow));
        ExpectMsg<WriteTagBatchResponse>(r => !r.Success && r.CorrelationId == "batch", TimeSpan.FromSeconds(1));
        gate.SetResult();
        actor.Tell(Subscribe());
        ExpectMsg<SubscribeTagsResponse>();
        Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name is "WriteAsync" or "WriteBatchAsync");
    }

    [Fact]
    public void SubscriberTermination_ReleasesLastTagHandle()
    {
        var adapter = Adapter();
        var actor = Actor(adapter);
        var subscriber = CreateTestProbe();
        actor.Tell(Subscribe(), subscriber.Ref);
        subscriber.ExpectMsg<SubscribeTagsResponse>();
        Sys.Stop(subscriber.Ref);
        AwaitCondition(() => adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UnsubscribeAsync" && Equals(c.GetArguments()[0], "handle")), TimeSpan.FromSeconds(1));
    }

    [Fact]
    public void OldOwnerUnsubscribe_DoesNotRemoveReplacementTags()
    {
        var adapter = Adapter();
        SubscriptionCallback? callback = null;
        adapter.SubscribeAsync("tag", Arg.Do<SubscriptionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("handle");
        var actor = Actor(adapter);
        var old = CreateTestProbe();
        var current = CreateTestProbe();
        actor.Tell(Subscribe("old"), old.Ref);
        old.ExpectMsg<SubscribeTagsResponse>();
        actor.Tell(Subscribe("new"), current.Ref);
        current.ExpectMsg<SubscribeTagsResponse>();
        actor.Tell(new UnsubscribeTagsRequest("u", "instance", "conn", DateTimeOffset.UtcNow), old.Ref);
        actor.Tell(Subscribe("barrier"), current.Ref);
        current.ExpectMsg<SubscribeTagsResponse>();
        callback!("tag", new TagValue(7, QualityCode.Good, DateTimeOffset.UtcNow));
        current.ExpectMsg<TagValueUpdate>(u => Equals(u.Value, 7));
        Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "UnsubscribeAsync");
    }

    [Fact]
    public void DuplicateCompletion_WithLiveHandle_DoesNotUnsubscribeIt()
    {
        var adapter = Adapter();
        var actor = Actor(adapter);
        actor.Tell(Subscribe());
        ExpectMsg<SubscribeTagsResponse>();
        actor.Tell(new DataConnectionActor.SubscribeCompleted(Subscribe("duplicate"), TestActor,
            [new DataConnectionActor.SubscribeTagResult("tag", false, true, "handle", null)], [], 0, adapter));
        ExpectMsg<SubscribeTagsResponse>();
        Assert.DoesNotContain(adapter.ReceivedCalls(), c => c.GetMethodInfo().Name == "UnsubscribeAsync");
    }

    [Fact]
    public void SourceScopedSnapshotEnd_IsNotBroadcastToOtherSources()
    {
        var adapter = Adapter();
        var alarm = (IAlarmSubscribableConnection)adapter;
        AlarmTransitionCallback? callback = null;
        alarm.SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(), Arg.Do<AlarmTransitionCallback>(c => callback = c), Arg.Any<CancellationToken>()).Returns("alarm-handle");
        var actor = Actor(adapter);
        var a = CreateTestProbe(); var b = CreateTestProbe();
        actor.Tell(new SubscribeAlarmsRequest("a", "a", "conn", "A", null, DateTimeOffset.UtcNow), a.Ref);
        a.ExpectMsg<SubscribeAlarmsResponse>();
        actor.Tell(new SubscribeAlarmsRequest("b", "b", "conn", "B", null, DateTimeOffset.UtcNow), b.Ref);
        b.ExpectMsg<SubscribeAlarmsResponse>();
        callback!(new NativeAlarmTransition("", "B", "", AlarmTransitionKind.SnapshotComplete,
            new AlarmConditionState(false, false, null, AlarmShelveState.Unshelved, false, 0), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", ""));
        b.ExpectMsg<NativeAlarmTransitionUpdate>();
        a.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }
}