using Akka.Actor; using Akka.TestKit.Xunit2; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using NSubstitute; using NSubstitute.ExceptionExtensions; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Serialization; using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; /// /// ScadaBridge#138 at the OPC UA adapter: /// against a substitute client (status / failure mapping) and against real in-process OPC UA /// servers (the actual i=2267 node), plus the whole failover through the real /// against a real server pair. /// public sealed class OpcUaServiceLevelTests : TestKit { private readonly string _clientPki = Path.Combine(Path.GetTempPath(), "sb138-client", Guid.NewGuid().ToString("N")); /// Initializes the test kit. public OpcUaServiceLevelTests() : base(@"akka.loglevel = DEBUG") { } /// protected override void Dispose(bool disposing) { base.Dispose(disposing); try { Directory.Delete(_clientPki, recursive: true); } catch { /* best-effort */ } } private OpcUaGlobalOptions ClientOptions() => new() { ApplicationName = "SB138Client", ApplicationUri = "urn:localhost:SB138Client", OwnCertificateStorePath = Path.Combine(_clientPki, "own"), TrustedPeerStorePath = Path.Combine(_clientPki, "trusted"), TrustedIssuerStorePath = Path.Combine(_clientPki, "issuers"), RejectedCertificateStorePath = Path.Combine(_clientPki, "rejected"), AutoCreateApplicationCertificate = true, }; private static IDictionary Flat(string url, OpcUaServiceLevelFailoverConfig? sl = null) => OpcUaEndpointConfigSerializer.ToFlatDict(new OpcUaEndpointConfig { EndpointUrl = url, SecurityMode = OpcUaSecurityMode.None, ServiceLevelFailover = sl, }); private static (OpcUaDataConnection Adapter, IOpcUaClient Client) WithSubstituteClient() { var client = Substitute.For(); client.IsConnected.Returns(true); var factory = Substitute.For(); factory.Create().Returns(client); return (new OpcUaDataConnection(factory, NullLogger.Instance), client); } // ── Status / failure mapping (substitute client) ── /// The value comes from i=2267; a Good read of a byte is that byte. [Fact] public async Task Read_GoodByte_IsTheLevel_FromTheServiceLevelNode() { var (adapter, client) = WithSubstituteClient(); client.ReadValueAsync("i=2267", Arg.Any()) .Returns(((object?)(byte)90, DateTime.UtcNow, 0u)); await adapter.ConnectAsync(new Dictionary()); Assert.Equal((byte)90, await adapter.ReadServiceLevelAsync()); } /// A Bad status, or a value that is not a byte-ranged integer, is unknown (null). [Fact] public async Task Read_BadStatusOrNonByteValue_IsUnknown() { var (adapter, client) = WithSubstituteClient(); await adapter.ConnectAsync(new Dictionary()); client.ReadValueAsync("i=2267", Arg.Any()) .Returns(((object?)(byte)250, DateTime.UtcNow, 0x80340000u)); // BadNodeIdUnknown Assert.Null(await adapter.ReadServiceLevelAsync()); client.ReadValueAsync("i=2267", Arg.Any()) .Returns(((object?)"250", DateTime.UtcNow, 0u)); Assert.Null(await adapter.ReadServiceLevelAsync()); client.ReadValueAsync("i=2267", Arg.Any()) .Returns(((object?)300, DateTime.UtcNow, 0u)); Assert.Null(await adapter.ReadServiceLevelAsync()); client.ReadValueAsync("i=2267", Arg.Any()) .Returns(((object?)240, DateTime.UtcNow, 0u)); // Int32 in range → accepted Assert.Equal((byte)240, await adapter.ReadServiceLevelAsync()); } /// /// A failed read is unknown and — unlike — never /// raises Disconnected: the poll must not drive the reconnect machine. /// [Fact] public async Task Read_Failure_IsUnknown_AndNeverRaisesDisconnected() { var (adapter, client) = WithSubstituteClient(); client.ReadValueAsync("i=2267", Arg.Any()) .ThrowsAsync(new InvalidOperationException("session dropped")); var disconnected = 0; adapter.Disconnected += () => disconnected++; await adapter.ConnectAsync(new Dictionary()); Assert.Null(await adapter.ReadServiceLevelAsync()); Assert.Equal(0, disconnected); // Positive control: the ordinary read path DOES raise it for the same failure. client.ReadValueAsync("ns=2;s=x", Arg.Any()) .ThrowsAsync(new InvalidOperationException("session dropped")); await Assert.ThrowsAsync(() => adapter.ReadAsync("ns=2;s=x")); Assert.Equal(1, disconnected); } /// Before connect (no client) the level is unknown, and no read is attempted. [Fact] public async Task Read_NotConnected_IsUnknown() { var (adapter, client) = WithSubstituteClient(); Assert.Null(await adapter.ReadServiceLevelAsync()); await adapter.ConnectAsync(new Dictionary()); client.IsConnected.Returns(false); Assert.Null(await adapter.ReadServiceLevelAsync()); await client.DidNotReceive().ReadValueAsync(Arg.Any(), Arg.Any()); } // ── Real OPC UA servers ── /// The real adapter reads the real server's Server_ServiceLevel as it changes. [Fact] public async Task RealServer_ReadServiceLevel_TracksTheServersValue() { await using var server = await SecureOpcUaTestServer.StartAsync(); await using var adapter = new OpcUaDataConnection( new RealOpcUaClientFactory(ClientOptions()), NullLogger.Instance); await adapter.ConnectAsync(Flat(server.Url)); foreach (var level in new byte[] { 250, 240, 90, 50, 1 }) { server.SetServiceLevel(level); Assert.Equal(level, await adapter.ReadServiceLevelAsync()); } } /// /// End to end on two real servers through the production factory: the connection starts on a /// primary that is the Warm standby (90) while the backup is Active (250), and ends up on the /// backup. Then the mirror case on a fresh connection — today's 250/240 pair — stays on the /// primary. /// [Fact] public async Task RealServerPair_StandbyPrimary_FailsOverToTheActiveBackup_ActivePair_StaysPut() { await using var a = await SecureOpcUaTestServer.StartAsync(); await using var b = await SecureOpcUaTestServer.StartAsync(); var factory = new DataConnectionFactory(NullLoggerFactory.Instance, Options.Create(ClientOptions())); var options = new DataConnectionOptions { ReconnectInterval = TimeSpan.FromMilliseconds(200), ServiceLevelPollInterval = TimeSpan.FromMilliseconds(100), ServiceLevelProbeTimeout = TimeSpan.FromSeconds(20), }; var sl = new OpcUaServiceLevelFailoverConfig { DegradedWindowSeconds = 1, MinDwellSeconds = 0 }; async Task Endpoint(IActorRef actor) => (await actor.Ask(new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(5))) .ActiveEndpoint; // Warm: A is the standby, B the Active. a.SetServiceLevel(90); b.SetServiceLevel(250); var health = Substitute.For(); var primary = Flat(a.Url, sl); var warm = Sys.ActorOf(Props.Create(() => new DataConnectionActor( "warm", factory.Create("OpcUa", primary), options, health, factory, "OpcUa", primary, Flat(b.Url), 3, null))); await AwaitConditionAsync(async () => await Endpoint(warm) == "Backup", TimeSpan.FromSeconds(30)); AwaitCondition(() => health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) && c.GetArguments() is ["warm", "Connected to backup"]), TimeSpan.FromSeconds(30)); Sys.Stop(warm); // Today: A is the Secondary (240), B the Primary (250) — both Active band. a.SetServiceLevel(240); var today = Sys.ActorOf(Props.Create(() => new DataConnectionActor( "today", factory.Create("OpcUa", primary), options, health, factory, "OpcUa", primary, Flat(b.Url), 3, null))); AwaitCondition(() => health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) && c.GetArguments() is ["today", "Connected to primary"]), TimeSpan.FromSeconds(30)); await Task.Delay(TimeSpan.FromSeconds(2.5)); // > window, ~25 polls Assert.Equal("Primary", await Endpoint(today)); Assert.DoesNotContain(health.ReceivedCalls(), c => c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) && c.GetArguments() is ["today", "Connected to backup"]); } }