lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshRouterDispatchTests.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Communication;

/// <summary>
///     Proves the five central→node publish sites go through the transport router when one is
///     wired, and fall back to the mediator when it is not.
/// </summary>
/// <remarks>
///     The fallback is what keeps the ~12 pre-existing coordinator and admin-operations tests
///     meaningful: they construct these actors without a router and must still exercise the real
///     DistributedPubSub path, not a silently-disabled one.
/// </remarks>
public class MeshRouterDispatchTests : ControlPlaneActorTestBase
{
    private const string DeploymentsTopic = "deployments";

    [Fact]
    public void Deploy_dispatch_goes_to_the_router_when_one_is_wired()
    {
        var router = CreateTestProbe();
        var dbFactory = NewInMemoryDbFactory();
        var coordinator = Sys.ActorOf(
            ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: router.Ref));

        var deploymentId = new DeploymentId(Guid.NewGuid());
        coordinator.Tell(new Commons.Messages.Deploy.DispatchDeployment(
            deploymentId, new RevisionHash("rev-1"), CorrelationId.NewId()));

        var cmd = router.ExpectMsg<MeshCommand>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        cmd.Topic.ShouldBe(DeploymentsTopic);
        cmd.Message.ShouldBeOfType<Commons.Messages.Deploy.DispatchDeployment>();
    }

    [Fact]
    public void Deploy_dispatch_falls_back_to_the_mediator_when_no_router_is_wired()
    {
        var subscriber = CreateTestProbe();
        DistributedPubSub.Get(Sys).Mediator.Tell(
            new Subscribe(DeploymentsTopic, subscriber.Ref), subscriber.Ref);
        subscriber.ExpectMsg<SubscribeAck>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);

        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(NewInMemoryDbFactory()));

        var deploymentId = new DeploymentId(Guid.NewGuid());
        coordinator.Tell(new Commons.Messages.Deploy.DispatchDeployment(
            deploymentId, new RevisionHash("rev-1"), CorrelationId.NewId()));

        // Presence budget, not 5 s: the coordinator's first ConfigDb (EF InMemory) read sits in front
        // of the dispatch and is a cold multi-second cost on a loaded CI runner (#606).
        subscriber.ExpectMsg<Commons.Messages.Deploy.DispatchDeployment>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
    }

    [Fact]
    public void Restart_driver_goes_to_the_router_on_the_driver_control_topic()
    {
        var router = CreateTestProbe();
        var admin = Sys.ActorOf(AdminOperationsActor.Props(
            NewInMemoryDbFactory(),
            CreateTestProbe().Ref,
            Enumerable.Empty<IDriverProbe>(),
            TagConfigValidationMode.Warn,
            router.Ref));

        admin.Tell(new RestartDriver("C1", "driver-1", "alice", Guid.NewGuid()));

        var cmd = router.ExpectMsg<MeshCommand>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        cmd.Topic.ShouldBe(DriverControlTopic.Name);
        cmd.Message.ShouldBeOfType<RestartDriver>();
    }

    [Fact]
    public void Reconnect_driver_goes_to_the_router_on_the_driver_control_topic()
    {
        var router = CreateTestProbe();
        var admin = Sys.ActorOf(AdminOperationsActor.Props(
            NewInMemoryDbFactory(),
            CreateTestProbe().Ref,
            Enumerable.Empty<IDriverProbe>(),
            TagConfigValidationMode.Warn,
            router.Ref));

        admin.Tell(new ReconnectDriver("C1", "driver-1", "alice", Guid.NewGuid()));

        var cmd = router.ExpectMsg<MeshCommand>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        cmd.Topic.ShouldBe(DriverControlTopic.Name);
        cmd.Message.ShouldBeOfType<ReconnectDriver>();
    }

    [Fact]
    public void Alarm_acknowledge_goes_to_the_router_on_the_alarm_commands_topic()
    {
        var router = CreateTestProbe();
        var admin = Sys.ActorOf(AdminOperationsActor.Props(
            NewInMemoryDbFactory(),
            CreateTestProbe().Ref,
            Enumerable.Empty<IDriverProbe>(),
            TagConfigValidationMode.Warn,
            router.Ref));

        admin.Tell(new AcknowledgeAlarmCommand("alarm-1", "alice", "c", CorrelationId.NewId()));

        var cmd = router.ExpectMsg<MeshCommand>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        cmd.Topic.ShouldBe(AlarmCommandsTopic.Name);
        cmd.Message.ShouldBeOfType<AlarmCommand>();
    }

    [Fact]
    public void Alarm_shelve_goes_to_the_router_on_the_alarm_commands_topic()
    {
        var router = CreateTestProbe();
        var admin = Sys.ActorOf(AdminOperationsActor.Props(
            NewInMemoryDbFactory(),
            CreateTestProbe().Ref,
            Enumerable.Empty<IDriverProbe>(),
            TagConfigValidationMode.Warn,
            router.Ref));

        admin.Tell(new ShelveAlarmCommand("alarm-1", "alice", ShelveKind.OneShot, null, "c", CorrelationId.NewId()));

        var cmd = router.ExpectMsg<MeshCommand>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        cmd.Topic.ShouldBe(AlarmCommandsTopic.Name);
        cmd.Message.ShouldBeOfType<AlarmCommand>();
    }
}