using Akka.Actor; using Akka.TestKit.Xunit2; using NSubstitute; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery; using ZB.MOM.WW.ScadaBridge.Communication.Actors; namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; /// /// WP-4: Tests for SiteCommunicationActor message routing to local actors. /// public class SiteCommunicationActorTests : TestKit { private readonly CommunicationOptions _options = new(); public SiteCommunicationActorTests() : base(@"akka.loglevel = DEBUG") { } [Fact] public void RefreshDeploymentCommand_ForwardedToDeploymentManager() { var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var command = new RefreshDeploymentCommand( "dep1", "inst1", "hash1", "admin", DateTimeOffset.UtcNow, "https://central/", "tok1"); siteActor.Tell(command); dmProbe.ExpectMsg(msg => msg.DeploymentId == "dep1"); } [Fact] public void LifecycleCommands_ForwardedToDeploymentManager() { var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); siteActor.Tell(new DisableInstanceCommand("cmd1", "inst1", DateTimeOffset.UtcNow)); dmProbe.ExpectMsg(); siteActor.Tell(new EnableInstanceCommand("cmd2", "inst1", DateTimeOffset.UtcNow)); dmProbe.ExpectMsg(); siteActor.Tell(new DeleteInstanceCommand("cmd3", "inst1", DateTimeOffset.UtcNow)); dmProbe.ExpectMsg(); } [Fact] public void DeploymentStateQuery_ForwardedToDeploymentManager() { // DeploymentManager-006: the site-before-redeploy query travels over the // ClusterClient command/control transport and is routed to the local // Deployment Manager, which owns the deployed-config store. var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var request = new DeploymentStateQueryRequest("corr-q", "inst1", DateTimeOffset.UtcNow); siteActor.Tell(request); dmProbe.ExpectMsg(msg => msg.CorrelationId == "corr-q"); } [Fact] public void IntegrationCall_WithoutHandler_ReturnsFailure() { var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var request = new IntegrationCallRequest( "corr1", "site1", "inst1", "ExtSys1", "GetData", new Dictionary(), DateTimeOffset.UtcNow); siteActor.Tell(request); ExpectMsg(msg => !msg.Success && msg.ErrorMessage == "Integration handler not available"); } [Fact] public void IntegrationCall_WithHandler_ForwardedToHandler() { var dmProbe = CreateTestProbe(); var handlerProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); // Register integration handler siteActor.Tell(new RegisterLocalHandler(LocalHandlerType.Integration, handlerProbe.Ref)); var request = new IntegrationCallRequest( "corr1", "site1", "inst1", "ExtSys1", "GetData", new Dictionary(), DateTimeOffset.UtcNow); siteActor.Tell(request); handlerProbe.ExpectMsg(msg => msg.CorrelationId == "corr1"); } // The site→central send-forwarding tests that used to live here (NotificationSubmit / // NotificationStatusQuery, with and without a registered central ClusterClient) were removed // with the Akka transport in the ClusterClient→gRPC migration's Phase 4: the actor no longer // owns the wire, it delegates to an injected ICentralTransport, and those seven sends (plus the // reply routing and the fail-loud fallback) are now covered by SiteCommunicationActorTransportTests. [Fact] public void EventLogQuery_WithoutHandler_ReturnsFailure() { var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var request = new EventLogQueryRequest( "corr1", "site1", null, null, null, null, null, null, null, 25, DateTimeOffset.UtcNow); siteActor.Tell(request); ExpectMsg(msg => !msg.Success); } // ── Task 5 (#22): central→site Retry/Discard relay for parked cached calls ── [Fact] public void RetryParkedOperation_WithHandler_ForwardedToParkedMessageHandler() { var dmProbe = CreateTestProbe(); var handlerProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); siteActor.Tell(new RegisterLocalHandler(LocalHandlerType.ParkedMessages, handlerProbe.Ref)); var id = Commons.Types.TrackedOperationId.New(); siteActor.Tell(new RetryParkedOperation("corr-rp", id)); handlerProbe.ExpectMsg(msg => msg.CorrelationId == "corr-rp" && msg.TrackedOperationId.Equals(id)); } [Fact] public void DiscardParkedOperation_WithHandler_ForwardedToParkedMessageHandler() { var dmProbe = CreateTestProbe(); var handlerProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); siteActor.Tell(new RegisterLocalHandler(LocalHandlerType.ParkedMessages, handlerProbe.Ref)); var id = Commons.Types.TrackedOperationId.New(); siteActor.Tell(new DiscardParkedOperation("corr-dp", id)); handlerProbe.ExpectMsg(msg => msg.CorrelationId == "corr-dp" && msg.TrackedOperationId.Equals(id)); } [Fact] public void RetryParkedOperation_WithoutHandler_RepliesNotAppliedAck() { // No parked-message handler registered — the relay must get a definitive // non-applied ack, not silence (the SiteCallAuditActor's Ask must not // hang and then mis-report site-unreachable when the site IS reachable). var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); siteActor.Tell(new RetryParkedOperation("corr-no-handler", Commons.Types.TrackedOperationId.New())); var ack = ExpectMsg(); Assert.Equal("corr-no-handler", ack.CorrelationId); Assert.False(ack.Applied); Assert.NotNull(ack.ErrorMessage); } [Fact] public void DiscardParkedOperation_WithoutHandler_RepliesNotAppliedAck() { var dmProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); siteActor.Tell(new DiscardParkedOperation("corr-no-handler", Commons.Types.TrackedOperationId.New())); var ack = ExpectMsg(); Assert.False(ack.Applied); Assert.NotNull(ack.ErrorMessage); } // ── M7 OPC UA cross-cluster routing: Search (T15), WriteTag (T14), Verify (T17) ── // // Regression guard for the M7 dead-letter defect. These three interactive // commands have downstream handlers in DataConnectionManagerActor but were NOT // forwarded through SiteCommunicationActor → Deployment Manager, so they // dead-lettered and the central Ask timed out in the real cluster. They must // forward to the Deployment Manager proxy exactly like BrowseNodeCommand, with // the original Ask sender preserved so the result routes straight back. [Fact] public void SearchAddressSpaceCommand_ForwardedToDeploymentManager_SenderPreserved() { var dmProbe = CreateTestProbe(); var senderProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var command = new SearchAddressSpaceCommand("conn1", "Temp", 5, 100); siteActor.Tell(command, senderProbe.Ref); dmProbe.ExpectMsg(msg => msg.ConnectionName == "conn1"); Assert.Equal(senderProbe.Ref, dmProbe.LastSender); } [Fact] public void WriteTagRequest_ForwardedToDeploymentManager_SenderPreserved() { var dmProbe = CreateTestProbe(); var senderProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var request = new WriteTagRequest( "corr-w", "conn1", "Channel1.Device1.Tag1", 42, DateTimeOffset.UtcNow); siteActor.Tell(request, senderProbe.Ref); dmProbe.ExpectMsg(msg => msg.CorrelationId == "corr-w"); Assert.Equal(senderProbe.Ref, dmProbe.LastSender); } [Fact] public void VerifyEndpointCommand_ForwardedToDeploymentManager_SenderPreserved() { var dmProbe = CreateTestProbe(); var senderProbe = CreateTestProbe(); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref))); var command = new VerifyEndpointCommand("conn1", "OpcUa", "{}"); siteActor.Tell(command, senderProbe.Ref); dmProbe.ExpectMsg(msg => msg.ConnectionName == "conn1"); Assert.Equal(senderProbe.Ref, dmProbe.LastSender); } // ── Communication-018: heartbeat IsActive reflects this node's cluster role ── [Theory] [InlineData(true)] [InlineData(false)] public void Heartbeat_StampsIsActive_FromInjectedCheck(bool isActive) { // Communication-018: HeartbeatMessage.IsActive must reflect the actual // active/standby role of this node, not a hard-coded `true`. The actor // now takes a Func override (defaulting to a real Akka.Cluster // leader check in production); tests inject a stub so they do not need // to bring up a full cluster in the TestKit ActorSystem. var dmProbe = CreateTestProbe(); var transport = Substitute.For(); var fastHeartbeatOptions = new CommunicationOptions { ApplicationHeartbeatInterval = TimeSpan.FromMilliseconds(100) }; Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", fastHeartbeatOptions, dmProbe.Ref, () => isActive, null, transport))); // The heartbeat now rides the injected ICentralTransport (SendHeartbeat), not a // ClusterClient. The captured message must carry the injected active/standby flag. AwaitAssert( () => transport.Received().SendHeartbeat( Arg.Is(h => h.IsActive == isActive && h.SiteId == "site1"), Arg.Any()), TimeSpan.FromSeconds(3)); } // ---- Central→site failover relay (Task 10) ---------------------------------- // The failover action is injected for the same reason isActiveCheck is: performing a // real Cluster.Leave would require Akka.Cluster in the TestKit ActorSystem. These // tests pin the ROUTING and GUARD logic; the actual Leave against a real two-node // site cluster is covered by SiteFailoverRelayTests in the integration suite. [Fact] public void TriggerSiteFailover_IssuesTheLeave_AndAcksWithTheTarget() { var dmProbe = CreateTestProbe(); string? roleAskedFor = null; Func failOver = role => { roleAskedFor = role; return "akka.tcp://scadabridge@site1-a:8082"; }; var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref, null, failOver))); siteActor.Tell(new TriggerSiteFailover("corr-1", "site1")); var ack = ExpectMsg(); Assert.True(ack.Accepted); Assert.Equal("corr-1", ack.CorrelationId); Assert.Equal("akka.tcp://scadabridge@site1-a:8082", ack.TargetAddress); Assert.Null(ack.ErrorMessage); // Site singletons are scoped to the site-specific role, not the base "Site" role; // failing over the wrong scope would move the wrong node. Assert.Equal("site-site1", roleAskedFor); } [Fact] public void TriggerSiteFailover_RefusesWhenThereIsNoPeer() { var dmProbe = CreateTestProbe(); Func failOver = _ => null; var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref, null, failOver))); siteActor.Tell(new TriggerSiteFailover("corr-2", "site1")); var ack = ExpectMsg(); Assert.False(ack.Accepted); Assert.Null(ack.TargetAddress); Assert.NotNull(ack.ErrorMessage); } [Fact] public void TriggerSiteFailover_RefusesACommandAddressedToAnotherSite() { // A misrouted command must be refused, not silently acted on — acting would fail // over a site the operator never selected. var dmProbe = CreateTestProbe(); var invoked = false; Func failOver = _ => { invoked = true; return "addr"; }; var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref, null, failOver))); siteActor.Tell(new TriggerSiteFailover("corr-3", "site2")); var ack = ExpectMsg(); Assert.False(ack.Accepted); Assert.Contains("site2", ack.ErrorMessage); Assert.False(invoked); } [Fact] public void TriggerSiteFailover_FaultInTheLeave_IsReportedNotThrown() { var dmProbe = CreateTestProbe(); Func failOver = _ => throw new InvalidOperationException("cluster unavailable"); var siteActor = Sys.ActorOf(Props.Create(() => new SiteCommunicationActor("site1", _options, dmProbe.Ref, null, failOver))); siteActor.Tell(new TriggerSiteFailover("corr-4", "site1")); var ack = ExpectMsg(); Assert.False(ack.Accepted); Assert.Contains("cluster unavailable", ack.ErrorMessage); } }