Joseph Doherty
361e7f41ba
The MxGateway alarm mapper emits the SnapshotComplete framing sentinel with
empty SourceReference/SourceObjectReference. HandleAlarmTransitionReceived
routed every transition by prefix match against the subscriber's source, so
the empty-ref sentinel ('' .StartsWith("<src>.") == false) was dropped for
any specific source. The NativeAlarmActor buffers snapshot conditions and only
flushes them on SnapshotComplete, so statically-active native alarms delivered
only in the initial snapshot (no later live transition) never surfaced.
Broadcast the SnapshotComplete sentinel to all alarm subscribers (bypassing the
source match + type filter) so each NativeAlarmActor's snapshot swap completes.
Adds a regression test using the real empty-ref sentinel against a specific
(prefix) source.
235 lines
12 KiB
C#
235 lines
12 KiB
C#
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;
|
|
|
|
/// <summary>Task-10: native alarm subscribe + source-ref routing + unavailable signal.</summary>
|
|
public class DataConnectionActorAlarmTests : TestKit
|
|
{
|
|
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
|
|
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
|
|
private readonly DataConnectionOptions _options = new()
|
|
{
|
|
ReconnectInterval = TimeSpan.FromMilliseconds(100),
|
|
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
|
|
WriteTimeout = TimeSpan.FromSeconds(5)
|
|
};
|
|
|
|
private static NativeAlarmTransition Raise(string sourceRef, string sourceObj) =>
|
|
Raise(sourceRef, sourceObj, "AnalogLimit.Hi");
|
|
|
|
private static NativeAlarmTransition Raise(string sourceRef, string sourceObj, string typeName,
|
|
AlarmTransitionKind kind = AlarmTransitionKind.Raise) =>
|
|
new(sourceRef, sourceObj, typeName, kind,
|
|
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 500),
|
|
"Process", "hi", "hi", "", "", null, DateTimeOffset.UtcNow, "92", "90");
|
|
|
|
private static (IDataConnection Adapter, Func<AlarmTransitionCallback?> Cb) BuildAlarmAdapter()
|
|
{
|
|
AlarmTransitionCallback? cb = null;
|
|
var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
|
|
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
|
.Returns(Task.CompletedTask);
|
|
((IAlarmSubscribableConnection)adapter)
|
|
.SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
|
Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
|
|
.Returns(Task.FromResult("alarm-sub-1"));
|
|
return (adapter, () => cb);
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_RoutesTransitionToInstanceSubscriber()
|
|
{
|
|
AlarmTransitionCallback? cb = null;
|
|
var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
|
|
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
|
.Returns(Task.CompletedTask);
|
|
((IAlarmSubscribableConnection)adapter)
|
|
.SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
|
Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
|
|
.Returns(Task.FromResult("alarm-sub-1"));
|
|
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Tank01", null, DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
|
|
Assert.NotNull(cb);
|
|
cb!(Raise("Tank01.Hi", "Tank01"));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.SourceObjectReference == "Tank01");
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_OnNonAlarmCapableAdapter_RepliesFailure()
|
|
{
|
|
var adapter = Substitute.For<IDataConnection>(); // not IAlarmSubscribableConnection
|
|
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
|
.Returns(Task.CompletedTask);
|
|
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Tank01", null, DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => !m.Success && m.ErrorMessage != null);
|
|
}
|
|
|
|
// ── M2.4 (#8): conditionFilter is now applied client-side in the actor ──
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_WithTypeFilter_DeliversOnlyMatchingTypes()
|
|
{
|
|
var (adapter, getCb) = BuildAlarmAdapter();
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Tank01",
|
|
"AnalogLimit.Hi,AnalogLimit.Lo", DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
var cb = getCb();
|
|
Assert.NotNull(cb);
|
|
|
|
// Non-matching type is dropped (no message delivered).
|
|
cb!(Raise("Tank01.HiHi", "Tank01", "AnalogLimit.HiHi"));
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(250));
|
|
|
|
// Matching type is delivered.
|
|
cb!(Raise("Tank01.Hi", "Tank01", "AnalogLimit.Hi"));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.AlarmTypeName == "AnalogLimit.Hi");
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_WithNullFilter_DeliversAllTypes()
|
|
{
|
|
var (adapter, getCb) = BuildAlarmAdapter();
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Tank01", null, DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
var cb = getCb();
|
|
Assert.NotNull(cb);
|
|
|
|
cb!(Raise("Tank01.HiHi", "Tank01", "AnalogLimit.HiHi"));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.AlarmTypeName == "AnalogLimit.HiHi");
|
|
cb!(Raise("Tank01.Lo", "Tank01", "DiscreteAlarm"));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.AlarmTypeName == "DiscreteAlarm");
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_FilterMatch_IgnoresCaseAndWhitespace()
|
|
{
|
|
var (adapter, getCb) = BuildAlarmAdapter();
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Tank01",
|
|
" analoglimit.hi ,\tDISCRETEALARM ", DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
var cb = getCb();
|
|
Assert.NotNull(cb);
|
|
|
|
cb!(Raise("Tank01.Hi", "Tank01", "AnalogLimit.Hi")); // case differs from filter
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.AlarmTypeName == "AnalogLimit.Hi");
|
|
cb!(Raise("Tank01.Disc", "Tank01", "DiscreteAlarm"));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.AlarmTypeName == "DiscreteAlarm");
|
|
cb!(Raise("Tank01.HiHi", "Tank01", "AnalogLimit.HiHi")); // not listed
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(250));
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_GatewayWideFeed_IsFilteredClientSide()
|
|
{
|
|
// MxGateway has no server-side filter: its adapter opens ONE gateway-wide
|
|
// feed and the actor is the authoritative gate. A filtered source must
|
|
// only see its own matching types even though the feed carries everything.
|
|
var (adapter, getCb) = BuildAlarmAdapter();
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "MxGateway")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Reactor",
|
|
"HighTemp", DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
var cb = getCb();
|
|
Assert.NotNull(cb);
|
|
|
|
// Gateway-wide feed delivers a transition for a different source object —
|
|
// dropped by source routing.
|
|
cb!(Raise("Pump.Fault", "Pump", "HighTemp"));
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
// Right source, wrong type — dropped by the client-side type gate.
|
|
cb!(Raise("Reactor.LowTemp", "Reactor", "LowTemp"));
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
// Right source, right type — delivered.
|
|
cb!(Raise("Reactor.HighTemp", "Reactor", "HighTemp"));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u =>
|
|
u.Transition.SourceObjectReference == "Reactor" && u.Transition.AlarmTypeName == "HighTemp");
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_WithFilter_StillForwardsSnapshotCompleteSentinel()
|
|
{
|
|
// The SnapshotComplete framing sentinel (empty AlarmTypeName) must survive
|
|
// the type gate so the NativeAlarmActor's snapshot swap can complete.
|
|
var (adapter, getCb) = BuildAlarmAdapter();
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Tank01",
|
|
"AnalogLimit.Hi", DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
var cb = getCb();
|
|
Assert.NotNull(cb);
|
|
|
|
// Snapshot-complete sentinel: empty source refs (the framing marker) but
|
|
// routed because every subscriber receives it; never type-filtered.
|
|
cb!(new NativeAlarmTransition("Tank01", "Tank01", "", AlarmTransitionKind.SnapshotComplete,
|
|
new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
|
|
"", "", "", "", "", null, DateTimeOffset.UtcNow, "", ""));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.Kind == AlarmTransitionKind.SnapshotComplete);
|
|
}
|
|
|
|
[Fact]
|
|
public void SubscribeAlarms_RealEmptyRefSnapshotComplete_IsBroadcastToSpecificSource()
|
|
{
|
|
// Regression: the real MxGatewayAlarmMapper.SnapshotComplete() emits the
|
|
// sentinel with EMPTY SourceReference / SourceObjectReference. With a
|
|
// specific (prefix) source like "Reactor." the per-source match
|
|
// ("".StartsWith("Reactor.") == false) used to drop it, stranding the
|
|
// buffered snapshot in the NativeAlarmActor forever — statically-active
|
|
// conditions (delivered only in the snapshot) never surfaced. The sentinel
|
|
// must be broadcast to every subscriber so the snapshot swap completes.
|
|
var (adapter, getCb) = BuildAlarmAdapter();
|
|
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
|
"conn", adapter, _options, _health, _factory, "MxGateway")));
|
|
|
|
actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Reactor.", null, DateTimeOffset.UtcNow));
|
|
ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
|
var cb = getCb();
|
|
Assert.NotNull(cb);
|
|
|
|
// Active alarm under the source arrives in the snapshot and routes by prefix.
|
|
cb!(new NativeAlarmTransition(
|
|
"Galaxy!Area.Reactor.HeartbeatTimeoutAlarm", "Reactor.HeartbeatTimeoutAlarm", "Syst",
|
|
AlarmTransitionKind.Snapshot,
|
|
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 400),
|
|
"Area", "", "", "", "", null, DateTimeOffset.UtcNow, "", ""));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u =>
|
|
u.Transition.Kind == AlarmTransitionKind.Snapshot &&
|
|
u.Transition.SourceObjectReference == "Reactor.HeartbeatTimeoutAlarm");
|
|
|
|
// Real sentinel with EMPTY refs — must still reach the "Reactor." subscriber.
|
|
cb!(new NativeAlarmTransition("", "", "", AlarmTransitionKind.SnapshotComplete,
|
|
new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
|
|
"", "", "", "", "", null, DateTimeOffset.UtcNow, "", ""));
|
|
ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.Kind == AlarmTransitionKind.SnapshotComplete);
|
|
}
|
|
}
|