Joseph Doherty
778 lines
29 KiB
C#
778 lines
29 KiB
C#
using Google.Protobuf.WellKnownTypes;
|
|
using Grpc.Core;
|
|
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;
|
|
|
|
public class SiteStreamGrpcClientTests
|
|
{
|
|
[Fact]
|
|
public void ConnectedHeaderTimeout_IsImmutablePerClient()
|
|
{
|
|
var property = typeof(SiteStreamGrpcClient).GetProperty("ConnectedHeaderTimeout",
|
|
System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.Static)!;
|
|
Assert.False(property.GetMethod!.IsStatic);
|
|
Assert.Null(property.SetMethod);
|
|
}
|
|
|
|
[Fact]
|
|
public void ConvertToDomainEvent_AttributeChanged_MapsCorrectly()
|
|
{
|
|
var ts = DateTimeOffset.UtcNow;
|
|
var evt = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "corr-1",
|
|
AttributeChanged = new AttributeValueUpdate
|
|
{
|
|
InstanceUniqueName = "Site1.Pump01",
|
|
AttributePath = "Modules.IO",
|
|
AttributeName = "Temperature",
|
|
Value = "42.5",
|
|
Quality = Quality.Good,
|
|
Timestamp = Timestamp.FromDateTimeOffset(ts)
|
|
}
|
|
};
|
|
|
|
var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);
|
|
|
|
var attr = Assert.IsType<AttributeValueChanged>(result);
|
|
Assert.Equal("Site1.Pump01", attr.InstanceUniqueName);
|
|
Assert.Equal("Modules.IO", attr.AttributePath);
|
|
Assert.Equal("Temperature", attr.AttributeName);
|
|
Assert.Equal("42.5", attr.Value);
|
|
Assert.Equal("Good", attr.Quality);
|
|
Assert.Equal(ts, attr.Timestamp);
|
|
}
|
|
|
|
[Fact]
|
|
public void ConvertToDomainEvent_AlarmChanged_MapsCorrectly()
|
|
{
|
|
var ts = DateTimeOffset.UtcNow;
|
|
var raiseTime = ts.AddMinutes(-30);
|
|
var evt = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "corr-2",
|
|
AlarmChanged = new AlarmStateUpdate
|
|
{
|
|
InstanceUniqueName = "Site1.Motor01",
|
|
AlarmName = "T01.Hi",
|
|
State = AlarmStateEnum.AlarmStateActive,
|
|
Priority = 850,
|
|
Timestamp = Timestamp.FromDateTimeOffset(ts),
|
|
Kind = "NativeOpcUa",
|
|
Active = true,
|
|
Acknowledged = false,
|
|
Confirmed = false,
|
|
ShelveState = "TimedShelved",
|
|
Suppressed = true,
|
|
SourceReference = "T01.Hi",
|
|
AlarmTypeName = "AnalogLimit.Hi",
|
|
Category = "Process",
|
|
OperatorUser = "op2",
|
|
OperatorComment = "shelved",
|
|
OriginalRaiseTime = Timestamp.FromDateTimeOffset(raiseTime),
|
|
CurrentValue = "120",
|
|
LimitValue = "100"
|
|
}
|
|
};
|
|
|
|
var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);
|
|
|
|
var alarm = Assert.IsType<AlarmStateChanged>(result);
|
|
Assert.Equal("Site1.Motor01", alarm.InstanceUniqueName);
|
|
Assert.Equal("T01.Hi", alarm.AlarmName);
|
|
Assert.Equal(AlarmState.Active, alarm.State);
|
|
Assert.Equal(850, alarm.Priority);
|
|
Assert.Equal(ts, alarm.Timestamp);
|
|
|
|
// Native enrichment mapped back.
|
|
Assert.Equal(AlarmKind.NativeOpcUa, alarm.Kind);
|
|
Assert.True(alarm.Condition.Active);
|
|
Assert.False(alarm.Condition.Acknowledged);
|
|
Assert.Equal(AlarmShelveState.TimedShelved, alarm.Condition.Shelve);
|
|
Assert.True(alarm.Condition.Suppressed);
|
|
Assert.Equal(850, alarm.Condition.Severity);
|
|
Assert.Equal("T01.Hi", alarm.SourceReference);
|
|
Assert.Equal("AnalogLimit.Hi", alarm.AlarmTypeName);
|
|
Assert.Equal("Process", alarm.Category);
|
|
Assert.Equal("op2", alarm.OperatorUser);
|
|
Assert.Equal("shelved", alarm.OperatorComment);
|
|
Assert.Equal(raiseTime, alarm.OriginalRaiseTime);
|
|
Assert.Equal("120", alarm.CurrentValue);
|
|
Assert.Equal("100", alarm.LimitValue);
|
|
}
|
|
|
|
[Fact]
|
|
public void ConvertToDomainEvent_UnknownEvent_ReturnsNull()
|
|
{
|
|
var evt = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "corr-3"
|
|
// No oneof case set
|
|
};
|
|
|
|
var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);
|
|
|
|
Assert.Null(result);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(Quality.Good, "Good")]
|
|
[InlineData(Quality.Uncertain, "Uncertain")]
|
|
[InlineData(Quality.Bad, "Bad")]
|
|
[InlineData(Quality.Unspecified, "Unknown")]
|
|
public void MapQuality_AllValues(Quality input, string expected)
|
|
{
|
|
var result = SiteStreamGrpcClient.MapQuality(input);
|
|
Assert.Equal(expected, result);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(AlarmStateEnum.AlarmStateNormal, AlarmState.Normal)]
|
|
[InlineData(AlarmStateEnum.AlarmStateActive, AlarmState.Active)]
|
|
[InlineData(AlarmStateEnum.AlarmStateUnspecified, AlarmState.Normal)]
|
|
public void MapAlarmState_AllValues(AlarmStateEnum input, AlarmState expected)
|
|
{
|
|
var result = SiteStreamGrpcClient.MapAlarmState(input);
|
|
Assert.Equal(expected, result);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(AlarmLevelEnum.AlarmLevelNone, AlarmLevel.None)]
|
|
[InlineData(AlarmLevelEnum.AlarmLevelLow, AlarmLevel.Low)]
|
|
[InlineData(AlarmLevelEnum.AlarmLevelLowLow, AlarmLevel.LowLow)]
|
|
[InlineData(AlarmLevelEnum.AlarmLevelHigh, AlarmLevel.High)]
|
|
[InlineData(AlarmLevelEnum.AlarmLevelHighHigh, AlarmLevel.HighHigh)]
|
|
public void MapAlarmLevel_AllValues(AlarmLevelEnum input, AlarmLevel expected)
|
|
{
|
|
var result = SiteStreamGrpcClient.MapAlarmLevel(input);
|
|
Assert.Equal(expected, result);
|
|
}
|
|
|
|
[Fact]
|
|
public void ConvertToDomainEvent_AlarmChanged_PreservesLevel()
|
|
{
|
|
// Round-trip: a HiLo alarm emitted at HighHigh must come through with Level intact.
|
|
var evt = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "test",
|
|
AlarmChanged = new AlarmStateUpdate
|
|
{
|
|
InstanceUniqueName = "Pump1",
|
|
AlarmName = "TempAlarm",
|
|
State = AlarmStateEnum.AlarmStateActive,
|
|
Priority = 900,
|
|
Timestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
|
Level = AlarmLevelEnum.AlarmLevelHighHigh
|
|
}
|
|
};
|
|
|
|
var domain = SiteStreamGrpcClient.ConvertToDomainEvent(evt) as AlarmStateChanged;
|
|
|
|
Assert.NotNull(domain);
|
|
Assert.Equal(AlarmState.Active, domain.State);
|
|
Assert.Equal(AlarmLevel.HighHigh, domain.Level);
|
|
Assert.Equal(900, domain.Priority);
|
|
}
|
|
|
|
[Fact]
|
|
public void Unsubscribe_CancelsSubscription()
|
|
{
|
|
// We can't easily test the full Subscribe flow without a real gRPC server,
|
|
// but we can test the Unsubscribe path by registering a CTS directly.
|
|
// Use the internal AddSubscription helper for testability.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
var cts = new CancellationTokenSource();
|
|
client.AddSubscriptionForTesting("corr-test", cts);
|
|
|
|
Assert.False(cts.IsCancellationRequested);
|
|
|
|
client.Unsubscribe("corr-test");
|
|
|
|
Assert.True(cts.IsCancellationRequested);
|
|
}
|
|
|
|
[Fact]
|
|
public void Unsubscribe_NonExistent_DoesNotThrow()
|
|
{
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
client.Unsubscribe("does-not-exist"); // Should not throw
|
|
}
|
|
|
|
[Fact]
|
|
public async Task DisposeAsync_CancelsAllSubscriptions()
|
|
{
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
var cts1 = new CancellationTokenSource();
|
|
var cts2 = new CancellationTokenSource();
|
|
client.AddSubscriptionForTesting("corr-1", cts1);
|
|
client.AddSubscriptionForTesting("corr-2", cts2);
|
|
|
|
await client.DisposeAsync();
|
|
|
|
Assert.True(cts1.IsCancellationRequested);
|
|
Assert.True(cts2.IsCancellationRequested);
|
|
}
|
|
|
|
// --- Site-wide (SubscribeSite) alarm-only stream (plan #10 T3) ---
|
|
|
|
[Fact]
|
|
public void ConvertToAlarmEvent_AlarmChanged_ReturnsMappedAlarm()
|
|
{
|
|
// An alarm event on the site-wide stream is delivered to onAlarmEvent with full enrichment.
|
|
var ts = DateTimeOffset.UtcNow;
|
|
var evt = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "site-corr",
|
|
AlarmChanged = new AlarmStateUpdate
|
|
{
|
|
InstanceUniqueName = "Site1.Motor01",
|
|
AlarmName = "T01.Hi",
|
|
State = AlarmStateEnum.AlarmStateActive,
|
|
Priority = 700,
|
|
Timestamp = Timestamp.FromDateTimeOffset(ts),
|
|
Kind = "NativeOpcUa",
|
|
Active = true,
|
|
SourceReference = "T01.Hi"
|
|
}
|
|
};
|
|
|
|
var alarm = SiteStreamGrpcClient.ConvertToAlarmEvent(evt);
|
|
|
|
Assert.NotNull(alarm);
|
|
Assert.Equal("Site1.Motor01", alarm!.InstanceUniqueName);
|
|
Assert.Equal("T01.Hi", alarm.AlarmName);
|
|
Assert.Equal(AlarmState.Active, alarm.State);
|
|
Assert.Equal(AlarmKind.NativeOpcUa, alarm.Kind);
|
|
Assert.Equal("T01.Hi", alarm.SourceReference);
|
|
}
|
|
|
|
[Fact]
|
|
public void ConvertToAlarmEvent_AttributeChanged_ReturnsNull()
|
|
{
|
|
// Attribute events must never appear on the alarm-only site-wide stream; if one
|
|
// somehow arrives it is defensively filtered out rather than delivered or thrown.
|
|
var evt = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "site-corr",
|
|
AttributeChanged = new AttributeValueUpdate
|
|
{
|
|
InstanceUniqueName = "Site1.Pump01",
|
|
AttributePath = "Modules.IO",
|
|
AttributeName = "Temperature",
|
|
Value = "42.5",
|
|
Quality = Quality.Good,
|
|
Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow)
|
|
}
|
|
};
|
|
|
|
Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt));
|
|
}
|
|
|
|
[Fact]
|
|
public void ConvertToAlarmEvent_UnknownEvent_ReturnsNull()
|
|
{
|
|
var evt = new SiteStreamEvent { CorrelationId = "site-corr" };
|
|
Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SubscribeSiteAsync_OnTestOnlyClient_Throws()
|
|
{
|
|
// Guards against subscribing on a channel-less test double, mirroring SubscribeAsync.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
await Assert.ThrowsAsync<InvalidOperationException>(() =>
|
|
client.SubscribeSiteAsync("corr", _ => { }, _ => { }, () => { }, CancellationToken.None));
|
|
}
|
|
|
|
// --- WP1.1: graceful (status OK) stream completion is reported, not swallowed ---
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_ServerEndsStreamWithOk_InvokesOnCompleted_NotOnError()
|
|
{
|
|
// The site server caps every stream at GrpcMaxStreamLifetime (4h) and then ends the
|
|
// RPC with OK. That surfaces here as a read loop that simply runs out of events.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource();
|
|
var events = new List<SiteStreamEvent>();
|
|
Exception? error = null;
|
|
var completed = 0;
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-ok",
|
|
cts,
|
|
() => FakeCall(new StubStreamReader(new SiteStreamEvent { CorrelationId = "corr-ok" })),
|
|
events.Add,
|
|
ex => error = ex,
|
|
() => completed++);
|
|
|
|
Assert.Single(events);
|
|
Assert.Null(error);
|
|
Assert.Equal(1, completed);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_StreamFaults_InvokesOnError_NotOnCompleted()
|
|
{
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource();
|
|
Exception? error = null;
|
|
var completed = 0;
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-fault",
|
|
cts,
|
|
() => FakeCall(new StubStreamReader(
|
|
new RpcException(new Status(StatusCode.Unavailable, "site gone")))),
|
|
_ => { },
|
|
ex => error = ex,
|
|
() => completed++);
|
|
|
|
Assert.IsType<RpcException>(error);
|
|
Assert.Equal(0, completed);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_OwnCancellation_InvokesNeitherCallback()
|
|
{
|
|
// Our own Unsubscribe/reconnect is a teardown, not a fault and not a graceful end:
|
|
// the caller has already moved on to a newer stream.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource();
|
|
await cts.CancelAsync();
|
|
Exception? error = null;
|
|
var completed = 0;
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-cancel",
|
|
cts,
|
|
() => FakeCall(new StubStreamReader()),
|
|
_ => { },
|
|
ex => error = ex,
|
|
() => completed++);
|
|
|
|
Assert.Null(error);
|
|
Assert.Equal(0, completed);
|
|
}
|
|
|
|
// --- Foreign vs. own Cancelled (review F2) ---
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_ForeignCancelled_InvokesOnError()
|
|
{
|
|
// A Cancelled status we did NOT ask for — the PEER cancelled, or the channel was
|
|
// disposed underneath us. It used to be swallowed by an unguarded
|
|
// `when (ex.StatusCode == StatusCode.Cancelled)` filter, firing none of
|
|
// onError/onCompleted/onConnected: the consuming aggregator kept _streamDown=false,
|
|
// so IsLive stayed true forever and the reconcile tick's reopen guard never fired.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource(); // deliberately NOT cancelled
|
|
Exception? error = null;
|
|
var completed = 0;
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-foreign-cancel",
|
|
cts,
|
|
() => FakeCall(new StubStreamReader(
|
|
new RpcException(new Status(StatusCode.Cancelled, "cancelled by peer")))),
|
|
_ => { },
|
|
ex => error = ex,
|
|
() => completed++);
|
|
|
|
var rpc = Assert.IsType<RpcException>(error);
|
|
Assert.Equal(StatusCode.Cancelled, rpc.StatusCode);
|
|
Assert.Equal(0, completed);
|
|
Assert.False(cts.IsCancellationRequested);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_OwnCancelledRpcException_InvokesNeitherCallback()
|
|
{
|
|
// The other side of the same filter: OUR cancellation surfacing as
|
|
// RpcException(Cancelled) is a teardown, so neither callback fires.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource();
|
|
await cts.CancelAsync();
|
|
Exception? error = null;
|
|
var completed = 0;
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-own-cancel",
|
|
cts,
|
|
() => throw new RpcException(new Status(StatusCode.Cancelled, "we cancelled")),
|
|
_ => { },
|
|
ex => error = ex,
|
|
() => completed++);
|
|
|
|
Assert.Null(error);
|
|
Assert.Equal(0, completed);
|
|
}
|
|
|
|
// --- onConnected is only ever raised on real proof of a live peer (review F3) ---
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_HeadersArrive_InvokesOnConnectedOnce()
|
|
{
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource();
|
|
var connected = 0;
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-headers",
|
|
cts,
|
|
() => FakeCall(
|
|
new StubStreamReader(
|
|
new SiteStreamEvent { CorrelationId = "corr-headers" },
|
|
new SiteStreamEvent { CorrelationId = "corr-headers" }),
|
|
Task.FromResult(new Metadata())),
|
|
_ => { },
|
|
_ => { },
|
|
() => { },
|
|
() => connected++);
|
|
|
|
// Once — the two events must not re-raise it.
|
|
Assert.Equal(1, connected);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_HeaderTimeoutOnADeadSite_NeverReportsConnected()
|
|
{
|
|
// The bug: a header TIMEOUT used to be reported as connected. An unreachable/wedged
|
|
// site produces exactly that shape, so the aggregator cleared _streamDown, consumed
|
|
// its pending re-seed and fanned a full snapshot out at a site that never answered.
|
|
var client = SiteStreamGrpcClient.CreateForTesting(new CommunicationOptions { ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50) });
|
|
var cts = new CancellationTokenSource();
|
|
var connected = 0;
|
|
var neverArrives = new TaskCompletionSource<Metadata>();
|
|
|
|
// No headers AND no events: the site is dead. The stream ends (status OK) with
|
|
// no connected signal ever raised.
|
|
await client.ConsumeStreamAsync(
|
|
"corr-dead",
|
|
cts,
|
|
() => FakeCall(new StubStreamReader(), neverArrives.Task),
|
|
_ => { },
|
|
_ => { },
|
|
() => { },
|
|
() => connected++);
|
|
|
|
Assert.Equal(0, connected);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_HeaderTimeoutThenEvent_ReportsConnectedOnceFromTheEvent()
|
|
{
|
|
// The case the timeout was originally added for — a peer that defers its headers
|
|
// until the first message — is now covered by the event itself, which is real proof
|
|
// of a live peer. Connected must precede the event's delivery and fire only once.
|
|
var client = SiteStreamGrpcClient.CreateForTesting(new CommunicationOptions { ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50) });
|
|
var cts = new CancellationTokenSource();
|
|
var connected = 0;
|
|
var connectedBeforeFirstEvent = false;
|
|
var events = 0;
|
|
var neverArrives = new TaskCompletionSource<Metadata>();
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-late-headers",
|
|
cts,
|
|
() => FakeCall(
|
|
new StubStreamReader(
|
|
new SiteStreamEvent { CorrelationId = "corr-late-headers" },
|
|
new SiteStreamEvent { CorrelationId = "corr-late-headers" }),
|
|
neverArrives.Task),
|
|
_ =>
|
|
{
|
|
if (events == 0) connectedBeforeFirstEvent = connected == 1;
|
|
events++;
|
|
},
|
|
_ => { },
|
|
() => { },
|
|
() => connected++);
|
|
|
|
Assert.Equal(2, events);
|
|
Assert.Equal(1, connected);
|
|
Assert.True(connectedBeforeFirstEvent);
|
|
}
|
|
|
|
// ── R2: batch unpacking on the client ───────────────────────────────────────
|
|
|
|
private static SiteStreamEvent Attr(string value, DateTimeOffset ts) => new()
|
|
{
|
|
AttributeChanged = new AttributeValueUpdate
|
|
{
|
|
InstanceUniqueName = "SiteA.Pump01",
|
|
AttributePath = "Modules.IO",
|
|
AttributeName = "Seq",
|
|
Value = value,
|
|
Quality = Quality.Good,
|
|
Timestamp = Timestamp.FromDateTimeOffset(ts)
|
|
}
|
|
};
|
|
|
|
[Fact]
|
|
public void ForEachEvent_PlainFrame_IsDeliveredAsIs()
|
|
{
|
|
// An OLD SITE (or any un-negotiated stream) sends one event per frame. The new
|
|
// client's unpack path must pass it straight through — this is the new-central ↔
|
|
// old-site skew direction.
|
|
var frame = Attr("1", DateTimeOffset.UnixEpoch);
|
|
var seen = new List<SiteStreamEvent>();
|
|
|
|
SiteStreamGrpcClient.ForEachEvent(frame, seen.Add);
|
|
|
|
Assert.Same(frame, Assert.Single(seen));
|
|
}
|
|
|
|
[Fact]
|
|
public void ForEachEvent_BatchFrame_UnpacksInOrderPreservingPerEventTimestamps()
|
|
{
|
|
// Order and per-event Timestamp fidelity are the two properties the downstream
|
|
// consumers (SiteAlarmAggregatorActor, DebugStreamBridgeActor) and the end-to-end
|
|
// latency measurement depend on.
|
|
var t0 = new DateTimeOffset(2026, 8, 15, 9, 0, 0, TimeSpan.Zero);
|
|
var frame = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "corr-batch",
|
|
Batch = new SiteStreamEventBatch
|
|
{
|
|
Events =
|
|
{
|
|
Attr("0", t0),
|
|
Attr("1", t0.AddMilliseconds(3)),
|
|
Attr("2", t0.AddMilliseconds(11))
|
|
}
|
|
}
|
|
};
|
|
|
|
var seen = new List<SiteStreamEvent>();
|
|
SiteStreamGrpcClient.ForEachEvent(frame, seen.Add);
|
|
|
|
Assert.Equal(["0", "1", "2"], seen.Select(e => e.AttributeChanged.Value));
|
|
Assert.Equal(
|
|
[t0, t0.AddMilliseconds(3), t0.AddMilliseconds(11)],
|
|
seen.Select(e => e.AttributeChanged.Timestamp.ToDateTimeOffset()));
|
|
}
|
|
|
|
[Fact]
|
|
public void ForEachEvent_IgnoresNestedAndUnknownInnerCases()
|
|
{
|
|
// The server never nests a batch inside a batch. A nested (or empty) inner frame
|
|
// from a malformed or hostile peer must be skipped, not followed — unpacking is
|
|
// deliberately non-recursive so a crafted frame cannot drive unbounded recursion.
|
|
var frame = new SiteStreamEvent
|
|
{
|
|
CorrelationId = "corr-nested",
|
|
Batch = new SiteStreamEventBatch
|
|
{
|
|
Events =
|
|
{
|
|
Attr("0", DateTimeOffset.UnixEpoch),
|
|
new SiteStreamEvent { Batch = new SiteStreamEventBatch { Events = { Attr("hidden", DateTimeOffset.UnixEpoch) } } },
|
|
new SiteStreamEvent(),
|
|
Attr("1", DateTimeOffset.UnixEpoch)
|
|
}
|
|
}
|
|
};
|
|
|
|
var seen = new List<SiteStreamEvent>();
|
|
SiteStreamGrpcClient.ForEachEvent(frame, seen.Add);
|
|
|
|
Assert.Equal(["0", "1"], seen.Select(e => e.AttributeChanged.Value));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ConsumeStream_MixedBatchedAndPlainFrames_DeliverEveryEventInOrder()
|
|
{
|
|
// A reconnect can straddle a site upgrade, so one stream may legitimately carry
|
|
// both frame shapes. Driving the real ConsumeStreamAsync with the real unpack
|
|
// proves the combination is flat and ordered from the consumer's point of view.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
var cts = new CancellationTokenSource();
|
|
var delivered = new List<string>();
|
|
|
|
void Deliver(SiteStreamEvent e) => delivered.Add(e.AttributeChanged.Value);
|
|
|
|
await client.ConsumeStreamAsync(
|
|
"corr-mixed",
|
|
cts,
|
|
() => FakeCall(new StubStreamReader(
|
|
Attr("0", DateTimeOffset.UnixEpoch),
|
|
new SiteStreamEvent
|
|
{
|
|
CorrelationId = "corr-mixed",
|
|
Batch = new SiteStreamEventBatch
|
|
{
|
|
Events = { Attr("1", DateTimeOffset.UnixEpoch), Attr("2", DateTimeOffset.UnixEpoch) }
|
|
}
|
|
},
|
|
Attr("3", DateTimeOffset.UnixEpoch))),
|
|
frame => SiteStreamGrpcClient.ForEachEvent(frame, Deliver),
|
|
_ => { },
|
|
() => { });
|
|
|
|
Assert.Equal(["0", "1", "2", "3"], delivered);
|
|
}
|
|
|
|
private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(StubStreamReader reader) =>
|
|
FakeCall(reader, Task.FromResult(new Metadata()));
|
|
|
|
private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(
|
|
StubStreamReader reader, Task<Metadata> responseHeaders) =>
|
|
new(reader,
|
|
responseHeaders,
|
|
() => Status.DefaultSuccess,
|
|
() => new Metadata(),
|
|
() => { });
|
|
|
|
/// <summary>
|
|
/// Server stream stand-in: yields the queued events, then either ends the stream (the
|
|
/// status-OK completion the site's max-lifetime cap produces) or throws.
|
|
/// </summary>
|
|
private sealed class StubStreamReader : IAsyncStreamReader<SiteStreamEvent>
|
|
{
|
|
private readonly Queue<SiteStreamEvent> _events;
|
|
private readonly Exception? _fault;
|
|
|
|
public StubStreamReader(params SiteStreamEvent[] events)
|
|
{
|
|
_events = new Queue<SiteStreamEvent>(events);
|
|
}
|
|
|
|
public StubStreamReader(Exception fault)
|
|
: this()
|
|
{
|
|
_fault = fault;
|
|
}
|
|
|
|
public SiteStreamEvent Current { get; private set; } = null!;
|
|
|
|
public Task<bool> MoveNext(CancellationToken cancellationToken)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
|
|
if (_events.Count > 0)
|
|
{
|
|
Current = _events.Dequeue();
|
|
return Task.FromResult(true);
|
|
}
|
|
|
|
return _fault is null ? Task.FromResult(false) : Task.FromException<bool>(_fault);
|
|
}
|
|
}
|
|
|
|
// --- Communication-003 regression tests ---
|
|
|
|
[Fact]
|
|
public void RegisterSubscription_ReusedCorrelationId_CancelsPriorCts_WithoutDisposingIt()
|
|
{
|
|
// Two SubscribeAsync calls briefly sharing a correlation ID (reconnect race).
|
|
// Inserting the second must CANCEL the first — and must NOT dispose it (arch-review
|
|
// 2026-08-19 P3-1, re-pinning the earlier cancel-AND-dispose contract): the prior
|
|
// handler's ConsumeStreamAsync is still running and still reads prior.Token, so
|
|
// disposing from the replacing path throws ObjectDisposedException inside that loop.
|
|
// The owning SubscribeAsync/SubscribeSiteAsync disposes it as its loop unwinds.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
using var first = new CancellationTokenSource();
|
|
using var second = new CancellationTokenSource();
|
|
|
|
client.RegisterSubscription("corr-shared", first);
|
|
client.RegisterSubscription("corr-shared", second);
|
|
|
|
Assert.True(first.IsCancellationRequested);
|
|
// Still usable by the prior handler that owns it — no ObjectDisposedException.
|
|
Assert.True(first.Token.IsCancellationRequested);
|
|
|
|
// The second (live) CTS must remain intact.
|
|
Assert.False(second.IsCancellationRequested);
|
|
}
|
|
|
|
[Fact]
|
|
public void RegisterSubscription_ReplacingAnAlreadyDisposedPriorCts_DoesNotThrowIntoTheNewSubscription()
|
|
{
|
|
// The prior handler's loop already unwound and disposed its own CTS while a replacement
|
|
// subscription is registering. The unguarded prior.Cancel() threw ObjectDisposedException
|
|
// straight out of the NEW SubscribeAsync, which reads it as a fault of the CURRENT
|
|
// generation and burns a reconnect on a phantom failure.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
var prior = new CancellationTokenSource();
|
|
client.RegisterSubscription("corr-shared", prior);
|
|
prior.Dispose();
|
|
|
|
using var replacement = new CancellationTokenSource();
|
|
client.RegisterSubscription("corr-shared", replacement);
|
|
|
|
// The replacement owns the slot and is untouched.
|
|
Assert.False(replacement.IsCancellationRequested);
|
|
client.Unsubscribe("corr-shared");
|
|
Assert.True(replacement.IsCancellationRequested);
|
|
}
|
|
|
|
[Fact]
|
|
public void Unsubscribe_AfterClientDisposal_IsANoOp()
|
|
{
|
|
// Disposal cancels every live subscription and clears the map; a late Unsubscribe from
|
|
// the consuming actor's teardown must not throw (an ObjectDisposedException here used to
|
|
// escape into the caller's cleanup path).
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
using var cts = new CancellationTokenSource();
|
|
client.AddSubscriptionForTesting("corr-1", cts);
|
|
|
|
client.Dispose();
|
|
Assert.True(cts.IsCancellationRequested);
|
|
|
|
client.Unsubscribe("corr-1");
|
|
client.Unsubscribe("never-registered");
|
|
}
|
|
|
|
[Fact]
|
|
public void Dispose_WithAnAlreadyDisposedSubscriptionCts_StillCancelsTheRest()
|
|
{
|
|
// ReleaseResources iterated the live CTSes unguarded, so one already-disposed entry
|
|
// aborted the whole loop — leaving later subscriptions uncancelled AND (via the
|
|
// exception escaping into SiteStreamGrpcClientFactory's disposal loop) later clients'
|
|
// channels undisposed.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
var disposedEarly = new CancellationTokenSource();
|
|
using var live = new CancellationTokenSource();
|
|
client.AddSubscriptionForTesting("corr-disposed", disposedEarly);
|
|
client.AddSubscriptionForTesting("corr-live", live);
|
|
disposedEarly.Dispose();
|
|
|
|
// Deterministic regardless of dictionary iteration order: the disposed entry is always
|
|
// visited, so the unguarded loop always threw out of Dispose().
|
|
Assert.Null(Record.Exception(client.Dispose));
|
|
|
|
Assert.True(live.IsCancellationRequested);
|
|
}
|
|
|
|
[Fact]
|
|
public void RemoveSubscription_OnlyRemovesOwnCts_NotAReplacement()
|
|
{
|
|
// First call's finally must NOT remove the second call's live entry.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
var first = new CancellationTokenSource();
|
|
var second = new CancellationTokenSource();
|
|
|
|
client.RegisterSubscription("corr-shared", first);
|
|
// A racing second SubscribeAsync replaces the entry.
|
|
client.RegisterSubscription("corr-shared", second);
|
|
|
|
// The first call's finally runs and tries to remove its (already-replaced) entry.
|
|
client.RemoveSubscription("corr-shared", first);
|
|
|
|
// The live (second) subscription must still be cancellable via Unsubscribe.
|
|
Assert.False(second.IsCancellationRequested);
|
|
client.Unsubscribe("corr-shared");
|
|
Assert.True(second.IsCancellationRequested);
|
|
}
|
|
}
|