using Akka.Actor; using Akka.TestKit.Xunit2; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using NSubstitute; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; using ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment; using ZB.MOM.WW.ScadaBridge.Communication.Actors; namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; /// /// Tests that routes a site→central /// through the scoped /// and pipes the resulting back to the original /// sender (the site's transport path). Mirrors the audit-ingest routing tests. /// public class CentralCommunicationActorReconcileTests : TestKit { /// Verifies that the response to a reconcile request is routed back to the original sender. [Fact] public void ReconcileSiteRequest_RoutesResponseToSender() { var deploymentRepo = Substitute.For(); var siteRepo = Substitute.For(); // GetAllSitesAsync is called by the actor's periodic refresh; keep it empty. siteRepo.GetAllSitesAsync(Arg.Any()) .Returns(new List()); siteRepo.GetSiteByIdentifierAsync("site1", Arg.Any()) .Returns(new Site("Site One", "site1") { Id = 7 }); deploymentRepo.GetExpectedDeploymentsForSiteAsync(7, Arg.Any()) .Returns(new List { new(2, "inst-B", "rev2", "dep-B", true), }); deploymentRepo.GetDeployedSnapshotsByInstanceIdsAsync( Arg.Any>(), Arg.Any()) .Returns(new Dictionary { [2] = new("dep-B", "rev2", "{\"cfg\":\"B\"}") { InstanceId = 2 }, }); // Every candidate is staged fresh: the batch hands each one back as its live row. deploymentRepo.StagePendingBatchIfAbsentAsync( Arg.Any>(), Arg.Any(), Arg.Any()) .Returns(ci => (IReadOnlyDictionary)ci.ArgAt>(0) .ToDictionary(p => p.InstanceId)); var options = Options.Create(new CommunicationOptions { CentralFetchBaseUrl = "https://central.example:9000", PendingDeploymentTtl = TimeSpan.FromMinutes(5), }); var services = new ServiceCollection(); services.AddScoped(_ => deploymentRepo); services.AddScoped(_ => siteRepo); services.AddSingleton(options); services.AddSingleton>( NullLogger.Instance); services.AddScoped(); var sp = services.BuildServiceProvider(); var transport = Substitute.For(); var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); // Node B is missing inst-B entirely → it should come back as a gap item. actor.Tell(new ReconcileSiteRequest( "site1", "node-b", new Dictionary())); var response = ExpectMsg(TimeSpan.FromSeconds(5)); var gap = Assert.Single(response.Gap); Assert.Equal("inst-B", gap.InstanceUniqueName); Assert.Equal("dep-B", gap.DeploymentId); Assert.False(string.IsNullOrWhiteSpace(gap.FetchToken)); } /// Verifies that cancelling the originating gRPC call stops central reconcile work (review 2026-10-21 03/P-1b). [Fact] public void CancelledCall_StopsCentralReconcileWork() { var deploymentRepo = Substitute.For(); var siteRepo = Substitute.For(); siteRepo.GetAllSitesAsync(Arg.Any()).Returns(new List()); var lookupStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); CancellationToken observed = default; siteRepo.GetSiteByIdentifierAsync("site1", Arg.Any()).Returns(async ci => { observed = ci.Arg(); lookupStarted.TrySetResult(); await Task.Delay(Timeout.Infinite, observed); // a slow central DB return (Site?)new Site("Site One", "site1") { Id = 7 }; }); var services = new ServiceCollection(); services.AddScoped(_ => deploymentRepo); services.AddScoped(_ => siteRepo); services.AddSingleton(Options.Create(new CommunicationOptions())); services.AddSingleton>( NullLogger.Instance); services.AddScoped(); var sp = services.BuildServiceProvider(); var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor( sp, Substitute.For()))); using var call = new CancellationTokenSource(); actor.Tell(new ReconcileSiteCall( new ReconcileSiteRequest("site1", "node-b", new Dictionary()), call.Token)); AwaitCondition(() => lookupStarted.Task.IsCompleted, TimeSpan.FromSeconds(3)); // The site gives up (deadline / disconnect). call.Cancel(); AwaitCondition(() => observed.IsCancellationRequested, TimeSpan.FromSeconds(3)); ExpectMsg(TimeSpan.FromSeconds(3)); deploymentRepo.DidNotReceiveWithAnyArgs().GetExpectedDeploymentsForSiteAsync(default, default); } }