using System; using System.Collections.Generic; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; using FluentAssertions; using Xunit; using ZB.MOM.WW.LmxProxy.Host.Domain; using ZB.MOM.WW.LmxProxy.Host.Subscriptions; namespace ZB.MOM.WW.LmxProxy.Host.Tests.Subscriptions { public class SubscriptionManagerTests { /// Fake IScadaClient for testing (no COM dependency). private class FakeScadaClient : IScadaClient { public bool IsConnected => true; public ConnectionState ConnectionState => ConnectionState.Connected; public event EventHandler? ConnectionStateChanged; public Task ConnectAsync(CancellationToken ct = default) => Task.CompletedTask; public Task DisconnectAsync(CancellationToken ct = default) => Task.CompletedTask; public Task ReadAsync(string address, CancellationToken ct = default) => Task.FromResult(Vtq.Good(42.0)); public Task> ReadBatchAsync(IEnumerable addresses, CancellationToken ct = default) => Task.FromResult>(new Dictionary()); public Task WriteAsync(string address, object value, CancellationToken ct = default) => Task.CompletedTask; public Task WriteBatchAsync(IReadOnlyDictionary values, CancellationToken ct = default) => Task.CompletedTask; public Task<(bool flagReached, int elapsedMs)> WriteBatchAndWaitAsync( IReadOnlyDictionary values, string flagTag, object flagValue, int timeoutMs, int pollIntervalMs, CancellationToken ct = default) => Task.FromResult((false, 0)); public Task SubscribeAsync(IEnumerable addresses, Action callback, CancellationToken ct = default) => Task.FromResult(new FakeSubscriptionHandle()); public ValueTask DisposeAsync() => default; // Suppress unused event warning internal void FireEvent() => ConnectionStateChanged?.Invoke(this, null!); private class FakeSubscriptionHandle : IAsyncDisposable { public ValueTask DisposeAsync() => default; } } [Fact] public void Subscribe_ReturnsChannelReader() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); var reader = sm.Subscribe("client1", new[] { "Tag1", "Tag2" }, cts.Token); reader.Should().NotBeNull(); } [Fact] public async Task OnTagValueChanged_FansOutToSubscribedClients() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); var reader = sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); var vtq = Vtq.Good(42.0); sm.OnTagValueChanged("Motor.Speed", vtq); var result = await reader.ReadAsync(cts.Token); result.address.Should().Be("Motor.Speed"); result.vtq.Value.Should().Be(42.0); result.vtq.Quality.Should().Be(Quality.Good); } [Fact] public async Task OnTagValueChanged_MultipleClients_BothReceive() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); var reader1 = sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); var reader2 = sm.Subscribe("client2", new[] { "Motor.Speed" }, cts.Token); sm.OnTagValueChanged("Motor.Speed", Vtq.Good(99.0)); var r1 = await reader1.ReadAsync(cts.Token); var r2 = await reader2.ReadAsync(cts.Token); r1.vtq.Value.Should().Be(99.0); r2.vtq.Value.Should().Be(99.0); } [Fact] public void OnTagValueChanged_NonSubscribedTag_NoDelivery() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); var reader = sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); sm.OnTagValueChanged("Motor.Torque", Vtq.Good(10.0)); // Channel should be empty reader.TryRead(out _).Should().BeFalse(); } [Fact] public void UnsubscribeClient_CompletesChannel() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); var reader = sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); sm.UnsubscribeClient("client1"); // Channel should be completed reader.Completion.IsCompleted.Should().BeTrue(); } [Fact] public void UnsubscribeClient_RemovesFromTagSubscriptions() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); sm.UnsubscribeClient("client1"); var stats = sm.GetStats(); stats.TotalClients.Should().Be(0); stats.TotalTags.Should().Be(0); } [Fact] public void RefCounting_LastClientUnsubscribeRemovesTag() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); sm.Subscribe("client2", new[] { "Motor.Speed" }, cts.Token); sm.GetStats().TotalTags.Should().Be(1); sm.UnsubscribeClient("client1"); sm.GetStats().TotalTags.Should().Be(1); // client2 still subscribed sm.UnsubscribeClient("client2"); sm.GetStats().TotalTags.Should().Be(0); // last client gone } [Fact] public void NotifyDisconnection_SendsBadQualityToAll() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); var reader = sm.Subscribe("client1", new[] { "Motor.Speed", "Motor.Torque" }, cts.Token); sm.NotifyDisconnection(); // Should receive 2 bad quality messages reader.TryRead(out var r1).Should().BeTrue(); r1.vtq.Quality.Should().Be(Quality.Bad_NotConnected); reader.TryRead(out var r2).Should().BeTrue(); r2.vtq.Quality.Should().Be(Quality.Bad_NotConnected); } [Fact] public void Backpressure_DropOldest_DropsWhenFull() { using var sm = new SubscriptionManager(new FakeScadaClient(), channelCapacity: 3); using var cts = new CancellationTokenSource(); var reader = sm.Subscribe("client1", new[] { "Motor.Speed" }, cts.Token); // Fill the channel beyond capacity for (int i = 0; i < 10; i++) { sm.OnTagValueChanged("Motor.Speed", Vtq.Good((double)i)); } // Should have exactly 3 messages (capacity limit) int count = 0; while (reader.TryRead(out _)) count++; count.Should().Be(3); } [Fact] public void GetStats_ReturnsCorrectCounts() { using var sm = new SubscriptionManager(new FakeScadaClient()); using var cts = new CancellationTokenSource(); sm.Subscribe("c1", new[] { "Tag1", "Tag2" }, cts.Token); sm.Subscribe("c2", new[] { "Tag2", "Tag3" }, cts.Token); var stats = sm.GetStats(); stats.TotalClients.Should().Be(2); stats.TotalTags.Should().Be(3); // Tag1, Tag2, Tag3 stats.ActiveSubscriptions.Should().Be(4); // c1:Tag1, c1:Tag2, c2:Tag2, c2:Tag3 } } }