Joseph Doherty
442 lines
16 KiB
C#
442 lines
16 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 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);
|
|
}
|
|
|
|
private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(StubStreamReader reader) =>
|
|
new(reader,
|
|
Task.FromResult(new Metadata()),
|
|
() => 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_CancelsAndDisposesPriorCts()
|
|
{
|
|
// Two SubscribeAsync calls briefly sharing a correlation ID (reconnect race).
|
|
// Inserting the second must cancel + dispose the first so it does not leak.
|
|
var client = SiteStreamGrpcClient.CreateForTesting();
|
|
|
|
var first = new CancellationTokenSource();
|
|
var second = new CancellationTokenSource();
|
|
|
|
client.RegisterSubscription("corr-shared", first);
|
|
client.RegisterSubscription("corr-shared", second);
|
|
|
|
Assert.True(first.IsCancellationRequested);
|
|
// Disposed CTS throws ObjectDisposedException when its token is touched.
|
|
Assert.Throws<ObjectDisposedException>(() => _ = first.Token);
|
|
|
|
// The second (live) CTS must remain intact.
|
|
Assert.False(second.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);
|
|
}
|
|
}
|