Joseph Doherty
233 lines
10 KiB
C#
233 lines
10 KiB
C#
using System.Collections.Concurrent;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
|
|
|
|
/// <summary>
|
|
/// Unit tests for the STAB-2 fix: native ADS notifications are re-registered onto a fresh
|
|
/// client after a reconnect, instead of being silently orphaned. There is <b>no TwinCAT docker
|
|
/// fixture</b> (the 13 integration tests need a real TC3 XAR target), so these fake-client tests
|
|
/// are the <i>authoritative</i> automated coverage for the reconnect path.
|
|
/// </summary>
|
|
[Trait("Category", "Unit")]
|
|
public sealed class TwinCATReconnectReplayTests
|
|
{
|
|
private const string Host = "ads://5.23.91.23.1.1:851";
|
|
|
|
private static (TwinCATDriver drv, FakeTwinCATClientFactory factory) NewNativeDriver(
|
|
bool probe = false, TimeSpan? probeInterval = null, params TwinCATTagDefinition[] tags)
|
|
{
|
|
var factory = new FakeTwinCATClientFactory();
|
|
var drv = new TwinCATDriver(new TwinCATDriverOptions
|
|
{
|
|
Devices = [new TwinCATDeviceOptions(Host)],
|
|
Tags = tags,
|
|
Probe = new TwinCATProbeOptions
|
|
{
|
|
Enabled = probe,
|
|
Interval = probeInterval ?? TimeSpan.FromMilliseconds(30),
|
|
Timeout = TimeSpan.FromMilliseconds(100),
|
|
},
|
|
UseNativeNotifications = true,
|
|
}, "drv-1", factory);
|
|
return (drv, factory);
|
|
}
|
|
|
|
/// <summary>
|
|
/// After a client drop, the next connection (driven here through a read) re-registers the
|
|
/// native notification with the same symbol / type / interval, a push on the fresh client
|
|
/// reaches OnDataChange, and the old handle is disposed.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Reconnect_replays_native_notifications_onto_the_fresh_client()
|
|
{
|
|
var (drv, factory) = NewNativeDriver(
|
|
tags: new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
var events = new ConcurrentQueue<DataChangeEventArgs>();
|
|
drv.OnDataChange += (_, e) => events.Enqueue(e);
|
|
|
|
var interval = TimeSpan.FromMilliseconds(250);
|
|
_ = await drv.SubscribeAsync(["Speed"], interval, TestContext.Current.CancellationToken);
|
|
|
|
var client1 = factory.Clients[0];
|
|
client1.Notifications.Count.ShouldBe(1);
|
|
var oldHandle = client1.Notifications[0];
|
|
|
|
// Simulate a wire-level drop, then trigger a reconnect via a read.
|
|
client1.SimulateWireDrop();
|
|
_ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken);
|
|
|
|
// A brand-new client was built and the notification replayed onto it — same intent.
|
|
factory.Clients.Count.ShouldBe(2);
|
|
var client2 = factory.Clients[1];
|
|
client2.Notifications.Count.ShouldBe(1);
|
|
var replayed = client2.Notifications[0];
|
|
replayed.SymbolPath.ShouldBe("MAIN.Speed");
|
|
replayed.Type.ShouldBe(TwinCATDataType.DInt);
|
|
|
|
// The old handle was disposed; a push on the FRESH client reaches OnDataChange.
|
|
oldHandle.Disposed.ShouldBeTrue();
|
|
client2.FireNotification("MAIN.Speed", 7777);
|
|
events.Count.ShouldBe(1);
|
|
events.Last().Snapshot.Value.ShouldBe(7777);
|
|
events.Last().FullReference.ShouldBe("Speed");
|
|
}
|
|
|
|
/// <summary>Every subscribed tag on the device is replayed, not just the first.</summary>
|
|
[Fact]
|
|
public async Task Reconnect_replays_all_registered_tags()
|
|
{
|
|
var (drv, factory) = NewNativeDriver(
|
|
tags:
|
|
[
|
|
new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt),
|
|
new TwinCATTagDefinition("B", Host, "MAIN.B", TwinCATDataType.Real),
|
|
]);
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
_ = await drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
|
|
|
|
factory.Clients[0].SimulateWireDrop();
|
|
_ = await drv.ReadAsync(["A"], TestContext.Current.CancellationToken);
|
|
|
|
factory.Clients.Count.ShouldBe(2);
|
|
factory.Clients[1].Notifications.Select(n => n.SymbolPath)
|
|
.ShouldBe(["MAIN.A", "MAIN.B"], ignoreOrder: true);
|
|
}
|
|
|
|
/// <summary>
|
|
/// After a reconnect + replay, unsubscribing disposes the CURRENT (replayed) handle and the
|
|
/// registration no longer replays on a subsequent reconnect.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Unsubscribe_after_reconnect_disposes_the_current_handle_and_stops_replaying()
|
|
{
|
|
var (drv, factory) = NewNativeDriver(
|
|
tags: new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt));
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
|
|
|
|
// First reconnect → replayed onto client2.
|
|
factory.Clients[0].SimulateWireDrop();
|
|
_ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
|
|
var client2 = factory.Clients[1];
|
|
client2.Notifications.Count.ShouldBe(1);
|
|
|
|
// Unsubscribe disposes the current (client2) handle.
|
|
await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
|
|
client2.Notifications.ShouldBeEmpty();
|
|
|
|
// A second reconnect must NOT resurrect the unsubscribed notification.
|
|
client2.SimulateWireDrop();
|
|
_ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
|
|
factory.Clients.Count.ShouldBe(3);
|
|
factory.Clients[2].Notifications.ShouldBeEmpty();
|
|
}
|
|
|
|
/// <summary>
|
|
/// A wire-level probe failure on a locally-"connected" client forces a recycle so the next
|
|
/// tick rebuilds the client — the compounding half of STAB-2 (the IsConnected fast-path
|
|
/// keys on port state, not wire liveness).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Probe_failure_recycles_the_client_and_rebuilds()
|
|
{
|
|
// First client's wire probe fails; the recycle nulls it and the next tick builds a second
|
|
// client whose probe succeeds.
|
|
var buildCount = 0;
|
|
var factory = new FakeTwinCATClientFactory
|
|
{
|
|
Customise = () =>
|
|
{
|
|
buildCount++;
|
|
// Client #1 reports a failed wire probe; #2+ succeed.
|
|
return new FakeTwinCATClient { ProbeResult = buildCount > 1 };
|
|
},
|
|
};
|
|
var drv = new TwinCATDriver(new TwinCATDriverOptions
|
|
{
|
|
Devices = [new TwinCATDeviceOptions(Host)],
|
|
Tags = [new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt)],
|
|
Probe = new TwinCATProbeOptions
|
|
{
|
|
Enabled = true,
|
|
Interval = TimeSpan.FromMilliseconds(30),
|
|
Timeout = TimeSpan.FromMilliseconds(100),
|
|
},
|
|
UseNativeNotifications = true,
|
|
}, "drv-probe", factory);
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
// The probe loop should recycle the wire-dead client and rebuild a healthy one.
|
|
await WaitForAsync(() => factory.Clients.Count >= 2, TimeSpan.FromSeconds(3));
|
|
factory.Clients.Count.ShouldBeGreaterThanOrEqualTo(2);
|
|
factory.Clients[0].DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
|
|
|
|
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
|
|
}
|
|
|
|
// ---- STAB-16 / STAB-17 / STAB-12 replay hardening ----
|
|
|
|
/// <summary>
|
|
/// STAB-16: a replay failure must not be silent — the registration is marked dead, a Bad
|
|
/// snapshot is pushed to subscribers (instead of the frozen-Good-stale value), and health
|
|
/// degrades preserving the last successful read.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task ReplayFailure_EmitsBadQuality_AndDegradesHealth()
|
|
{
|
|
var build = 0;
|
|
var factory = new FakeTwinCATClientFactory
|
|
{
|
|
// client1 normal; client2 (the reconnect target) fails AddNotification (replay) + reads.
|
|
Customise = () => new FakeTwinCATClient
|
|
{
|
|
ThrowOnAddNotification = ++build >= 2,
|
|
ThrowOnRead = build >= 2,
|
|
Exception = new InvalidOperationException("ADS notification quota exceeded"),
|
|
},
|
|
};
|
|
var drv = new TwinCATDriver(new TwinCATDriverOptions
|
|
{
|
|
Devices = [new TwinCATDeviceOptions(Host)],
|
|
Tags = [new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt)],
|
|
Probe = new TwinCATProbeOptions { Enabled = false },
|
|
UseNativeNotifications = true,
|
|
}, "drv-1", factory);
|
|
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
|
|
|
var events = new ConcurrentQueue<DataChangeEventArgs>();
|
|
drv.OnDataChange += (_, e) => events.Enqueue(e);
|
|
|
|
_ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(250), TestContext.Current.CancellationToken);
|
|
// Establish a last-successful-read on the healthy client1.
|
|
_ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken);
|
|
var lastGoodRead = drv.GetHealth().LastSuccessfulRead;
|
|
lastGoodRead.ShouldNotBeNull();
|
|
|
|
// Drop the wire; the next read reconnects onto client2, whose replay AddNotification fails.
|
|
factory.Clients[0].SimulateWireDrop();
|
|
_ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken);
|
|
|
|
// STAB-16: subscribers see the tag go Bad (not frozen Good), and health degraded while
|
|
// preserving the last successful read.
|
|
var bad = events.SingleOrDefault(e => e.FullReference == "Speed"
|
|
&& e.Snapshot.StatusCode == TwinCATStatusMapper.BadCommunicationError);
|
|
bad.ShouldNotBeNull("a failed replay must publish a Bad snapshot for the affected reference");
|
|
var health = drv.GetHealth();
|
|
health.State.ShouldBe(DriverState.Degraded);
|
|
health.LastSuccessfulRead.ShouldBe(lastGoodRead);
|
|
}
|
|
|
|
private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
|
|
{
|
|
var deadline = DateTime.UtcNow + timeout;
|
|
while (!condition() && DateTime.UtcNow < deadline)
|
|
await Task.Delay(20);
|
|
}
|
|
}
|