using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Backend.MxAccess; using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Backend.Stability; namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Tests; [Trait("Category", "Unit")] public sealed class GalaxyRuntimeProbeManagerTests { private sealed class FakeSubscriber { public readonly ConcurrentDictionary> Subs = new(); public readonly ConcurrentQueue UnsubCalls = new(); public bool FailSubscribeFor { get; set; } public string? FailSubscribeTag { get; set; } public Task Subscribe(string probe, Action cb) { if (FailSubscribeFor && string.Equals(probe, FailSubscribeTag, StringComparison.OrdinalIgnoreCase)) throw new InvalidOperationException("subscribe refused"); Subs[probe] = cb; return Task.CompletedTask; } public Task Unsubscribe(string probe) { UnsubCalls.Enqueue(probe); Subs.TryRemove(probe, out _); return Task.CompletedTask; } } private static Vtq Good(bool scanState) => new(scanState, DateTime.UtcNow, 192); private static Vtq Bad() => new(null, DateTime.UtcNow, 0); [Fact] public async Task Sync_subscribes_to_ScanState_per_host() { var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", GalaxyRuntimeProbeManager.CategoryWinPlatform), new HostProbeTarget("EngineB", GalaxyRuntimeProbeManager.CategoryAppEngine), }); mgr.ActiveProbeCount.ShouldBe(2); subs.Subs.ShouldContainKey("PlatformA.ScanState"); subs.Subs.ShouldContainKey("EngineB.ScanState"); } [Fact] public async Task Sync_is_idempotent_on_repeat_call_with_same_set() { var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe); var targets = new[] { new HostProbeTarget("PlatformA", 1) }; await mgr.SyncAsync(targets); await mgr.SyncAsync(targets); mgr.ActiveProbeCount.ShouldBe(1); subs.Subs.Count.ShouldBe(1); subs.UnsubCalls.Count.ShouldBe(0); } [Fact] public async Task Sync_unadvises_removed_hosts() { var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1), new HostProbeTarget("PlatformB", 1), }); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); mgr.ActiveProbeCount.ShouldBe(1); subs.UnsubCalls.ShouldContain("PlatformB.ScanState"); } [Fact] public async Task Subscribe_failure_rolls_back_host_entry_so_later_transitions_do_not_fire_stale_events() { var subs = new FakeSubscriber { FailSubscribeFor = true, FailSubscribeTag = "PlatformA.ScanState" }; using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); mgr.ActiveProbeCount.ShouldBe(0); // rolled back mgr.GetState("PlatformA").ShouldBe(HostRuntimeState.Unknown); } [Fact] public async Task Unknown_to_Running_does_not_fire_StateChanged() { var now = new DateTime(2026, 4, 18, 10, 0, 0, DateTimeKind.Utc); var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe, () => now); var transitions = new ConcurrentQueue(); mgr.StateChanged += (_, t) => transitions.Enqueue(t); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(true)); mgr.GetState("PlatformA").ShouldBe(HostRuntimeState.Running); transitions.Count.ShouldBe(0); // startup transition, operators don't care } [Fact] public async Task Running_to_Stopped_fires_StateChanged_with_both_states() { var now = new DateTime(2026, 4, 18, 10, 0, 0, DateTimeKind.Utc); var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe, () => now); var transitions = new ConcurrentQueue(); mgr.StateChanged += (_, t) => transitions.Enqueue(t); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(true)); // Unknown→Running (silent) subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(false)); // Running→Stopped (fires) transitions.Count.ShouldBe(1); transitions.TryDequeue(out var t).ShouldBeTrue(); t!.TagName.ShouldBe("PlatformA"); t.OldState.ShouldBe(HostRuntimeState.Running); t.NewState.ShouldBe(HostRuntimeState.Stopped); } [Fact] public async Task Stopped_to_Running_fires_StateChanged_for_recovery() { var now = new DateTime(2026, 4, 18, 10, 0, 0, DateTimeKind.Utc); var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe, () => now); var transitions = new ConcurrentQueue(); mgr.StateChanged += (_, t) => transitions.Enqueue(t); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(true)); // Unknown→Running (silent) subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(false)); // Running→Stopped (fires) subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(true)); // Stopped→Running (fires) transitions.Count.ShouldBe(2); } [Fact] public async Task Unknown_to_Stopped_fires_StateChanged_for_first_known_bad_signal() { var now = new DateTime(2026, 4, 18, 10, 0, 0, DateTimeKind.Utc); var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe, () => now); var transitions = new ConcurrentQueue(); mgr.StateChanged += (_, t) => transitions.Enqueue(t); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); // First callback is bad-quality — we must flag the host Stopped so operators see it. subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Bad()); transitions.Count.ShouldBe(1); transitions.TryDequeue(out var t).ShouldBeTrue(); t!.OldState.ShouldBe(HostRuntimeState.Unknown); t.NewState.ShouldBe(HostRuntimeState.Stopped); } [Fact] public async Task Repeated_Good_Running_callbacks_do_not_fire_duplicate_events() { var now = new DateTime(2026, 4, 18, 10, 0, 0, DateTimeKind.Utc); var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe, () => now); var count = 0; mgr.StateChanged += (_, _) => Interlocked.Increment(ref count); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1) }); for (var i = 0; i < 5; i++) subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(true)); count.ShouldBe(0); // only the silent Unknown→Running on the first, no events after } [Fact] public async Task Unknown_callback_for_non_tracked_probe_is_dropped() { var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe); mgr.OnProbeCallback("ProbeForSomeoneElse.ScanState", Good(true)); mgr.ActiveProbeCount.ShouldBe(0); } [Fact] public async Task Snapshot_reports_current_state_for_every_tracked_host() { var now = new DateTime(2026, 4, 18, 10, 0, 0, DateTimeKind.Utc); var subs = new FakeSubscriber(); using var mgr = new GalaxyRuntimeProbeManager(subs.Subscribe, subs.Unsubscribe, () => now); await mgr.SyncAsync(new[] { new HostProbeTarget("PlatformA", 1), new HostProbeTarget("EngineB", 3), }); subs.Subs["PlatformA.ScanState"]("PlatformA.ScanState", Good(true)); // Running subs.Subs["EngineB.ScanState"]("EngineB.ScanState", Bad()); // Stopped var snap = mgr.SnapshotStates(); snap.Count.ShouldBe(2); snap.ShouldContain(s => s.TagName == "PlatformA" && s.State == HostRuntimeState.Running); snap.ShouldContain(s => s.TagName == "EngineB" && s.State == HostRuntimeState.Stopped); } [Fact] public void IsRuntimeHost_recognizes_WinPlatform_and_AppEngine_category_ids() { new HostProbeTarget("X", GalaxyRuntimeProbeManager.CategoryWinPlatform).IsRuntimeHost.ShouldBeTrue(); new HostProbeTarget("X", GalaxyRuntimeProbeManager.CategoryAppEngine).IsRuntimeHost.ShouldBeTrue(); new HostProbeTarget("X", 4 /* $Area */).IsRuntimeHost.ShouldBeFalse(); new HostProbeTarget("X", 11 /* $ApplicationObject */).IsRuntimeHost.ShouldBeFalse(); } }