using System.Diagnostics.Metrics; using System.Text.Json; using Akka.Actor; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Observability; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; /// /// archreview 01/S-1: when the applier reports a degraded apply (a rebuild that threw, or per-node /// materialise failures), must surface it — increment the /// otopcua.opcua.apply.failed meter and log at Error — instead of the optimistic Info line. The /// happy path must stay Info-only (zero increments). /// public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase { /// A rebuild whose sink throws increments otopcua.opcua.apply.failed (kind=rebuild). R2-07 — a /// pure-add no longer rebuilds, so the degraded-rebuild path is exercised via a rebuild-forcing change: /// a first PureAdd deploy establishes the applied composition, then a rename (ChangedEquipment ⇒ Rebuild) /// drives the throwing SafeRebuild. [Fact] public void Rebuild_when_sink_throws_increments_apply_failed_meter() { using var recorder = new MeterRecorder("otopcua.opcua.apply.failed"); var db = NewInMemoryDbFactory(); var sink = new ThrowOnRebuildSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); var dep1 = SeedEquipmentDeployment(db, ("eq-1", "eq-1")); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); // First deploy: PureAdd — no rebuild, so the throwing sink is not hit yet. actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); Thread.Sleep(200); recorder.Total.ShouldBe(0); // Second deploy: eq-1 RENAMED ⇒ ChangedEquipment ⇒ Rebuild ⇒ the sink's RebuildAddressSpace throws. var dep2 = SeedEquipmentDeployment(db, ("eq-1", "eq-1-renamed")); actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); AwaitAssert(() => { recorder.Total.ShouldBeGreaterThanOrEqualTo(1); recorder.WithTag("kind", "rebuild").ShouldBeGreaterThanOrEqualTo(1); }, duration: TimeSpan.FromSeconds(2)); } /// A clean rebuild does NOT increment the apply-failed meter (Info-only happy path). [Fact] public void Rebuild_happy_path_does_not_increment_apply_failed_meter() { using var recorder = new MeterRecorder("otopcua.opcua.apply.failed"); var db = NewInMemoryDbFactory(); var sink = new NoopSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); SeedEquipmentDeployment(db, ("eq-1", "eq-1")); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); // Give the actor time to process, then confirm the failed meter never fired. Thread.Sleep(400); recorder.Total.ShouldBe(0); } private void AwaitAssert(Action assertion, TimeSpan duration) { var deadline = DateTime.UtcNow + duration; Exception? last = null; while (DateTime.UtcNow < deadline) { try { assertion(); return; } catch (Exception ex) { last = ex; Thread.Sleep(25); } } if (last is not null) throw last; } private static Guid SeedEquipmentDeployment(IDbContextFactory dbFactory, params (string Id, string Name)[] equipment) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new { Equipment = equipment.Select(e => new { EquipmentId = e.Id, MachineCode = e.Id.ToUpperInvariant(), UnsLineId = "line-1", Name = e.Name, }).ToArray(), DriverInstances = Array.Empty(), ScriptedAlarms = Array.Empty(), }); var id = Guid.NewGuid(); using var ctx = dbFactory.CreateDbContext(); ctx.Deployments.Add(new Deployment { DeploymentId = id, RevisionHash = new string('a', 64), Status = DeploymentStatus.Sealed, CreatedBy = "test", SealedAtUtc = DateTime.UtcNow, ArtifactBlob = artifact, }); ctx.SaveChanges(); return id; } /// A sink whose RebuildAddressSpace throws (drives the applier's SafeRebuild catch → RebuildFailed). private sealed class ThrowOnRebuildSink : IOpcUaAddressSpaceSink { public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { } public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { } public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { } public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { } public void RebuildAddressSpace() => throw new InvalidOperationException("simulated rebuild fault"); public void RaiseNodesAddedModelChange(string affectedNodeId) { } } /// A no-op sink — a clean apply (no failures). private sealed class NoopSink : IOpcUaAddressSpaceSink { public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { } public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc) { } public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, bool isNative = false) { } public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName) { } public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { } public void RebuildAddressSpace() { } public void RaiseNodesAddedModelChange(string affectedNodeId) { } } /// Listens to a single instrument by name on the central meter and tallies value + tags. private sealed class MeterRecorder : IDisposable { private readonly string _name; private readonly MeterListener _listener; private long _total; private readonly List[]> _tagSets = new(); private readonly object _gate = new(); public MeterRecorder(string instrumentName) { _name = instrumentName; _listener = new MeterListener { InstrumentPublished = (instrument, listener) => { if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name) listener.EnableMeasurementEvents(instrument); } }; _listener.SetMeasurementEventCallback((_, value, tags, _) => { lock (_gate) { _total += value; _tagSets.Add(tags.ToArray()); } }); _listener.Start(); } public long Total { get { lock (_gate) return _total; } } public int WithTag(string key, string value) { lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value))); } public void Dispose() => _listener.Dispose(); } }