Joseph Doherty
b8202e1147
v2-ci / unit-tests (push) Has been cancelled
Awaited deadline oracles get the runner token; sync-over-async tests made async; nullable child models asserted via ShouldNotBeNull() return values; obsolete SDK ctors replaced with the telemetry overloads; dead locals/fixture and the redundant DI PackageReference removed. No suppressions. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
97 lines
4.2 KiB
C#
97 lines
4.2 KiB
C#
using System.Runtime.CompilerServices;
|
|
using System.Threading.Channels;
|
|
using Google.Protobuf.WellKnownTypes;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
|
|
|
|
/// <summary>
|
|
/// Review 2026-10-21 G-10: a dispatch-loop exit left the read loop draining into a channel
|
|
/// nobody read, with health Healthy and nothing telling the supervisor — a clean termination
|
|
/// evading the watchdog. The exit now reports a stream fault. The fault is injected through
|
|
/// the pump's handle factory, the one dispatch-side seam that runs outside the per-handler
|
|
/// isolation.
|
|
/// </summary>
|
|
public sealed class EventPumpDispatchFaultTests
|
|
{
|
|
private const int WaitMs = 10_000;
|
|
|
|
private sealed class ChannelSubscriber : IGalaxySubscriber
|
|
{
|
|
private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
|
|
|
|
public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
|
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
|
=> Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
|
|
|
|
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
|
{
|
|
yield return ev;
|
|
}
|
|
}
|
|
|
|
public ValueTask EmitAsync(int handle) => _events.Writer.WriteAsync(new MxEvent
|
|
{
|
|
Family = MxEventFamily.OnDataChange,
|
|
ItemHandle = handle,
|
|
Value = new MxValue { DoubleValue = 1 },
|
|
Quality = 192,
|
|
SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
|
|
});
|
|
}
|
|
|
|
/// <summary>A dispatch-loop exception reports a stream fault carrying that exception.</summary>
|
|
[Fact]
|
|
public async Task Dispatch_loop_exit_reports_a_stream_fault()
|
|
{
|
|
var subscriber = new ChannelSubscriber();
|
|
var registry = new SubscriptionRegistry();
|
|
registry.Register(1, [new TagBinding("Tank01.Level", 7)]);
|
|
var fault = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
await using var pump = new EventPump(
|
|
subscriber, registry,
|
|
handleFactory: _ => throw new InvalidDataException("synthetic dispatch failure"),
|
|
channelCapacity: 4, clientName: "DispatchFault",
|
|
onStreamFault: ex => fault.TrySetResult(ex));
|
|
pump.Start();
|
|
|
|
await subscriber.EmitAsync(7);
|
|
|
|
(await Task.WhenAny(fault.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(fault.Task,
|
|
"a dispatch-loop exit must be reported as a stream fault");
|
|
(await fault.Task).ShouldBeOfType<InvalidDataException>();
|
|
}
|
|
|
|
/// <summary>Positive control: the same event with a working handle factory dispatches and reports no fault.</summary>
|
|
[Fact]
|
|
public async Task Same_event_with_a_working_factory_dispatches_without_a_fault()
|
|
{
|
|
var subscriber = new ChannelSubscriber();
|
|
var registry = new SubscriptionRegistry();
|
|
registry.Register(1, [new TagBinding("Tank01.Level", 7)]);
|
|
var faulted = false;
|
|
var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
await using var pump = new EventPump(
|
|
subscriber, registry, channelCapacity: 4, clientName: "DispatchFault",
|
|
onStreamFault: _ => faulted = true);
|
|
pump.OnDataChange += (_, e) => delivered.TrySetResult(e);
|
|
pump.Start();
|
|
|
|
await subscriber.EmitAsync(7);
|
|
|
|
(await Task.WhenAny(delivered.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(delivered.Task);
|
|
(await delivered.Task).FullReference.ShouldBe("Tank01.Level");
|
|
faulted.ShouldBeFalse();
|
|
}
|
|
}
|