using System.Threading.Channels; using Akka.Actor; using Akka.TestKit.Xunit2; using Google.Protobuf.WellKnownTypes; using ScadaLink.Commons.Messages.Streaming; using ScadaLink.Commons.Types.Enums; using ScadaLink.Communication.Actors; using ScadaLink.Communication.Grpc; using AlarmState = ScadaLink.Commons.Types.Enums.AlarmState; namespace ScadaLink.Communication.Tests.Grpc; public class StreamRelayActorTests : TestKit { [Fact] public void RelaysAttributeValueChanged_ToProtoEvent() { var channel = Channel.CreateUnbounded(); var correlationId = "corr-attr-1"; var actor = Sys.ActorOf(Props.Create(() => new StreamRelayActor(correlationId, channel.Writer))); var timestamp = new DateTimeOffset(2026, 3, 21, 10, 30, 0, TimeSpan.Zero); var domainEvent = new AttributeValueChanged( "Site1.Pump01", "Modules.Pressure", "CurrentPSI", 42.5, "Good", timestamp); actor.Tell(domainEvent); var success = channel.Reader.TryRead(out var protoEvent); if (!success) { // Give a moment for async processing Thread.Sleep(500); success = channel.Reader.TryRead(out protoEvent); } Assert.True(success, "Expected a proto event on the channel"); Assert.NotNull(protoEvent); Assert.Equal(SiteStreamEvent.EventOneofCase.AttributeChanged, protoEvent.EventCase); Assert.Equal(correlationId, protoEvent.CorrelationId); var attr = protoEvent.AttributeChanged; Assert.Equal("Site1.Pump01", attr.InstanceUniqueName); Assert.Equal("Modules.Pressure", attr.AttributePath); Assert.Equal("CurrentPSI", attr.AttributeName); Assert.Equal("42.5", attr.Value); Assert.Equal(Quality.Good, attr.Quality); Assert.Equal(Timestamp.FromDateTimeOffset(timestamp), attr.Timestamp); } [Fact] public void RelaysAlarmStateChanged_ToProtoEvent() { var channel = Channel.CreateUnbounded(); var correlationId = "corr-alarm-1"; var actor = Sys.ActorOf(Props.Create(() => new StreamRelayActor(correlationId, channel.Writer))); var timestamp = new DateTimeOffset(2026, 3, 21, 11, 0, 0, TimeSpan.Zero); var domainEvent = new AlarmStateChanged( "Site1.Pump01", "HighPressure", AlarmState.Active, 2, timestamp); actor.Tell(domainEvent); var success = channel.Reader.TryRead(out var protoEvent); if (!success) { Thread.Sleep(500); success = channel.Reader.TryRead(out protoEvent); } Assert.True(success, "Expected a proto event on the channel"); Assert.NotNull(protoEvent); Assert.Equal(SiteStreamEvent.EventOneofCase.AlarmChanged, protoEvent.EventCase); Assert.Equal(correlationId, protoEvent.CorrelationId); var alarm = protoEvent.AlarmChanged; Assert.Equal("Site1.Pump01", alarm.InstanceUniqueName); Assert.Equal("HighPressure", alarm.AlarmName); Assert.Equal(AlarmStateEnum.AlarmStateActive, alarm.State); Assert.Equal(2, alarm.Priority); Assert.Equal(Timestamp.FromDateTimeOffset(timestamp), alarm.Timestamp); } [Fact] public void SetsCorrelationId_OnAllEvents() { var channel = Channel.CreateUnbounded(); var correlationId = "corr-multi-42"; var actor = Sys.ActorOf(Props.Create(() => new StreamRelayActor(correlationId, channel.Writer))); var ts = DateTimeOffset.UtcNow; actor.Tell(new AttributeValueChanged("Inst1", "Path", "Name", 1, "Good", ts)); actor.Tell(new AlarmStateChanged("Inst1", "Alarm1", AlarmState.Normal, 1, ts)); actor.Tell(new AttributeValueChanged("Inst2", "Path2", "Name2", null, "Bad", ts)); // Allow messages to process Thread.Sleep(500); var events = new List(); while (channel.Reader.TryRead(out var evt)) { events.Add(evt); } Assert.Equal(3, events.Count); Assert.All(events, e => Assert.Equal(correlationId, e.CorrelationId)); } [Fact] public void DropsEvent_WhenChannelFull() { var channel = Channel.CreateBounded(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.Wait }); var correlationId = "corr-drop-1"; var actor = Sys.ActorOf(Props.Create(() => new StreamRelayActor(correlationId, channel.Writer))); var ts = DateTimeOffset.UtcNow; // Fill the channel with one item directly var filler = new SiteStreamEvent { CorrelationId = "filler" }; Assert.True(channel.Writer.TryWrite(filler)); // Send another event — should be dropped (channel full), no exception actor.Tell(new AttributeValueChanged("Inst1", "Path", "Name", 1, "Good", ts)); // Allow message to process Thread.Sleep(500); // Channel should still have exactly 1 item (the filler) Assert.True(channel.Reader.TryRead(out var item)); Assert.Equal("filler", item.CorrelationId); Assert.False(channel.Reader.TryRead(out _)); } [Theory] [InlineData("Good", Quality.Good)] [InlineData("Uncertain", Quality.Uncertain)] [InlineData("Bad", Quality.Bad)] [InlineData("Unknown", Quality.Unspecified)] [InlineData("", Quality.Unspecified)] [InlineData("good", Quality.Unspecified)] public void MapsQualityString_ToProtoEnum(string qualityString, Quality expectedProto) { var channel = Channel.CreateUnbounded(); var actor = Sys.ActorOf(Props.Create(() => new StreamRelayActor("corr", channel.Writer))); var ts = DateTimeOffset.UtcNow; actor.Tell(new AttributeValueChanged("Inst", "Path", "Name", 1, qualityString, ts)); Thread.Sleep(500); Assert.True(channel.Reader.TryRead(out var evt)); Assert.Equal(expectedProto, evt.AttributeChanged.Quality); } [Fact] public void NullValue_MapsToEmptyString() { var channel = Channel.CreateUnbounded(); var actor = Sys.ActorOf(Props.Create(() => new StreamRelayActor("corr", channel.Writer))); var ts = DateTimeOffset.UtcNow; actor.Tell(new AttributeValueChanged("Inst", "Path", "Name", null, "Good", ts)); Thread.Sleep(500); Assert.True(channel.Reader.TryRead(out var evt)); Assert.Equal("", evt.AttributeChanged.Value); } }