scadalink-design/tests/ScadaLink.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs

using Google.Protobuf.WellKnownTypes;
using ScadaLink.Communication.Grpc;
using ScadaLink.Commons.Messages.Streaming;
using ScadaLink.Commons.Types.Enums;

namespace ScadaLink.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 evt = new SiteStreamEvent
        {
            CorrelationId = "corr-2",
            AlarmChanged = new AlarmStateUpdate
            {
                InstanceUniqueName = "Site1.Motor01",
                AlarmName = "OverTemp",
                State = AlarmStateEnum.AlarmStateActive,
                Priority = 3,
                Timestamp = Timestamp.FromDateTimeOffset(ts)
            }
        };

        var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);

        var alarm = Assert.IsType<AlarmStateChanged>(result);
        Assert.Equal("Site1.Motor01", alarm.InstanceUniqueName);
        Assert.Equal("OverTemp", alarm.AlarmName);
        Assert.Equal(AlarmState.Active, alarm.State);
        Assert.Equal(3, alarm.Priority);
        Assert.Equal(ts, alarm.Timestamp);
    }

    [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);
    }

    // --- 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);
    }
}