Joseph Doherty
257 lines
12 KiB
C#
257 lines
12 KiB
C#
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.Instances;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
|
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.Communication.Tests;
|
|
|
|
/// <summary>
|
|
/// Tests for DebugStreamService session lifecycle.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Shares the <c>DebugStreamStatics</c> xUnit collection with <c>DebugStreamBridgeActorTests</c>
|
|
/// so the two classes never race on the actor's static test seams (see that class).
|
|
/// </remarks>
|
|
[Collection("DebugStreamStatics")]
|
|
public class DebugStreamServiceTests : TestKit
|
|
{
|
|
[Fact]
|
|
public async Task StartStreamAsync_StreamTerminatesBeforeSnapshot_ThrowsMeaningfulException()
|
|
{
|
|
// Regression test for Communication-001. When the debug stream terminates before
|
|
// the initial snapshot arrives, StartStreamAsync used to let the raw
|
|
// InvalidOperationException from onTerminatedWrapper escape its
|
|
// OperationCanceledException-only catch — the caller saw an untranslated exception
|
|
// and the failure path did not deterministically tear the bridge actor down.
|
|
// The fix catches any failure, tells the bridge actor StopDebugStream, and throws
|
|
// a descriptive exception that names the instance and wraps the underlying cause.
|
|
var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 };
|
|
var site = new Site("Site One", "site-1")
|
|
{
|
|
Id = 3,
|
|
GrpcNodeAAddress = "http://localhost:5100",
|
|
GrpcNodeBAddress = "http://localhost:5200"
|
|
};
|
|
|
|
var instanceRepo = Substitute.For<ITemplateEngineRepository>();
|
|
instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
|
|
var siteRepo = Substitute.For<ISiteRepository>();
|
|
siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).Returns(site);
|
|
|
|
var services = new ServiceCollection();
|
|
services.AddScoped(_ => instanceRepo);
|
|
services.AddScoped(_ => siteRepo);
|
|
using var provider = services.BuildServiceProvider();
|
|
|
|
var commProbe = CreateTestProbe();
|
|
var commService = new CommunicationService(
|
|
Options.Create(new CommunicationOptions()),
|
|
NullLogger<CommunicationService>.Instance);
|
|
commService.SetCommunicationActor(commProbe.Ref);
|
|
|
|
using var grpcFactory = new SiteStreamGrpcClientFactory(NullLoggerFactory.Instance);
|
|
var service = new DebugStreamService(
|
|
commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
|
|
service.SetActorSystem(Sys);
|
|
|
|
// Act — start the stream; it blocks awaiting the initial snapshot.
|
|
var startTask = service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { });
|
|
|
|
// The bridge actor's PreStart sends SubscribeDebugViewRequest to the comm actor;
|
|
// the envelope's sender is the bridge actor itself.
|
|
commProbe.ExpectMsg<SiteEnvelope>(TimeSpan.FromSeconds(5));
|
|
var bridgeActor = commProbe.LastSender;
|
|
|
|
// Simulate the site terminating the stream before any snapshot is delivered.
|
|
bridgeActor.Tell(new DebugStreamTerminated("site-1", "corr"));
|
|
|
|
// Assert — a descriptive exception that names the instance and wraps the cause,
|
|
// not the raw "terminated before snapshot received" InvalidOperationException.
|
|
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => startTask);
|
|
Assert.Contains("Site1.Pump01", ex.Message);
|
|
Assert.NotNull(ex.InnerException);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AttachedSession_IsKeptAliveByTheServiceKeepalive_AndDiesOnceDetached()
|
|
{
|
|
// The consumer half of the orphan net: holding a session in DebugStreamService IS
|
|
// "a consumer is attached", and the service's shared timer is what renews the bridge
|
|
// actor's window. Without this wiring every healthy session self-terminated one
|
|
// window after its snapshot (the old mailbox-based ReceiveTimeout had nothing
|
|
// recurring to reset it) and the consumer was told "Site disconnected".
|
|
var previousKeepalive = DebugStreamService.KeepaliveInterval;
|
|
var previousIdle = DebugStreamBridgeActor.ConsumerIdleTimeout;
|
|
var previousCheck = DebugStreamBridgeActor.ConsumerLivenessCheckInterval;
|
|
DebugStreamService.KeepaliveInterval = TimeSpan.FromMilliseconds(50);
|
|
DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400);
|
|
DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50);
|
|
try
|
|
{
|
|
var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 };
|
|
var site = new Site("Site One", "site-1")
|
|
{
|
|
Id = 3,
|
|
GrpcNodeAAddress = "http://localhost:5100",
|
|
GrpcNodeBAddress = "http://localhost:5200"
|
|
};
|
|
|
|
var instanceRepo = Substitute.For<ITemplateEngineRepository>();
|
|
instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
|
|
var siteRepo = Substitute.For<ISiteRepository>();
|
|
siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).Returns(site);
|
|
|
|
var services = new ServiceCollection();
|
|
services.AddScoped(_ => instanceRepo);
|
|
services.AddScoped(_ => siteRepo);
|
|
using var provider = services.BuildServiceProvider();
|
|
|
|
var commProbe = CreateTestProbe();
|
|
var commService = new CommunicationService(
|
|
Options.Create(new CommunicationOptions()),
|
|
NullLogger<CommunicationService>.Instance);
|
|
commService.SetCommunicationActor(commProbe.Ref);
|
|
|
|
// Mock gRPC factory: the real one would dial localhost:5100, fail, and trip the
|
|
// bridge actor's retry budget — a termination unrelated to the orphan net under
|
|
// test. The mock keeps the stream "up" so the only thing that can end this
|
|
// session is the consumer-liveness decision.
|
|
using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory(
|
|
new Grpc.MockSiteStreamGrpcClient());
|
|
using var service = new DebugStreamService(
|
|
commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
|
|
service.SetActorSystem(Sys);
|
|
|
|
var startTask = service.StartStreamAsync(
|
|
instanceId: 7, onEvent: _ => { }, onTerminated: () => { });
|
|
|
|
commProbe.ExpectMsg<SiteEnvelope>(TimeSpan.FromSeconds(5));
|
|
var bridgeActor = commProbe.LastSender;
|
|
Watch(bridgeActor);
|
|
|
|
// Resolve the snapshot so the session is fully established.
|
|
bridgeActor.Tell(new ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView.DebugViewSnapshot(
|
|
"Site1.Pump01",
|
|
new List<ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming.AttributeValueChanged>(),
|
|
new List<ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming.AlarmStateChanged>(),
|
|
DateTimeOffset.UtcNow));
|
|
var session = await startTask;
|
|
|
|
// Several orphan windows with NO traffic of any kind — only the service's
|
|
// keepalive timer. The session must survive.
|
|
await Task.Delay(TimeSpan.FromMilliseconds(1500));
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(100));
|
|
Assert.False(bridgeActor.IsNobody());
|
|
|
|
// Detach the consumer: the session leaves the registry, keepalives stop, and the
|
|
// actor is stopped (StopStream) — the orphan net is the backstop for the case
|
|
// where that explicit stop never happens.
|
|
service.StopStream(session.SessionId);
|
|
ExpectTerminated(bridgeActor, TimeSpan.FromSeconds(3));
|
|
}
|
|
finally
|
|
{
|
|
DebugStreamService.KeepaliveInterval = previousKeepalive;
|
|
DebugStreamBridgeActor.ConsumerIdleTimeout = previousIdle;
|
|
DebugStreamBridgeActor.ConsumerLivenessCheckInterval = previousCheck;
|
|
}
|
|
}
|
|
|
|
// --- arch-review 2026-08-19 P3-3: single-endpoint sites ---
|
|
|
|
[Theory]
|
|
[InlineData("http://localhost:5100", null)]
|
|
[InlineData(null, "http://localhost:5200")]
|
|
public async Task StartStreamAsync_SingleEndpointSite_UsesTheSoleAddressForBothNodeSlots(
|
|
string? nodeA, string? nodeB)
|
|
{
|
|
// Debug View used to hard-require BOTH gRPC node addresses and threw before ever
|
|
// creating the bridge actor, though single-endpoint sites are supported everywhere else
|
|
// (SiteAlarmLiveCacheService's N9 degeneration, SitePairChannelProvider's null-tolerant
|
|
// pair). The sole configured address now fills both slots, so the bridge actor's
|
|
// NodeA↔NodeB flip degenerates to a reconnect against the same node.
|
|
var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 };
|
|
var site = new Site("Site One", "site-1")
|
|
{
|
|
Id = 3,
|
|
GrpcNodeAAddress = nodeA,
|
|
GrpcNodeBAddress = nodeB
|
|
};
|
|
|
|
var instanceRepo = Substitute.For<ITemplateEngineRepository>();
|
|
instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
|
|
var siteRepo = Substitute.For<ISiteRepository>();
|
|
siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).Returns(site);
|
|
|
|
var services = new ServiceCollection();
|
|
services.AddScoped(_ => instanceRepo);
|
|
services.AddScoped(_ => siteRepo);
|
|
using var provider = services.BuildServiceProvider();
|
|
|
|
var commProbe = CreateTestProbe();
|
|
var commService = new CommunicationService(
|
|
Options.Create(new CommunicationOptions()),
|
|
NullLogger<CommunicationService>.Instance);
|
|
commService.SetCommunicationActor(commProbe.Ref);
|
|
|
|
using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory(
|
|
new Grpc.MockSiteStreamGrpcClient());
|
|
using var service = new DebugStreamService(
|
|
commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
|
|
service.SetActorSystem(Sys);
|
|
|
|
// Act — resolution must get past the address check and start a bridge actor.
|
|
var startTask = service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { });
|
|
|
|
commProbe.ExpectMsg<SiteEnvelope>(TimeSpan.FromSeconds(5));
|
|
var bridgeActor = commProbe.LastSender;
|
|
|
|
// The stream it opened dials the ONE configured address.
|
|
Assert.Equal(new[] { nodeA ?? nodeB }, grpcFactory.RequestedEndpoints);
|
|
|
|
// Unwind the session.
|
|
bridgeActor.Tell(new DebugStreamTerminated("site-1", "corr"));
|
|
await Assert.ThrowsAsync<InvalidOperationException>(() => startTask);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task StartStreamAsync_SiteWithNoGrpcAddresses_StillThrows()
|
|
{
|
|
// Degenerating a single endpoint must not degenerate into accepting NONE.
|
|
var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 };
|
|
var site = new Site("Site One", "site-1") { Id = 3 };
|
|
|
|
var instanceRepo = Substitute.For<ITemplateEngineRepository>();
|
|
instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
|
|
var siteRepo = Substitute.For<ISiteRepository>();
|
|
siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).Returns(site);
|
|
|
|
var services = new ServiceCollection();
|
|
services.AddScoped(_ => instanceRepo);
|
|
services.AddScoped(_ => siteRepo);
|
|
using var provider = services.BuildServiceProvider();
|
|
|
|
var commService = new CommunicationService(
|
|
Options.Create(new CommunicationOptions()),
|
|
NullLogger<CommunicationService>.Instance);
|
|
commService.SetCommunicationActor(CreateTestProbe().Ref);
|
|
|
|
using var grpcFactory = new SiteStreamGrpcClientFactory(NullLoggerFactory.Instance);
|
|
using var service = new DebugStreamService(
|
|
commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
|
|
service.SetActorSystem(Sys);
|
|
|
|
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() =>
|
|
service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { }));
|
|
Assert.Contains("no gRPC node addresses", ex.Message);
|
|
}
|
|
}
|