Files
scadalink-design/tests/ScadaLink.Communication.Tests/SiteCommunicationActorTests.cs
Joseph Doherty 3326bddeb0 feat(notification-outbox): async Notify.Send with status handle
Notify.To(list).Send(subject,body) now generates a NotificationId GUID,
enqueues a Notification-category message into the site Store-and-Forward
Engine, and returns the NotificationId immediately (Task<string>). The
NotificationId is the single idempotency key end-to-end: it is the S&F
message Id, it is carried inside the buffered NotificationSubmit payload,
and it is the id the forwarder submits to central.

NotificationForwarder now deserializes the buffered payload as a
NotificationSubmit and reads NotificationId from it (re-stamping only the
site-owned SourceSiteId / SourceInstanceId), instead of deriving the id
from StoreAndForwardMessage.Id.

Adds NotifyHelper.Status(id): queries central via the site communication
actor; reports the site-local Forwarding state while the notification is
still buffered at the site, maps central's response when found, and
Unknown otherwise. Adds a NotificationDeliveryStatus record.

SiteCommunicationActor gains a NotificationStatusQuery forwarding handler
mirroring NotificationSubmit. StoreAndForwardService.EnqueueAsync gains an
optional messageId parameter and exposes GetMessageByIdAsync.
2026-05-19 02:30:51 -04:00

218 lines
9.0 KiB
C#

using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.TestKit.Xunit2;
using ScadaLink.Commons.Messages.Deployment;
using ScadaLink.Commons.Messages.Lifecycle;
using ScadaLink.Commons.Messages.Integration;
using ScadaLink.Commons.Messages.Notification;
using ScadaLink.Commons.Messages.RemoteQuery;
using ScadaLink.Communication.Actors;
namespace ScadaLink.Communication.Tests;
/// <summary>
/// WP-4: Tests for SiteCommunicationActor message routing to local actors.
/// </summary>
public class SiteCommunicationActorTests : TestKit
{
private readonly CommunicationOptions _options = new();
public SiteCommunicationActorTests()
: base(@"akka.loglevel = DEBUG")
{
}
[Fact]
public void DeployCommand_ForwardedToDeploymentManager()
{
var dmProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
var command = new DeployInstanceCommand(
"dep1", "inst1", "hash1", "{}", "admin", DateTimeOffset.UtcNow);
siteActor.Tell(command);
dmProbe.ExpectMsg<DeployInstanceCommand>(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<DisableInstanceCommand>();
siteActor.Tell(new EnableInstanceCommand("cmd2", "inst1", DateTimeOffset.UtcNow));
dmProbe.ExpectMsg<EnableInstanceCommand>();
siteActor.Tell(new DeleteInstanceCommand("cmd3", "inst1", DateTimeOffset.UtcNow));
dmProbe.ExpectMsg<DeleteInstanceCommand>();
}
[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<DeploymentStateQueryRequest>(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<string, object?>(), DateTimeOffset.UtcNow);
siteActor.Tell(request);
ExpectMsg<IntegrationCallResponse>(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<string, object?>(), DateTimeOffset.UtcNow);
siteActor.Tell(request);
handlerProbe.ExpectMsg<IntegrationCallRequest>(msg => msg.CorrelationId == "corr1");
}
[Fact]
public void NotificationSubmit_WithCentralClient_ForwardedToCentralAndAckRoutedBack()
{
// The site forwards a buffered notification to central over the ClusterClient
// command/control transport; the central ack must route back to the original
// sender (the S&F forwarder's Ask), not to the SiteCommunicationActor.
var dmProbe = CreateTestProbe();
var centralClientProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
siteActor.Tell(new RegisterCentralClient(centralClientProbe.Ref));
var submit = new NotificationSubmit(
"notif-1", "Operators", "Subj", "Body", "site1", "inst1", "alarmScript",
DateTimeOffset.UtcNow);
siteActor.Tell(submit);
// Central client (acting as ClusterClient) receives a ClusterClient.Send wrapping
// the NotificationSubmit, addressed to the central communication actor. Fish past
// any periodic HeartbeatMessage the actor's timer may interleave.
var send = centralClientProbe.FishForMessage<ClusterClient.Send>(
s => s.Message is NotificationSubmit);
Assert.Equal("/user/central-communication", send.Path);
var forwarded = Assert.IsType<NotificationSubmit>(send.Message);
Assert.Equal("notif-1", forwarded.NotificationId);
// The ack is sent to the ClusterClient.Send's Sender — replying as that probe
// must land back at the test actor (the original Tell sender).
centralClientProbe.Reply(new NotificationSubmitAck("notif-1", Accepted: true, Error: null));
ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-1" && ack.Accepted);
}
[Fact]
public void NotificationSubmit_WithoutCentralClient_RepliesWithNonAccepted()
{
// No ClusterClient registered yet: the submit cannot be forwarded, so the actor
// replies with a non-accepted ack and the S&F forwarder treats it as transient.
var dmProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
var submit = new NotificationSubmit(
"notif-2", "Operators", "Subj", "Body", "site1", null, null,
DateTimeOffset.UtcNow);
siteActor.Tell(submit);
ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-2" && !ack.Accepted);
}
[Fact]
public void NotificationStatusQuery_WithCentralClient_ForwardedToCentralAndResponseRoutedBack()
{
// Notify.Status(id) issues a NotificationStatusQuery; the site actor forwards it
// to central over the ClusterClient command/control transport and the central
// response must route back to the original sender (the helper's Ask).
var dmProbe = CreateTestProbe();
var centralClientProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
siteActor.Tell(new RegisterCentralClient(centralClientProbe.Ref));
var query = new NotificationStatusQuery("corr-99", "notif-1");
siteActor.Tell(query);
var send = centralClientProbe.FishForMessage<ClusterClient.Send>(
s => s.Message is NotificationStatusQuery);
Assert.Equal("/user/central-communication", send.Path);
var forwarded = Assert.IsType<NotificationStatusQuery>(send.Message);
Assert.Equal("notif-1", forwarded.NotificationId);
// The response is sent to the ClusterClient.Send's Sender — replying as that
// probe must land back at the test actor (the original Tell sender).
centralClientProbe.Reply(new NotificationStatusResponse(
"corr-99", Found: true, Status: "Delivered", RetryCount: 0,
LastError: null, DeliveredAt: DateTimeOffset.UtcNow));
ExpectMsg<NotificationStatusResponse>(r => r.CorrelationId == "corr-99" && r.Found);
}
[Fact]
public void NotificationStatusQuery_WithoutCentralClient_RepliesWithNotFound()
{
// No ClusterClient registered yet: the query cannot reach central, so the actor
// replies Found: false. Notify.Status then falls back to the site S&F buffer.
var dmProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
siteActor.Tell(new NotificationStatusQuery("corr-100", "notif-2"));
ExpectMsg<NotificationStatusResponse>(
r => r.CorrelationId == "corr-100" && !r.Found);
}
[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<EventLogQueryResponse>(msg => !msg.Success);
}
}