From 376233b92c2abb0452d4ace9191987f73b96d5ee Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 12:01:23 -0400 Subject: [PATCH] test(opcua): over-the-wire subscription-survival on pure-add deploy (R2-07 T5) --- ...2-07-surgical-pure-adds-plan.md.tasks.json | 5 +- .../SubscriptionSurvivalTests.cs | 204 ++++++++++++++++++ 2 files changed, 207 insertions(+), 2 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs diff --git a/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json b/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json index d743e7e6..356b451c 100644 --- a/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json +++ b/archreview/plans/R2-07-surgical-pure-adds-plan.md.tasks.json @@ -45,10 +45,11 @@ { "id": "T5", "subject": "Phase 1: over-the-wire SubscriptionSurvivalTests \u2014 subscribe, pure-add, item survives + new node browsable", - "status": "pending", + "status": "deferred-live", "blockedBy": [ "T4b" - ] + ], + "note": "authored + compiles (dotnet build green); NOT run \u2014 *.IntegrationTests suite leaks ~16GB/run per the memory constraint. MUST run in the serial heavy integration pass (single-test, boots one in-process server + client)." }, { "id": "T6", diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs new file mode 100644 index 00000000..62ee88d7 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs @@ -0,0 +1,204 @@ +using System.Net; +using System.Net.Sockets; +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// R2-07 03/P1 — the HEADLINE proof: an over-the-wire subscription on an existing tag SURVIVES a +/// pure-add deploy (one new tag), instead of being killed by the full-rebuild teardown that pre-R2-07 +/// every topology change forced. Boots the real , materialises tag A, +/// opens a real client subscription + monitored item on A, then drives a PURE-ADD apply adding tag B +/// (classifier ⇒ PureAdd ⇒ no rebuild ⇒ the idempotent Materialise passes create B and leave A's +/// NodeState untouched), and asserts: +/// +/// A's monitored item still delivers a subsequent value write (the item is alive — no +/// BadNodeIdUnknown, no dead subscription); +/// B is browsable + readable (the add landed). +/// +/// Written against the master (pre-T2) behaviour this FAILS (the rebuild recreates A as a new NodeState +/// and the monitored item goes dead); it passes once the classifier routing (T2) + the publish-actor +/// announce (T4b) land. +/// Run note: this is a heavy in-process server+client integration test — run it in the +/// serial integration pass (it is NOT part of the lightweight unit sweep). The +/// namespace URI is ; the client resolves the +/// folder-scoped variable NodeIds against it. +/// +public sealed class SubscriptionSurvivalTests +{ + private const string ServerUri = "urn:OtOpcUa.SubscriptionSurvival"; + + /// A subscription on tag A survives a pure-add deploy adding tag B; B becomes readable. + [Fact] + public async Task Subscription_on_existing_tag_survives_pure_add_of_another_tag() + { + var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-{Guid.NewGuid():N}"); + var port = AllocateFreePort(); + var ct = TestContext.Current.CancellationToken; + + try + { + var options = new OpcUaApplicationHostOptions + { + ApplicationName = ServerUri, + ApplicationUri = ServerUri, + OpcUaPort = port, + PublicHostname = "127.0.0.1", + PkiStoreRoot = pkiRoot, + EnabledSecurityProfiles = new List { OpcUaSecurityProfile.None }, + AutoAcceptUntrustedClientCertificates = true, + }; + var server = new OtOpcUaSdkServer(); + await using var host = new OpcUaApplicationHost(options, NullLogger.Instance); + await host.StartAsync(server, ct); + + var nm = server.NodeManager!; + var sink = new SdkAddressSpaceSink(nm); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + + // --- Materialise tag A (equipment eq-1, tag "A") and seed it a Good value. --- + var withA = CompositionWith( + new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null)); + applier.MaterialiseHierarchy(withA); + applier.MaterialiseEquipmentTags(withA); + var nodeIdAString = Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "A"); + sink.WriteValue(nodeIdAString, 41, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow); + + // --- Open a client session + subscription + monitored item on A. --- + using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct); + var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri); + ns.ShouldBeGreaterThan((ushort)0); + var nodeIdA = new NodeId(nodeIdAString, ns); + + var received = new List(); + var gate = new object(); + var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 }; + session.AddSubscription(subscription); + await subscription.CreateAsync(ct); + + var monitored = new MonitoredItem(subscription.DefaultItem) + { + StartNodeId = nodeIdA, + AttributeId = Attributes.Value, + SamplingInterval = 50, + QueueSize = 10, + }; + monitored.Notification += (item, _) => + { + foreach (var v in item.DequeueValues()) + lock (gate) received.Add(v); + }; + subscription.AddItem(monitored); + await subscription.ApplyChangesAsync(ct); + + // Baseline: the initial value (41) is delivered. + await WaitUntilAsync(() => { lock (gate) return received.Count >= 1; }, TimeSpan.FromSeconds(5)); + + // --- Drive a PURE-ADD apply adding tag B (previous={A}, next={A,B}). --- + var withAB = CompositionWith( + new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null), + new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null)); + var plan = AddressSpacePlanner.Compute(withA, withAB); + plan.AddedEquipmentTags.Count.ShouldBe(1); + + var outcome = applier.Apply(plan); + outcome.RebuildCalled.ShouldBeFalse(); // PureAdd — no teardown + + // Materialise passes (as the publish actor runs them) then announce — over the FULL composition. + applier.MaterialiseHierarchy(withAB); + applier.MaterialiseEquipmentTags(withAB); + applier.AnnounceAddedNodes(plan); + + // --- A's monitored item must still be alive: a new write to A is delivered. --- + lock (gate) received.Clear(); + sink.WriteValue(nodeIdAString, 42, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow); + + await WaitUntilAsync(() => { lock (gate) return received.Any(v => Equals(v.Value, 42)); }, TimeSpan.FromSeconds(5)); + lock (gate) + { + received.ShouldContain(v => Equals(v.Value, 42)); + received.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown); + } + + // --- B is browsable + readable (the add landed). --- + var nodeIdB = new NodeId(Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "B"), ns); + sink.WriteValue(nodeIdB.Identifier.ToString()!, 7, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow); + var bValue = await session.ReadValueAsync(nodeIdB, ct); + bValue.StatusCode.ShouldNotBe(StatusCodes.BadNodeIdUnknown); + bValue.Value.ShouldBe(7); + } + finally + { + if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true); + } + } + + private static AddressSpaceComposition CompositionWith(params EquipmentTagPlan[] tags) => + new( + UnsAreas: Array.Empty(), + UnsLines: Array.Empty(), + EquipmentNodes: new[] { new EquipmentNode("eq-1", "Equipment 1", "") }, + DriverInstancePlans: Array.Empty(), + ScriptedAlarmPlans: Array.Empty()) + { + EquipmentTags = tags, + }; + + private static async Task OpenSessionAsync(string endpointUrl, CancellationToken ct) + { + var appConfig = new ApplicationConfiguration + { + ApplicationName = "OtOpcUa.SubscriptionSurvivalClient", + ApplicationUri = $"urn:OtOpcUa.SubscriptionSurvivalClient.{Guid.NewGuid():N}", + ApplicationType = ApplicationType.Client, + SecurityConfiguration = new SecurityConfiguration + { + ApplicationCertificate = new CertificateIdentifier(), + AutoAcceptUntrustedCertificates = true, + }, + ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 }, + }; + await appConfig.ValidateAsync(ApplicationType.Client, ct); + appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true; + + var endpoint = await CoreClientUtils.SelectEndpointAsync( + appConfig, endpointUrl, false, DefaultTelemetry.Create(_ => { }), ct); + var endpointConfiguration = EndpointConfiguration.Create(appConfig); + var configuredEndpoint = new ConfiguredEndpoint(null, endpoint, endpointConfiguration); + + return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync( + appConfig, + configuredEndpoint, + updateBeforeConnect: false, + checkDomain: false, + sessionName: "SubscriptionSurvivalTests", + sessionTimeout: 60_000, + identity: new UserIdentity(new AnonymousIdentityToken()), + preferredLocales: null, + ct: ct); + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) return; + await Task.Delay(50); + } + condition().ShouldBeTrue("condition not met within timeout"); + } + + private static int AllocateFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } +}