using Akka.Actor;
using Akka.TestKit.Xunit2;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Communication;
///
/// Proves the Phase 2 node-side leg joins up: a command handed to
/// reaches the real ,
/// not just a test probe.
///
///
///
/// The unit tests either side of this seam both pass with the wiring broken.
/// NodeCommunicationActorTests asserts a probe subscribed to the EventStream receives
/// the message; the driver-host tests drive the actor by direct Tell. Neither notices
/// if DriverHostActor.PreStart never subscribes. That gap is why this file exists.
///
///
/// Ordering matters, and it bit this test first time out. ActorOf returns
/// before PreStart has run, and an EventStream.Publish with no subscriber is
/// dropped silently — there is no buffering and no dead-letter. Relaying immediately after
/// spawning therefore fails intermittently. The warm-up below is not ceremony: it is the
/// only available proof that PreStart has completed. (In production the driver host
/// is spawned at startup, long before any deployment, so the same race does not arise.)
///
///
public class EventStreamCommandDeliveryTests : RuntimeActorTestBase
{
private static DispatchDeployment NewDispatch() => new(
new DeploymentId(Guid.NewGuid()),
new RevisionHash("rev-does-not-exist"),
new CorrelationId(Guid.NewGuid()));
[Fact]
public void A_dispatch_relayed_by_the_comm_actor_reaches_the_real_driver_host()
{
var ackRouter = CreateTestProbe();
var driverHost = Sys.ActorOf(
DriverHostActor.Props(NewInMemoryDbFactory(), new NodeId("127.0.0.1:4053"), ackRouter: ackRouter.Ref),
"driver-host");
var comm = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null), "node-communication");
// Warm-up: a direct Tell whose ack proves PreStart has completed, and with it the
// EventStream subscription. Without this the relay below races the subscription.
var warmup = NewDispatch();
driverHost.Tell(warmup);
ackRouter.ExpectMsg(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(warmup.DeploymentId);
// Now the real assertion — relayed exactly as central would send it. The comm actor only
// publishes node-locally, so the host must be subscribed for this to arrive at all.
var relayed = NewDispatch();
comm.Tell(relayed);
// The deployment id does not exist, so the host answers Failed. That is the stronger
// assertion, not a weaker one: an ack at all proves the command was received and handled,
// and only PreStart's EventStream subscription can have delivered it.
ackRouter.ExpectMsg(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(relayed.DeploymentId);
}
[Fact]
public void CONTROL_the_driver_host_acks_a_direct_tell()
{
// Falsifiability control. If the test above fails, this one says which half broke: green
// here means the host is healthy and the EventStream relay is at fault; red here means the
// host would not have acked either way and the relay is not implicated.
var ackRouter = CreateTestProbe();
var driverHost = Sys.ActorOf(
DriverHostActor.Props(NewInMemoryDbFactory(), new NodeId("127.0.0.1:4053"), ackRouter: ackRouter.Ref),
"driver-host-control");
var dispatch = NewDispatch();
driverHost.Tell(dispatch);
ackRouter.ExpectMsg(TimeSpan.FromSeconds(15)).DeploymentId.ShouldBe(dispatch.DeploymentId);
}
}
///
/// The alarm leg of the same wiring proof, in its own ActorSystem because the shared
/// harness pins loglevel = WARNING and the only
/// observable signal ScriptedAlarmHostActor gives for an unowned alarm is a Debug line.
///
public class AlarmCommandEventStreamDeliveryTests : TestKit
{
private static string DebugLevelHocon => @"
akka {
loglevel = ""DEBUG""
extensions = [
""Akka.Cluster.Tools.PublishSubscribe.DistributedPubSubExtensionProvider, Akka.Cluster.Tools""
]
actor {
provider = ""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster""
}
remote.dot-netty.tcp {
hostname = ""127.0.0.1""
port = 0
}
cluster {
seed-nodes = []
roles = [""driver""]
min-nr-of-members = 1
run-coordinated-shutdown-when-down = off
}
}";
public AlarmCommandEventStreamDeliveryTests()
: base(DebugLevelHocon)
{
var cluster = Akka.Cluster.Cluster.Get(Sys);
cluster.Join(cluster.SelfAddress);
AwaitCondition(
() => cluster.State.Members.Any(m => m.Status == Akka.Cluster.MemberStatus.Up),
TimeSpan.FromSeconds(5));
}
[Fact]
public void An_alarm_command_relayed_by_the_comm_actor_reaches_the_real_scripted_alarm_host()
{
// Needed for the same reason as the deploy leg: the comm actor's unit test only proves a
// probe receives it, and the alarm-host tests drive the actor by direct Tell. Neither
// notices a missing PreStart subscription.
var upstream = new DependencyMuxTagUpstreamSource();
var logger = new LoggerConfiguration().CreateLogger();
var engine = new ScriptedAlarmEngine(
upstream, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger);
var host = Sys.ActorOf(
ScriptedAlarmHostActor.Props(
CreateTestProbe().Ref, CreateTestProbe().Ref, upstream, engine, new NodeId("node-A")),
"scripted-alarm-host");
var comm = Sys.ActorOf(NodeCommunicationActor.Props(centralClient: null), "node-communication");
var cmd = new AlarmCommand("alarm-unowned", "Acknowledge", "alice", "c", null);
// Warm-up, OUTSIDE the real assertion: a direct Tell whose debug line proves PreStart has
// completed and the EventStream subscription is registered. An earlier draft did the direct
// Tell INSIDE the assertion window alongside the relay, which made the test vacuous — the
// direct Tell alone produced the log the assertion was waiting for.
EventFilter.Debug(contains: "ignoring AlarmCommand")
.ExpectOne(TimeSpan.FromSeconds(15), () => host.Tell(cmd));
// The real assertion: relayed ONLY through the comm actor, which publishes node-locally.
// The host owns no alarms, so it logs and stops — still a positive delivery signal, because
// that line is only reached if the message arrived.
EventFilter.Debug(contains: "ignoring AlarmCommand")
.ExpectOne(TimeSpan.FromSeconds(15), () => comm.Tell(cmd));
}
}