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.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");
}
// 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);
}
}