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.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
public sealed class OpcUaPublishActorTests : RuntimeActorTestBase
{
/// Verifies that message contracts are accepted without a pinned dispatcher in tests.
[Fact]
public void Accepts_message_contracts_without_pinned_dispatcher_in_tests()
{
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests());
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate("ns=2;s=Tag1", 42.0, OpcUaQuality.Good, DateTime.UtcNow));
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("ns=2;s=Alarm1", Snapshot(active: true), DateTime.UtcNow));
actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(240));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
/// Verifies that production props target the OPC UA synchronized dispatcher.
[Fact]
public void Production_Props_targets_opcua_synchronized_dispatcher()
{
var props = OpcUaPublishActor.Props();
props.Dispatcher.ShouldBe(OpcUaPublishActor.DispatcherId);
}
/// Verifies that AttributeValueUpdate routes to sink WriteValue.
[Fact]
public void AttributeValueUpdate_routes_to_sink_WriteValue()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate("ns=2;s=T1", 3.14, OpcUaQuality.Good, DateTime.UtcNow));
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate("ns=2;s=T2", "abc", OpcUaQuality.Uncertain, DateTime.UtcNow));
AwaitAssert(() =>
{
sink.Values.Count.ShouldBe(2);
sink.Values[0].NodeId.ShouldBe("ns=2;s=T1");
sink.Values[0].Value.ShouldBe(3.14);
sink.Values[0].Quality.ShouldBe(OpcUaQuality.Good);
sink.Values[1].Quality.ShouldBe(OpcUaQuality.Uncertain);
}, duration: TimeSpan.FromMilliseconds(500));
}
/// Verifies that AlarmStateUpdate routes to sink WriteAlarmCondition with the full snapshot.
[Fact]
public void AlarmStateUpdate_routes_to_sink_WriteAlarmCondition()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate(
"ns=2;s=A1", Snapshot(active: true, acknowledged: false, severity: 700), DateTime.UtcNow));
AwaitAssert(() =>
{
sink.Alarms.Count.ShouldBe(1);
sink.Alarms[0].AlarmNodeId.ShouldBe("ns=2;s=A1");
sink.Alarms[0].State.Active.ShouldBeTrue();
sink.Alarms[0].State.Acknowledged.ShouldBeFalse();
sink.Alarms[0].State.Severity.ShouldBe((ushort)700);
}, duration: TimeSpan.FromMilliseconds(500));
}
/// Builds a test with sensible defaults so each test
/// only specifies the fields it cares about.
private static AlarmConditionSnapshot Snapshot(
bool active = false,
bool acknowledged = true,
bool confirmed = true,
bool enabled = true,
AlarmShelvingKind shelving = AlarmShelvingKind.Unshelved,
ushort severity = 500,
string message = "test") =>
new(active, acknowledged, confirmed, enabled, shelving, severity, message);
/// Verifies that RebuildAddressSpace calls sink Rebuild.
[Fact]
public void RebuildAddressSpace_calls_sink_Rebuild()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
AwaitAssert(() => sink.RebuildCalls.ShouldBe(1), duration: TimeSpan.FromMilliseconds(500));
}
/// Verifies that ServiceLevelChanged publishes to IServiceLevelPublisher once per unique level.
[Fact]
public void ServiceLevelChanged_publishes_to_IServiceLevelPublisher_once_per_unique_level()
{
var publisher = new RecordingPublisher();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(serviceLevel: publisher));
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(240));
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(240)); // dedup
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 100 }),
duration: TimeSpan.FromMilliseconds(500));
}
/// Verifies that RedundancyStateChanged drives local ServiceLevel publish for primary leader.
[Fact]
public void RedundancyStateChanged_drives_local_ServiceLevel_publish_for_primary_leader()
{
var publisher = new RecordingPublisher();
var local = NodeId.Parse("primary-node");
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
serviceLevel: publisher, localNode: local));
var snapshot = new RedundancyStateChanged(
Nodes: new[]
{
new NodeRedundancyState(local, RedundancyRole.Primary,
IsClusterLeader: true, IsRoleLeaderForDriver: true, DateTime.UtcNow),
new NodeRedundancyState(NodeId.Parse("other-node"), RedundancyRole.Secondary,
IsClusterLeader: false, IsRoleLeaderForDriver: false, DateTime.UtcNow),
},
CorrelationId.NewId());
actor.Tell(snapshot);
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }),
duration: TimeSpan.FromMilliseconds(500));
}
/// Verifies that RedundancyStateChanged for secondary publishes 100.
[Fact]
public void RedundancyStateChanged_for_secondary_publishes_100()
{
var publisher = new RecordingPublisher();
var local = NodeId.Parse("secondary-node");
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(serviceLevel: publisher, localNode: local));
var snapshot = new RedundancyStateChanged(
Nodes: new[]
{
new NodeRedundancyState(local, RedundancyRole.Secondary,
IsClusterLeader: false, IsRoleLeaderForDriver: false, DateTime.UtcNow),
},
CorrelationId.NewId());
actor.Tell(snapshot);
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 100 }),
duration: TimeSpan.FromMilliseconds(500));
}
/// Test implementation of IOpcUaAddressSpaceSink that records calls.
private sealed class RecordingSink : IOpcUaAddressSpaceSink
{
/// Gets the queue of recorded value updates.
public ConcurrentQueue<(string NodeId, object? Value, OpcUaQuality Quality, DateTime Ts)> ValueQueue { get; } = new();
/// Gets the queue of recorded alarm condition updates.
public ConcurrentQueue<(string AlarmNodeId, AlarmConditionSnapshot State, DateTime Ts)> AlarmQueue { get; } = new();
/// Count of rebuild calls.
public int RebuildCalls;
/// Gets the list of recorded value updates.
public List<(string NodeId, object? Value, OpcUaQuality Quality, DateTime Ts)> Values =>
ValueQueue.ToList();
/// Gets the list of recorded alarm condition updates.
public List<(string AlarmNodeId, AlarmConditionSnapshot State, DateTime Ts)> Alarms =>
AlarmQueue.ToList();
/// Records a value update.
/// The OPC UA node identifier.
/// The attribute value.
/// The OPC UA quality code.
/// The timestamp of the update.
public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime ts) =>
ValueQueue.Enqueue((nodeId, value, quality, ts));
/// Records an alarm condition update.
/// The OPC UA alarm node identifier.
/// The full condition state snapshot.
/// The timestamp of the update.
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts) =>
AlarmQueue.Enqueue((alarmNodeId, state, ts));
/// Materialises an alarm condition (no-op in test).
/// The alarm node ID.
/// The equipment folder node ID.
/// The condition display name.
/// The domain alarm type.
/// The domain severity.
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity) { }
/// Ensures a folder exists (no-op in test).
/// The OPC UA folder node identifier.
/// The parent folder node identifier, or null for root.
/// The display name of the folder.
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { }
/// Ensures a variable exists (no-op in test).
/// The OPC UA variable node identifier.
/// The parent folder node identifier, or null for root.
/// The display name of the variable.
/// The OPC UA built-in type name.
/// Whether the node is created read/write.
public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable) { }
/// Records a rebuild call.
public void RebuildAddressSpace() => Interlocked.Increment(ref RebuildCalls);
}
/// Test implementation of IServiceLevelPublisher that records publishes.
private sealed class RecordingPublisher : IServiceLevelPublisher
{
private readonly ConcurrentQueue _q = new();
/// Gets the recorded service levels.
public byte[] Levels => _q.ToArray();
/// Records a service level publish.
/// The service level value to publish.
public void Publish(byte serviceLevel) => _q.Enqueue(serviceLevel);
}
}