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). [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); SeedEquipmentDeployment(db, "eq-1"); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); 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"); 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 void SeedEquipmentDeployment(IDbContextFactory dbFactory, params string[] equipmentIds) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new { Equipment = equipmentIds.Select(id => new { EquipmentId = id, MachineCode = id.ToUpperInvariant(), UnsLineId = "line-1", Name = id, }).ToArray(), DriverInstances = Array.Empty(), ScriptedAlarms = Array.Empty(), }); using var ctx = dbFactory.CreateDbContext(); ctx.Deployments.Add(new Deployment { DeploymentId = Guid.NewGuid(), RevisionHash = new string('a', 64), Status = DeploymentStatus.Sealed, CreatedBy = "test", SealedAtUtc = DateTime.UtcNow, ArtifactBlob = artifact, }); ctx.SaveChanges(); } /// 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(); } }