lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/NodeCommandGrpcServiceTests.cs

using Akka.Hosting;
using Akka.TestKit.Xunit2;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Host.Grpc;
using ZB.MOM.WW.OtOpcUa.Runtime;

namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Grpc;

/// <summary>
///     Verifies the node-side command ingress (ClusterClient→gRPC migration Phase 1): each of the four
///     central→node commands is decoded off its <see cref="CommandEnvelope"/> and handed to the
///     registered <c>NodeCommunicationActor</c> as the same CLR record the ClusterClient transport
///     delivered, an undecodable envelope is refused with <see cref="StatusCode.InvalidArgument"/>, and
///     a call landing in the boot window before the actor is registered is refused with
///     <see cref="StatusCode.Unavailable"/>.
/// </summary>
/// <remarks>
///     The actor is a <c>TestProbe</c> registered under <see cref="NodeCommunicationActorKey"/>: what
///     this service owes the rest of the system is the exact message the real actor already knows how to
///     re-emit, so asserting on what arrives at the registry key is the whole contract. The onward leg —
///     comm actor to EventStream to the real driver host — is pinned unchanged by
///     <c>EventStreamCommandDeliveryTests</c> in the Runtime test project.
/// </remarks>
public sealed class NodeCommandGrpcServiceTests : TestKit
{
    private static readonly CorrelationId AlarmCorrelation = new(Guid.NewGuid());

    private static ServerCallContext Ctx() => new FakeServerCallContext();

    private static NodeCommandGrpcService NewService(ActorRegistry registry) =>
        new(registry, NullLogger<NodeCommandGrpcService>.Instance);

    [Fact]
    public async Task Each_of_the_four_commands_reaches_the_comm_actor_as_its_domain_record()
    {
        var probe = CreateTestProbe();
        var registry = new ActorRegistry();
        registry.Register<NodeCommunicationActorKey>(probe.Ref);
        var service = NewService(registry);

        var dispatch = new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("rev-1"), new CorrelationId(Guid.NewGuid()));
        var restart = new RestartDriver("cluster-a", "drv-1", "alice", Guid.NewGuid());
        var reconnect = new ReconnectDriver("cluster-a", "drv-2", "bob", Guid.NewGuid());
        var alarm = new AlarmCommand("alarm-7", "Acknowledge", "carol", "looked at it", null);

        (await service.Execute(MeshTransportDtoMapper.ToEnvelope(dispatch), Ctx())).ShouldNotBeNull();
        probe.ExpectMsg<DispatchDeployment>().ShouldBe(dispatch);

        await service.Execute(MeshTransportDtoMapper.ToEnvelope(restart), Ctx());
        probe.ExpectMsg<RestartDriver>().ShouldBe(restart);

        await service.Execute(MeshTransportDtoMapper.ToEnvelope(reconnect), Ctx());
        probe.ExpectMsg<ReconnectDriver>().ShouldBe(reconnect);

        await service.Execute(MeshTransportDtoMapper.ToEnvelope(alarm, AlarmCorrelation), Ctx());
        probe.ExpectMsg<AlarmCommand>().ShouldBe(alarm);
    }

    [Fact]
    public async Task A_call_in_the_boot_window_before_the_actor_is_registered_is_unavailable()
    {
        // Nothing registered: the ActorSystem exists but AddOtOpcUaRuntime has not spawned the comm
        // actor yet. Unavailable (not Internal) because the condition is transient and retryable —
        // central's dialer may try the same node again a moment later and succeed.
        var service = NewService(new ActorRegistry());
        var dispatch = new DispatchDeployment(
            new DeploymentId(Guid.NewGuid()), new RevisionHash("rev-1"), new CorrelationId(Guid.NewGuid()));

        var ex = await Should.ThrowAsync<RpcException>(
            () => service.Execute(MeshTransportDtoMapper.ToEnvelope(dispatch), Ctx()));

        ex.StatusCode.ShouldBe(StatusCode.Unavailable);
    }

    [Fact]
    public async Task An_unset_oneof_is_invalid_argument()
    {
        var probe = CreateTestProbe();
        var registry = new ActorRegistry();
        registry.Register<NodeCommunicationActorKey>(probe.Ref);
        var service = NewService(registry);

        var ex = await Should.ThrowAsync<RpcException>(() => service.Execute(
            new CommandEnvelope { CorrelationId = Guid.NewGuid().ToString("N") }, Ctx()));

        ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }

    [Theory]
    [MemberData(nameof(UndecodableEnvelopes))]
    public async Task An_envelope_whose_fields_do_not_parse_is_invalid_argument(CommandEnvelope envelope)
    {
        var probe = CreateTestProbe();
        var registry = new ActorRegistry();
        registry.Register<NodeCommunicationActorKey>(probe.Ref);
        var service = NewService(registry);

        var ex = await Should.ThrowAsync<RpcException>(() => service.Execute(envelope, Ctx()));

        ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }

    public static TheoryData<CommandEnvelope> UndecodableEnvelopes() =>
    [
        // Deployment id is not a hyphenless GUID -> FormatException out of the mapper.
        new CommandEnvelope
        {
            CorrelationId = Guid.NewGuid().ToString("N"),
            DispatchDeployment = new DispatchDeploymentCommand
            {
                DeploymentId = "not-a-guid",
                RevisionHash = "rev-1",
                CorrelationId = Guid.NewGuid().ToString("N"),
            },
        },
        // Revision hash is blank -> ArgumentException out of RevisionHash.Parse.
        new CommandEnvelope
        {
            CorrelationId = Guid.NewGuid().ToString("N"),
            DispatchDeployment = new DispatchDeploymentCommand
            {
                DeploymentId = Guid.NewGuid().ToString("N"),
                RevisionHash = "",
                CorrelationId = Guid.NewGuid().ToString("N"),
            },
        },
        // Restart command with an unparseable correlation id -> FormatException.
        new CommandEnvelope
        {
            CorrelationId = Guid.NewGuid().ToString("N"),
            RestartDriver = new RestartDriverCommand
            {
                ClusterId = "cluster-a",
                DriverInstanceId = "drv-1",
                ActorByUserName = "alice",
                CorrelationId = "nope",
            },
        },
    ];

    [Fact]
    public async Task An_undecodable_envelope_in_the_boot_window_is_invalid_argument_not_unavailable()
    {
        // Ordering pin: decode runs BEFORE the registry lookup. Unavailable invites the caller to
        // retry, and an envelope this build cannot decode will never decode — reporting it as a
        // readiness problem would turn a permanent fault into an unbounded retry loop.
        var service = NewService(new ActorRegistry());

        var ex = await Should.ThrowAsync<RpcException>(() => service.Execute(
            new CommandEnvelope { CorrelationId = Guid.NewGuid().ToString("N") }, Ctx()));

        ex.StatusCode.ShouldBe(StatusCode.InvalidArgument);
    }

    /// <summary>Minimal <see cref="ServerCallContext"/>; this RPC reads nothing off the context but its method name.</summary>
    private sealed class FakeServerCallContext : ServerCallContext
    {
        protected override string MethodCore => "/mesh_transport.v1.NodeCommandService/Execute";
        protected override string HostCore => "localhost";
        protected override string PeerCore => "ipv4:127.0.0.1:0";
        protected override DateTime DeadlineCore => DateTime.MaxValue;
        protected override Metadata RequestHeadersCore => [];
        protected override CancellationToken CancellationTokenCore => CancellationToken.None;
        protected override Metadata ResponseTrailersCore { get; } = [];
        protected override Status StatusCore { get; set; }
        protected override WriteOptions? WriteOptionsCore { get; set; }
        protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());

        protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
            throw new NotSupportedException();

        protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
    }
}