46 lines
1.9 KiB
C#
46 lines
1.9 KiB
C#
using Akka.Actor;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
|
using ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet;
|
|
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
|
|
|
|
public sealed class FleetStatusBroadcasterTests : ControlPlaneActorTestBase
|
|
{
|
|
[Fact]
|
|
public void Self_member_up_lands_in_snapshot()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
Sys.ActorOf(FleetStatusBroadcaster.Props(broadcast: msg => probe.Ref.Tell(msg)));
|
|
|
|
// The cluster self-joined in the harness — InitialStateAsEvents replays MemberUp for self.
|
|
var snapshot = probe.ExpectMsg<FleetStatusChanged>(TimeSpan.FromSeconds(5));
|
|
snapshot.Nodes.Count.ShouldBeGreaterThanOrEqualTo(1);
|
|
snapshot.Nodes.Any(n => n.Health == FleetNodeHealth.Healthy).ShouldBeTrue();
|
|
}
|
|
|
|
[Fact]
|
|
public void Heartbeat_updates_revision_in_next_snapshot()
|
|
{
|
|
var probe = CreateTestProbe();
|
|
var actor = Sys.ActorOf(FleetStatusBroadcaster.Props(broadcast: msg => probe.Ref.Tell(msg)));
|
|
|
|
// Drain the initial MemberUp snapshot.
|
|
probe.ExpectMsg<FleetStatusChanged>(TimeSpan.FromSeconds(5));
|
|
|
|
// Send a heartbeat for a node we know about (the self node from the cluster).
|
|
var selfHost = Akka.Cluster.Cluster.Get(Sys).SelfAddress.Host!;
|
|
var rev = RevisionHash.Parse(new string('c', 64));
|
|
actor.Tell(new FleetStatusBroadcaster.DriverHostStatusHeartbeat(NodeId.Parse(selfHost), rev));
|
|
|
|
// Wait for next periodic broadcast and assert the revision propagated.
|
|
var next = probe.FishForMessage<FleetStatusChanged>(
|
|
isMessage: m => m.Nodes.Any(n => n.CurrentRevision == rev),
|
|
max: TimeSpan.FromSeconds(10));
|
|
next.Nodes.Single(n => n.NodeId.Value == selfHost).CurrentRevision.ShouldBe(rev);
|
|
}
|
|
}
|