using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
///
/// TestKit coverage for 's Primary-only historization gate.
/// The actor caches this node's from the redundancy-state
/// topic and SKIPS the sink enqueue when the local node is Secondary/Detached so a future
/// per-node feeder writes exactly once across the warm-redundant pair. Unknown/null role
/// default-writes (single-node deploys + the boot window must never silently drop historization).
///
public sealed class HistorianAdapterActorTests : RuntimeActorTestBase
{
/// The local node id the gating tests construct the adapter with.
private static readonly NodeId LocalNode = new("node-A");
/// A short window we allow the fire-and-forget enqueue to land within.
private static readonly TimeSpan Settle = TimeSpan.FromMilliseconds(500);
/// Thread-safe fake sink that records every call.
private sealed class RecordingSink : IAlarmHistorianSink
{
private readonly object _lock = new();
private readonly List _events = new();
/// The number of calls observed so far.
public int EnqueueCount { get { lock (_lock) { return _events.Count; } } }
/// A snapshot of every event enqueued so far (in arrival order).
public IReadOnlyList Events
{
get { lock (_lock) { return _events.ToArray(); } }
}
///
public Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
{
lock (_lock)
{
_events.Add(evt);
}
return Task.CompletedTask;
}
///
public HistorianSinkStatus GetStatus() => new(
QueueDepth: 0,
DeadLetterDepth: 0,
LastDrainUtc: null,
LastSuccessUtc: null,
LastError: null,
DrainState: HistorianDrainState.Idle);
}
/// Builds a minimal for the gate tests.
private static AlarmHistorianEvent SampleEvent() => new(
AlarmId: "alm-1",
EquipmentPath: "Area/Line/Equip",
AlarmName: "HiHi",
AlarmTypeName: "LimitAlarm",
Severity: AlarmSeverity.High,
EventKind: "Activated",
Message: "level high",
User: "system",
Comment: null,
TimestampUtc: DateTime.UtcNow);
/// Tell a snapshot marking
/// with so the gate observes the local role.
private static void TellRedundancyRole(IActorRef actor, RedundancyRole role) =>
actor.Tell(new RedundancyStateChanged(
new[]
{
new NodeRedundancyState(
NodeId: LocalNode,
Role: role,
IsClusterLeader: role == RedundancyRole.Primary,
IsRoleLeaderForDriver: role == RedundancyRole.Primary,
AsOfUtc: DateTime.UtcNow),
},
CorrelationId.NewId()));
/// Default-write (T1): before any redundancy snapshot — the boot window and the steady
/// state for single-node deploys — the adapter MUST historize. Constructed WITH a localNode but
/// no snapshot sent, so the cached role is unknown ⇒ default-write.
[Fact]
public void Default_before_redundancy_state_historizes()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
actor.Tell(SampleEvent());
AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), Settle);
}
/// Secondary suppression (T2): when the cached local role is Secondary, the adapter MUST
/// NOT enqueue to the durable sink (the Primary writes the single copy).
[Fact]
public void Secondary_node_does_not_historize()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Secondary);
actor.Tell(SampleEvent());
// Give the (suppressed) fire-and-forget a stable window, then assert nothing landed.
ExpectNoMsg(Settle);
sink.EnqueueCount.ShouldBe(0);
}
/// Detached suppression (T3): a Detached node likewise MUST NOT historize.
[Fact]
public void Detached_node_does_not_historize()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Detached);
actor.Tell(SampleEvent());
ExpectNoMsg(Settle);
sink.EnqueueCount.ShouldBe(0);
}
/// Primary writes (T4): when the cached local role is Primary, the adapter historizes as
/// normal (this is the single copy the durable sink sees).
[Fact]
public void Primary_node_historizes()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Primary);
actor.Tell(SampleEvent());
AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), Settle);
}
/// Absent-node default-historize (T5): a snapshot that mentions only a DIFFERENT node
/// must NOT update the local cached role — the actor's own node is absent, so the role stays
/// null/unknown and the default-historize path must fire. Partial/stale snapshots MUST NOT
/// silently suppress historization for nodes not yet observed.
[Fact]
public void Redundancy_snapshot_without_local_node_leaves_role_unknown_and_historizes()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
// Send a snapshot that only describes a peer node — the local node is absent.
actor.Tell(new RedundancyStateChanged(
new[]
{
new NodeRedundancyState(
NodeId: new NodeId("some-other-node"),
Role: RedundancyRole.Secondary,
IsClusterLeader: false,
IsRoleLeaderForDriver: false,
AsOfUtc: DateTime.UtcNow),
},
CorrelationId.NewId()));
actor.Tell(SampleEvent());
// Local role is still unknown ⇒ default-historize path: sink must record exactly one enqueue.
AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), Settle);
}
/// Builds an (the shape published on the alerts
/// DPS topic) for the translate tests, with overridable severity / type / comment / kind.
/// is bool? so tests can pass null to simulate the
/// rolling-restart / cross-version case (missing field → CLR default null).
private static AlarmTransitionEvent SampleTransition(
int severity = 750,
string alarmTypeName = "LimitAlarm",
string? comment = "note",
string transitionKind = "Activated",
bool? historizeToAveva = true) => new(
AlarmId: "alm-9",
EquipmentPath: "Area/Line/Equip",
AlarmName: "HiHi",
TransitionKind: transitionKind,
Severity: severity,
Message: "level high",
User: "operator1",
TimestampUtc: DateTime.UtcNow,
AlarmTypeName: alarmTypeName,
Comment: comment,
HistorizeToAveva: historizeToAveva);
/// Alerts translate (T6): an off the alerts topic
/// is translated to an and historized by default (unknown role).
/// The translation must carry AlarmId, AlarmTypeName, EventKind (← TransitionKind), Severity bucket,
/// and Comment through faithfully.
[Fact]
public void Alerts_transition_is_historized_by_default()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
actor.Tell(SampleTransition());
AwaitAssert(
() =>
{
sink.EnqueueCount.ShouldBe(1);
var e = sink.Events.ShouldHaveSingleItem();
e.AlarmId.ShouldBe("alm-9");
e.AlarmTypeName.ShouldBe("LimitAlarm");
e.EventKind.ShouldBe("Activated");
e.Severity.ShouldBe(AlarmSeverity.High);
e.Comment.ShouldBe("note");
},
Settle);
}
/// Secondary suppression for alerts (T7): a Secondary node must NOT historize a transition
/// off the alerts topic — the Primary writes the single copy (DistributedPubSub fans the
/// single publish to BOTH nodes' historian adapters, so the gate is what makes it exactly-once).
[Fact]
public void Secondary_node_does_not_historize_alerts_transition()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Secondary);
actor.Tell(SampleTransition());
ExpectNoMsg(Settle);
sink.EnqueueCount.ShouldBe(0);
}
/// Primary writes alerts (T8): a Primary node historizes a transition off the
/// alerts topic (the single copy the durable sink sees).
[Fact]
public void Primary_node_historizes_alerts_transition()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Primary);
actor.Tell(SampleTransition());
AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), Settle);
}
/// Per-alarm opt-out (T8b): a Primary node must NOT historize a transition whose
/// HistorizeToAveva is false — that flag is a per-alarm opt-out of DURABLE
/// historization only. The live alerts fan-out already happened upstream (the publish is NOT
/// gated on this flag), so only the durable sink write is suppressed.
[Fact]
public void Primary_node_does_not_historize_when_opted_out()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Primary);
actor.Tell(SampleTransition(historizeToAveva: false));
ExpectNoMsg(Settle);
sink.EnqueueCount.ShouldBe(0);
}
/// Rolling-restart default-on (T8c): when HistorizeToAveva is null — the shape
/// a cross-version / rolling-restart deserialize produces (old-format message missing the field maps to
/// the CLR default null for bool?) — a Primary node MUST historize. null is the
/// safe default-on posture: no audit row is dropped at a handover, matching the AlarmTypeName
/// null-coalesce precedent in the same HistorianAdapterActor.Translate.
[Fact]
public void Primary_historizes_when_flag_is_null()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink, LocalNode));
TellRedundancyRole(actor, RedundancyRole.Primary);
actor.Tell(SampleTransition(historizeToAveva: null));
AwaitAssert(() => sink.EnqueueCount.ShouldBe(1), Settle);
}
/// Severity buckets (T9): the OPC UA 1–1000 numeric severity on the transition maps onto
/// the coarse at the same ceilings ScriptedAlarmHostActor.SeverityToInt
/// emits (Low≤250, Medium≤500, High≤750, Critical otherwise). Driven end-to-end through the enqueue.
[Theory]
[InlineData(250, AlarmSeverity.Low)]
[InlineData(251, AlarmSeverity.Medium)]
[InlineData(500, AlarmSeverity.Medium)]
[InlineData(750, AlarmSeverity.High)]
[InlineData(751, AlarmSeverity.Critical)]
[InlineData(1000, AlarmSeverity.Critical)]
public void Alerts_transition_severity_buckets(int severity, AlarmSeverity expected)
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
actor.Tell(SampleTransition(severity: severity));
AwaitAssert(
() => sink.Events.ShouldHaveSingleItem().Severity.ShouldBe(expected),
Settle);
}
/// Rolling-restart null default (T10): an old-format transition deserialized by Akka's JSON
/// serializer applies the CLR default (null) to AlarmTypeName rather than the record's
/// "AlarmCondition" call-site default. The translation must null-coalesce that back to
/// "AlarmCondition" so the historian never stores a null alarm type. Forced here by constructing the
/// transition with AlarmTypeName: null! (simulating the post-deserialization shape).
[Fact]
public void Alerts_transition_with_missing_AlarmTypeName_defaults()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(HistorianAdapterActor.Props(sink));
actor.Tell(SampleTransition(alarmTypeName: null!));
AwaitAssert(
() => sink.Events.ShouldHaveSingleItem().AlarmTypeName.ShouldBe("AlarmCondition"),
Settle);
}
}