using System.Collections.Concurrent; using Akka.Actor; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; /// /// Runtime guard for the Phase 6.1 resilience wiring (arch-review #10 / RESILIENCE-DISPATCH-GAP): /// proves the routes each driver-capability call through the /// injected at RUNTIME, with the correct /// and resolved host key — not just that the calls compile wrapped /// (the OTOPCUA0001 analyzer is the compile-time guard; this is the behavioural one). /// A recording invoker stands in for the real CapabilityInvoker so the test stays in the /// Polly-free Runtime layer. /// public sealed class DriverInstanceActorResilienceWiringTests : RuntimeActorTestBase { /// /// A bulk subscribe issued on connect (via ResubscribeDesired) routes through the invoker /// with keyed on the driver-instance id (single-host /// fallback for a set that may span hosts). The driver's SubscribeAsync only runs from /// inside the invoker's call-site, so SubscribeCount >= 1 already proves the route. /// [Fact] public async Task Subscribe_routes_through_the_capability_invoker() { var invoker = new RecordingInvoker(); var driver = new SubscribableStubDriver(); var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, invoker: invoker)); actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100))); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); AwaitCondition(() => driver.SubscribeCount >= 1, TimeSpan.FromSeconds(3)); // Barrier: round-trip a Connected-only Ask so the async subscribe handler has fully recorded. await actor.Ask( new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), TimeSpan.FromSeconds(3)); invoker.Calls.ShouldContain(c => c.Capability == DriverCapability.Subscribe && c.Host == driver.DriverInstanceId); } /// /// A write routes through the invoker's ExecuteWriteAsync path (recorded as /// ) with the host resolved via /// — proving per-host breaker keying is wired, not the /// single-host fallback. A write that arrives before Connected fast-fails WITHOUT /// touching the invoker, so the loop drives to Connected and the recorded call is the real one. /// [Fact] public async Task Write_routes_through_the_invoker_with_resolved_host() { var invoker = new RecordingInvoker(); var driver = new HostResolvingWritableDriver(); // ResolveHost("tag-1") => "plc-7" var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, invoker: invoker)); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); // Retry until the write succeeds (i.e. the actor reached Connected). Connecting-state writes // fast-fail on a synchronous path that never reaches the invoker, so they don't pollute Calls. DriverInstanceActor.WriteAttributeResult? reply = null; for (var i = 0; i < 60 && reply is not { Success: true }; i++) { reply = await actor.Ask( new DriverInstanceActor.WriteAttribute("tag-1", 42), TimeSpan.FromSeconds(1)); if (reply is not { Success: true }) await Task.Delay(25); } reply.ShouldNotBeNull(); reply!.Success.ShouldBeTrue(); driver.Writes.Count.ShouldBeGreaterThanOrEqualTo(1); // the invoker's call-site actually ran invoker.Calls.ShouldContain(c => c.Capability == DriverCapability.Write && c.Host == "plc-7"); // 03/S12: the dispatch site must default writes to NON-idempotent so the invoker's no-retry arm is // authoritative — a command-shaped write (pulse, counter, supervisory) must never replay. invoker.WriteCalls.ShouldContain(c => c.Host == "plc-7" && c.IsIdempotent == false); } /// Records every capability + resolved-host key routed through it, then delegates to the /// real call-site so the driver still runs. Stands in for CapabilityInvoker without pulling /// Polly / Core into the Runtime test. private sealed class RecordingInvoker : IDriverCapabilityInvoker { public ConcurrentQueue<(DriverCapability Capability, string Host)> Calls { get; } = new(); public ConcurrentQueue<(string Host, bool IsIdempotent)> WriteCalls { get; } = new(); public ValueTask ExecuteAsync( DriverCapability capability, string hostName, Func> callSite, CancellationToken cancellationToken) { Calls.Enqueue((capability, hostName)); return callSite(cancellationToken); } public ValueTask ExecuteAsync( DriverCapability capability, string hostName, Func callSite, CancellationToken cancellationToken) { Calls.Enqueue((capability, hostName)); return callSite(cancellationToken); } public ValueTask ExecuteWriteAsync( string hostName, bool isIdempotent, Func> callSite, CancellationToken cancellationToken) { Calls.Enqueue((DriverCapability.Write, hostName)); WriteCalls.Enqueue((hostName, isIdempotent)); return callSite(cancellationToken); } } /// A writable driver that also resolves a per-tag host, so the write path exercises /// 's ResolveHost (via ). private sealed class HostResolvingWritableDriver : IDriver, IWritable, IPerCallHostResolver { public List Writes { get; } = new(); public string DriverInstanceId => "host-resolving-driver"; public string DriverType => "Stub"; public string ResolveHost(string fullReference) => "plc-7"; public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); public long GetMemoryFootprint() => 0; public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; public Task> WriteAsync( IReadOnlyList writes, CancellationToken cancellationToken) { Writes.AddRange(writes); IReadOnlyList results = writes.Select(_ => new WriteResult(0u)).ToList(); return Task.FromResult(results); } } }