using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
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;
///
/// ScadaBridge#138 / OtOpcUa Warm design §5 "Alarms are compatible", at the actor: the promotion
/// refresh OtOpcUa sends on its own (an unsolicited RefreshStart / retained conditions / RefreshEnd
/// on an EXISTING alarm feed, already framed by RealOpcUaClient as SnapshotStart / Snapshot /
/// SnapshotComplete under a new epoch — see RealOpcUaPromotionRefreshTests) goes through the
/// existing snapshot handling: it is forwarded in order to the subscriber, and it RECONCILES the
/// connection's alarm cache — a condition the promoted node no longer retains is gone from what a
/// later subscriber is replayed.
///
public class DataConnectionPromotionRefreshTests : TestKit
{
private const string Source = "Tank01";
private readonly ISiteHealthCollector _health = Substitute.For();
private readonly IDataConnectionFactory _factory = Substitute.For();
private readonly DataConnectionOptions _options = new()
{
ReconnectInterval = TimeSpan.FromMilliseconds(100),
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
WriteTimeout = TimeSpan.FromSeconds(5),
};
private static NativeAlarmTransition Boundary(AlarmTransitionKind kind, Guid epoch) => new(
"", Source, "", kind,
new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
"", "", "", "", "", null, DateTimeOffset.UtcNow, "", "",
SnapshotSourceReference: Source, SnapshotId: epoch);
private static NativeAlarmTransition Condition(
string condition, AlarmTransitionKind kind, Guid? epoch, bool active = true, bool acked = false) => new(
$"{Source}.{condition}", Source, "AlarmConditionType", kind,
new AlarmConditionState(active, acked, null, AlarmShelveState.Unshelved, false, 500),
"", "", "msg", "", "", null, DateTimeOffset.UtcNow, "", "",
SnapshotSourceReference: epoch is null ? null : Source, SnapshotId: epoch);
///
/// Subscribe-time snapshot {Hi, Lo}; then the unsolicited promotion snapshot {Hi} (Lo cleared
/// while the node was Standby). The subscriber sees the promotion frame in order, and a second
/// subscriber that joins afterwards is replayed Hi only.
///
[Fact]
public void UnsolicitedPromotionSnapshot_IsForwardedInOrder_AndReconcilesTheCache()
{
AlarmTransitionCallback? cb = null;
var adapter = Substitute.For();
adapter.ConnectAsync(Arg.Any>(), Arg.Any())
.Returns(Task.CompletedTask);
((IAlarmSubscribableConnection)adapter)
.SubscribeAlarmsAsync(Arg.Any(), Arg.Any(),
Arg.Do(c => cb = c), Arg.Any())
.Returns(Task.FromResult("alarm-sub-1"));
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
"conn", adapter, _options, _health, _factory, "OpcUa")));
var first = CreateTestProbe();
actor.Tell(new SubscribeAlarmsRequest("c1", "inst1", "conn", Source, null, DateTimeOffset.UtcNow), first);
first.ExpectMsg(m => m.Success);
Assert.NotNull(cb);
// Subscribe-time refresh on the then-Active node.
var subscribeEpoch = Guid.NewGuid();
cb!(Boundary(AlarmTransitionKind.SnapshotStart, subscribeEpoch));
cb(Condition("Hi", AlarmTransitionKind.Snapshot, subscribeEpoch));
cb(Condition("Lo", AlarmTransitionKind.Snapshot, subscribeEpoch));
cb(Boundary(AlarmTransitionKind.SnapshotComplete, subscribeEpoch));
first.ReceiveN(4, TimeSpan.FromSeconds(3));
// Promotion on the other node: an unsolicited refresh on this existing feed.
var promotionEpoch = Guid.NewGuid();
cb(Boundary(AlarmTransitionKind.SnapshotStart, promotionEpoch));
cb(Condition("Hi", AlarmTransitionKind.Snapshot, promotionEpoch));
cb(Boundary(AlarmTransitionKind.SnapshotComplete, promotionEpoch));
var frame = first.ReceiveN(3, TimeSpan.FromSeconds(3)).Cast().ToList();
Assert.Equal(
[AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete],
frame.Select(u => u.Transition.Kind));
Assert.All(frame, u => Assert.Equal(promotionEpoch, u.Transition.SnapshotId));
Assert.Equal($"{Source}.Hi", frame[1].Transition.SourceReference);
// A live event after the promotion is still routed live.
cb(Condition("Hi", AlarmTransitionKind.Acknowledge, epoch: null, acked: true));
first.ExpectMsg(u => u.Transition.Kind == AlarmTransitionKind.Acknowledge);
// A subscriber joining now is replayed the reconciled state: Hi only, Lo is gone.
var second = CreateTestProbe();
actor.Tell(new SubscribeAlarmsRequest("c2", "inst2", "conn", Source, null, DateTimeOffset.UtcNow), second);
var replay = second.ReceiveN(4, TimeSpan.FromSeconds(3));
var transitions = replay.OfType().Select(u => u.Transition).ToList();
Assert.Single(replay.OfType(), r => r.Success);
Assert.Equal(
[AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete],
transitions.Select(t => t.Kind));
Assert.Equal([$"{Source}.Hi"], transitions.Where(t => t.Kind == AlarmTransitionKind.Snapshot).Select(t => t.SourceReference));
second.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
}