test(s7): failing guards for the poll-fork retirement (PERF-3, STAB-6)

2026-07-13 12:05:08 -04:00
parent b494d09d9b
commit 0d879363f8
2 changed files with 131 additions and 1 deletions
@@ -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"
      ]
@@ -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;
+
+/// <summary>
+///     Guards for retiring the bespoke S7 poll fork onto the shared <c>PollGroupEngine</c>
+///     (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.
+/// </summary>
+[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<object?> ReadAsync(string address, CancellationToken ct)
+        {
+            if (owner.ReadThrows) throw new PlcException(ErrorCode.ConnectionError, "poll failure");
+            return Task.FromResult<object?>((ushort)42);
+        }
+
+        public Task<byte[]> 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)],
+    };
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<DataChangeEventArgs>();
+        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
+    }
+
+    /// <summary>
+    ///     STAB-6-S7 regression: rapidly subscribing + unsubscribing must not fault a poll loop by
+    ///     disposing its CTS out from under an in-flight <c>Task.Delay</c>. Post-migration the engine
+    ///     drains each loop before disposing its CTS. Asserts the whole churn completes cleanly and
+    ///     the driver still serves reads afterward.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    [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);
+    }
+}