diff --git a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json index 12fda5e7..e802ac57 100644 --- a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json +++ b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json @@ -95,7 +95,7 @@ { "id": "B3.5", "subject": "B3: failing S7PollEngineMigrationTests (array-cadence PERF-3, rapid sub/unsub STAB-6, sustained-failure health+backoff)", - "status": "pending", + "status": "completed", "blockedBy": [ "B3.2" ] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7PollEngineMigrationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7PollEngineMigrationTests.cs new file mode 100644 index 00000000..f3e6d8fc --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7PollEngineMigrationTests.cs @@ -0,0 +1,130 @@ +using S7.Net; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using S7NetDataType = global::S7.Net.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.S7.Tests; + +/// +/// Guards for retiring the bespoke S7 poll fork onto the shared PollGroupEngine +/// (CONV-1). The engine brings two fixes the fork lacked — structural array diffing (PERF-3) +/// and drain-before-CTS-dispose on unsubscribe (STAB-6-S7) — and keeps failure→health parity. +/// +[Trait("Category", "Unit")] +public sealed class S7PollEngineMigrationTests +{ + // A fake that opens cleanly and serves a CONSTANT 6-byte block for array reads (three + // big-endian UInt16 words) — every ReadArrayAsync decodes a fresh ushort[3] with identical + // contents, so a reference-equality diff (the fork) refires every tick while a structural + // diff (the engine) fires only once. + private sealed class ConstantArrayFactory : IS7PlcFactory + { + public bool ReadThrows { get; set; } + public IS7Plc Create(S7CpuType cpuType, string host, int port, short rack, short slot, TimeSpan timeout) + => new ConstantArrayPlc(this); + } + + private sealed class ConstantArrayPlc(ConstantArrayFactory owner) : IS7Plc + { + public bool IsConnected { get; private set; } + public Task OpenAsync(CancellationToken ct) { IsConnected = true; return Task.CompletedTask; } + public void Close() => IsConnected = false; + + public Task ReadAsync(string address, CancellationToken ct) + { + if (owner.ReadThrows) throw new PlcException(ErrorCode.ConnectionError, "poll failure"); + return Task.FromResult((ushort)42); + } + + public Task ReadBytesAsync(S7NetDataType area, int db, int startByteAdr, int count, CancellationToken ct) + { + if (owner.ReadThrows) throw new PlcException(ErrorCode.ConnectionError, "poll failure"); + // Fresh array instance each call, identical contents: BE words 1, 2, 3. + var block = new byte[count]; + for (var i = 0; i + 1 < block.Length; i += 2) + block[i + 1] = (byte)(i / 2 + 1); + return Task.FromResult(block); + } + + public Task WriteAsync(string address, object value, CancellationToken ct) => Task.CompletedTask; + public Task WriteBytesAsync(S7NetDataType area, int db, int startByteAdr, byte[] value, CancellationToken ct) => Task.CompletedTask; + public Task ReadStatusAsync(CancellationToken ct) => Task.CompletedTask; + public void Dispose() => IsConnected = false; + } + + private static S7DriverOptions ArrayOptions() => new() + { + Host = "192.0.2.1", + Timeout = TimeSpan.FromMilliseconds(250), + Probe = new S7ProbeOptions { Enabled = false }, + Tags = [new S7TagDefinition("Arr", "DB1.DBW0", S7DataType.UInt16, Writable: false, ArrayCount: 3)], + }; + + /// + /// PERF-3 regression: a subscribed array tag whose contents are identical across polls + /// (fresh instance each read) must fire only the initial change, not on every tick. The + /// fork's reference-equality diff refires every poll. + /// + [Fact] + public async Task ArrayTag_UnchangedBetweenPolls_DoesNotRefire() + { + var factory = new ConstantArrayFactory(); + using var drv = new S7Driver(ArrayOptions(), "s7-arr-perf3", factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + var events = new System.Collections.Concurrent.ConcurrentQueue(); + drv.OnDataChange += (_, e) => events.Enqueue(e); + + var h = await drv.SubscribeAsync(["Arr"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + await Task.Delay(500, TestContext.Current.CancellationToken); // several poll cycles + await drv.UnsubscribeAsync(h, TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); // only the initial-data push + } + + /// + /// STAB-6-S7 regression: rapidly subscribing + unsubscribing must not fault a poll loop by + /// disposing its CTS out from under an in-flight Task.Delay. Post-migration the engine + /// drains each loop before disposing its CTS. Asserts the whole churn completes cleanly and + /// the driver still serves reads afterward. + /// + [Fact] + public async Task RapidSubscribeUnsubscribe_NoObjectDisposedException() + { + var factory = new ConstantArrayFactory(); + using var drv = new S7Driver(ArrayOptions(), "s7-rapid", factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + for (var i = 0; i < 50; i++) + { + var h = await drv.SubscribeAsync(["Arr"], TimeSpan.FromMilliseconds(20), TestContext.Current.CancellationToken); + await drv.UnsubscribeAsync(h, TestContext.Current.CancellationToken); + } + + // Driver still works after the churn — no wedged state. + var read = await drv.ReadAsync(["Arr"], TestContext.Current.CancellationToken); + read[0].StatusCode.ShouldBe(0u); + } + + /// + /// Failure→health parity: a subscription polling a persistently-failing device degrades the + /// driver health (the fork's HandlePollFailure behaviour, now delivered by the engine's + /// onError→HandlePollError wiring + the read path's own status mapping). + /// + [Fact] + public async Task SustainedPollFailure_DegradesHealth() + { + var factory = new ConstantArrayFactory { ReadThrows = true }; + using var drv = new S7Driver(ArrayOptions(), "s7-pollfail", factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + await drv.SubscribeAsync(["Arr"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken); + + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(3); + while (drv.GetHealth().State == DriverState.Healthy && DateTime.UtcNow < deadline) + await Task.Delay(25, TestContext.Current.CancellationToken); + + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + } +}