using Akka.Actor;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.TestHost;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
///
/// The site command plane's gRPC front door (T1B.2): the same PSK gate and readiness convention
/// as SiteStreamGrpcServer, and a decode→dispatch→encode round trip through the ONE
/// for one command in each of the six oneof groups. Runs
/// in-process over — no ports, no containers — with the interceptor
/// registered BY TYPE on AddGrpc (never pre-registered in DI), the shape that keeps
/// Grpc.AspNetCore's own activation in the picture. See
/// for why that matters.
///
public sealed class SiteCommandGrpcServiceTests : IDisposable
{
private const string SiteKey = "the-site-a-command-key";
private const string SiteId = "site-a";
private readonly ActorSystem _system = ActorSystem.Create("sitecmd-tests");
///
public void Dispose() => _system.Dispose();
/// A dispatcher whose Deployment Manager proxy is a canned-reply responder.
private SiteCommandDispatcher DispatcherWithResponder(out IActorRef responder)
{
responder = _system.ActorOf(Props.Create(() => new Responder()));
var dispatcher = new SiteCommandDispatcher(SiteId, responder, (_, _) => null);
// The parked handler is node-local; point it at the responder too so the Parked group can
// be exercised end-to-end (the null-guard path is covered by the dispatcher unit tests).
dispatcher.RegisterParkedMessageHandler(responder);
return dispatcher;
}
private SiteCommandGrpcService ReadyService()
{
var service = new SiteCommandGrpcService(NullLogger.Instance);
service.SetReady(DispatcherWithResponder(out _));
return service;
}
private static async Task StartHost(SiteCommandGrpcService service)
=> await new HostBuilder()
.ConfigureWebHost(web => web
.UseTestServer()
.ConfigureServices(services =>
{
// BY TYPE on AddGrpc — never AddSingleton the interceptor (see the sibling test).
services.AddGrpc(o => o.Interceptors.Add());
services.AddSingleton(Options.Create(new CommunicationOptions { GrpcPsk = SiteKey }));
services.AddSingleton(service);
})
.Configure(app =>
{
app.UseRouting();
app.UseEndpoints(e => e.MapGrpcService());
}))
.StartAsync();
private static SiteCommandService.SiteCommandServiceClient Client(IHost host, string? key)
{
var server = host.GetTestServer();
var options = new GrpcChannelOptions { HttpHandler = server.CreateHandler() };
if (key is not null)
{
options.WithSiteCredentials(new FixedPskProvider(key), SiteId);
}
var channel = GrpcChannel.ForAddress(server.BaseAddress, options);
return new SiteCommandService.SiteCommandServiceClient(channel);
}
private sealed class FixedPskProvider(string key) : ISitePskProvider
{
public ValueTask GetAsync(string siteId, CancellationToken ct) => new(key);
public void Invalidate(string siteId) { }
}
// ── Auth gating (delegated to ControlPlaneAuthInterceptor, proven wired here) ──
[Fact]
public async Task NoCredentials_IsRejected_WithPermissionDenied()
{
using var host = await StartHost(ReadyService());
var client = Client(host, key: null);
var ex = await Assert.ThrowsAsync(() => client.ExecuteQueryAsync(
new QueryRequest { UnsubscribeDebugView = new UnsubscribeDebugViewRequestDto() }).ResponseAsync);
Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode);
}
[Fact]
public async Task WrongKey_IsRejected_WithPermissionDenied()
{
using var host = await StartHost(ReadyService());
var client = Client(host, "some-other-key");
var ex = await Assert.ThrowsAsync(() => client.ExecuteQueryAsync(
new QueryRequest { UnsubscribeDebugView = new UnsubscribeDebugViewRequestDto() }).ResponseAsync);
Assert.Equal(StatusCode.PermissionDenied, ex.StatusCode);
}
// ── Readiness ──
[Fact]
public async Task BeforeSetReady_IsRejected_WithUnavailable_EvenWithACorrectKey()
{
// Auth passes (correct key); the readiness gate then rejects until the site actor graph is up.
var unready = new SiteCommandGrpcService(NullLogger.Instance);
using var host = await StartHost(unready);
var client = Client(host, SiteKey);
var ex = await Assert.ThrowsAsync(() => client.ExecuteLifecycleAsync(
new LifecycleRequest { EnableInstance = new EnableInstanceCommandDto { CommandId = "c" } }).ResponseAsync);
Assert.Equal(StatusCode.Unavailable, ex.StatusCode);
}
// ── decode → dispatch → encode, one command per oneof group ──
[Fact]
public async Task ExecuteLifecycle_RoundTripsThroughTheDispatcher()
{
using var host = await StartHost(ReadyService());
var client = Client(host, SiteKey);
var command = new EnableInstanceCommand("cmd-1", "Site1.Pump1", DateTimeOffset.UtcNow);
var reply = await client.ExecuteLifecycleAsync(
new LifecycleRequest { EnableInstance = SiteCommandDtoMapper.ToProto(command) });
var decoded = Assert.IsType(SiteCommandDtoMapper.FromLifecycleReply(reply));
Assert.True(decoded.Success);
Assert.Equal("cmd-1", decoded.CommandId);
}
[Fact]
public async Task ExecuteOpcUa_RoundTripsThroughTheDispatcher()
{
using var host = await StartHost(ReadyService());
var client = Client(host, SiteKey);
var command = new BrowseNodeCommand("conn", null, null, null);
var reply = await client.ExecuteOpcUaAsync(
new OpcUaRequest { BrowseNode = SiteCommandDtoMapper.ToProto(command) });
Assert.IsType(SiteCommandDtoMapper.FromOpcUaReply(reply));
}
[Fact]
public async Task ExecuteQuery_DebugSnapshot_RoundTripsThroughTheDispatcher()
{
using var host = await StartHost(ReadyService());
var client = Client(host, SiteKey);
var command = new DebugSnapshotRequest("Site1.Pump1", "corr-4");
var reply = await client.ExecuteQueryAsync(
new QueryRequest { DebugSnapshot = SiteCommandDtoMapper.ToProto(command) });
var decoded = Assert.IsType(SiteCommandDtoMapper.FromQueryReply(reply));
Assert.Equal("Site1.Pump1", decoded.InstanceUniqueName);
}
[Fact]
public async Task ExecuteQuery_UnsubscribeDebugView_ReturnsTheFireAndForgetAck()
{
// Fire-and-forget: the service Tells the target and returns the synthetic ack so a unary
// RPC still answers, keeping the caller-visible fire-and-forget semantics.
using var host = await StartHost(ReadyService());
var client = Client(host, SiteKey);
var reply = await client.ExecuteQueryAsync(new QueryRequest
{
UnsubscribeDebugView = SiteCommandDtoMapper.ToProto(
new UnsubscribeDebugViewRequest("Site1.Pump1", "corr-6")),
});
Assert.Equal(QueryReply.ReplyOneofCase.UnsubscribeDebugView, reply.ReplyCase);
}
[Fact]
public async Task ExecuteParked_RoundTripsThroughTheNodeLocalHandler()
{
using var host = await StartHost(ReadyService());
var client = Client(host, SiteKey);
var command = new ParkedMessageQueryRequest("corr-7", SiteId, 2, 25, DateTimeOffset.UtcNow);
var reply = await client.ExecuteParkedAsync(
new ParkedRequest { ParkedMessageQuery = SiteCommandDtoMapper.ToProto(command) });
var decoded = Assert.IsType(SiteCommandDtoMapper.FromParkedReply(reply));
Assert.True(decoded.Success);
Assert.Equal("corr-7", decoded.CorrelationId);
}
[Fact]
public async Task ExecuteRoute_RoundTripsThroughTheDispatcher()
{
using var host = await StartHost(ReadyService());
var client = Client(host, SiteKey);
var command = new RouteToCallRequest(
"corr-12", "Site1.Pump1", "Start", new Dictionary(), DateTimeOffset.UtcNow, null);
var reply = await client.ExecuteRouteAsync(
new RouteRequest { RouteToCall = SiteCommandDtoMapper.ToProto(command) });
var decoded = Assert.IsType(SiteCommandDtoMapper.FromRouteReply(reply));
Assert.True(decoded.Success);
Assert.Equal("corr-12", decoded.CorrelationId);
}
// ── Failover: the reply completes before the leave is initiated ──
[Fact]
public async Task TriggerFailover_ReturnsTheAck_BeforeTheLeaveIsInitiated()
{
// The resolver records the order of its calls: PrepareFailover does a DRY-RUN resolve to
// build the ack (recorded synchronously, before the RPC returns), and the real leave runs
// only on the deferred CommitLeave — so the recorded order is always resolve-then-leave,
// i.e. the ack is on the wire before the node begins leaving.
var events = new List();
var leaveHappened = new ManualResetEventSlim(false);
Func resolve = (_, dryRun) =>
{
lock (events) { events.Add(dryRun ? "resolve" : "leave"); }
if (!dryRun) leaveHappened.Set();
return "akka.tcp://scadabridge@site-a-node-a:8082";
};
var service = new SiteCommandGrpcService(NullLogger.Instance);
var dispatcher = new SiteCommandDispatcher(SiteId, _system.ActorOf(Props.Create(() => new Responder())), resolve);
service.SetReady(dispatcher);
using var host = await StartHost(service);
var client = Client(host, SiteKey);
var reply = await client.TriggerFailoverAsync(
new TriggerSiteFailoverDto { CorrelationId = "corr-16", SiteId = SiteId });
Assert.True(reply.Accepted);
Assert.Equal("akka.tcp://scadabridge@site-a-node-a:8082", reply.TargetAddress);
Assert.True(leaveHappened.Wait(TimeSpan.FromSeconds(5)), "the deferred leave never ran");
lock (events)
{
Assert.Equal(new[] { "resolve", "leave" }, events);
}
}
/// Deployment Manager stand-in: replies each command with a canned reply of the right group.
private sealed class Responder : ReceiveActor
{
public Responder() => ReceiveAny(msg => Sender.Tell(ReplyFor(msg)));
private static object ReplyFor(object m) => m switch
{
EnableInstanceCommand e =>
new InstanceLifecycleResponse(e.CommandId, e.InstanceUniqueName, true, null, DateTimeOffset.UtcNow),
BrowseNodeCommand => new BrowseNodeResult([], false, null, null),
DebugSnapshotRequest d => new DebugViewSnapshot(d.InstanceUniqueName, [], [], DateTimeOffset.UtcNow),
RouteToCallRequest r =>
new RouteToCallResponse(r.CorrelationId, true, null, null, DateTimeOffset.UtcNow),
ParkedMessageQueryRequest p => new ParkedMessageQueryResponse(
p.CorrelationId, p.SiteId, [], 0, p.PageNumber, p.PageSize, true, null, DateTimeOffset.UtcNow),
_ => new Akka.Actor.Status.Failure(new InvalidOperationException($"no canned reply for {m.GetType().Name}")),
};
}
}