using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using Junction.Core.Polling; using Junction.Domain; using Junction.Domain.Models; using Junction.Domain.Protocols; using Microsoft.Extensions.Logging.Abstractions; using Xunit; namespace Junction.Tests.Unit { /// /// Behavior tests for using a hand-rolled fake driver. /// Timings are deliberately generous to avoid CI flake. /// public class PollingEngineTests { private static PollingEngine NewEngine() => new PollingEngine(NullLogger.Instance); private static Machine MachineWithInterval(TimeSpan interval) => new Machine(Guid.NewGuid(), "M1", "fake", null, interval); private static MachineSnapshot Snapshot(Guid machineId) => new MachineSnapshot( machineId, DateTimeOffset.UtcNow, ConnectionState.Connected, Array.Empty()); [Fact] public async Task RunAsync_PollsRepeatedly_OverShortInterval() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(25)); var driver = new FakeDriver(m => Result.Ok(Snapshot(machine.Id))); int count = 0; using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync(machine, driver, _ => Interlocked.Increment(ref count), cts.Token); await Task.Delay(200); cts.Cancel(); await loop; Assert.True(count >= 2, $"expected >= 2 polls, got {count}"); } [Fact] public async Task RunAsync_Cancellation_StopsLoop_NoException() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(20)); var driver = new FakeDriver(m => Result.Ok(Snapshot(machine.Id))); using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync(machine, driver, _ => { }, cts.Token); await Task.Delay(60); cts.Cancel(); // Must complete promptly and without propagating any exception. var completed = await Task.WhenAny(loop, Task.Delay(1000)) == loop; Assert.True(completed, "loop did not stop promptly after cancellation"); await loop; // would rethrow if it faulted Assert.True(loop.IsCompletedSuccessfully); } [Fact] public async Task RunAsync_DriverFail_DoesNotEmit_AndKeepsLooping() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(25)); int calls = 0; var emitted = new List(); // Fail on cycle 1, succeed thereafter. Proves the loop survives a fault. var driver = new FakeDriver(m => { int c = Interlocked.Increment(ref calls); if (c == 1) { return Result.Fail(OperationError.Of("fake", "boom")); } return Result.Ok(Snapshot(machine.Id)); }); using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync(machine, driver, s => { lock (emitted) { emitted.Add(s); } }, cts.Token); await Task.Delay(200); cts.Cancel(); await loop; Assert.True(calls >= 2, $"expected loop to continue past the failed cycle, calls={calls}"); lock (emitted) { // First (failed) cycle emitted nothing; a later Ok cycle did. Assert.NotEmpty(emitted); Assert.True(emitted.Count < calls, "a failed cycle must not have emitted a snapshot"); } } [Fact] public async Task RunAsync_Ok_ForwardsSnapshotIntact() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(20)); var expected = Snapshot(machine.Id); var driver = new FakeDriver(m => Result.Ok(expected)); MachineSnapshot? received = null; using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync(machine, driver, s => { received = s; cts.Cancel(); }, cts.Token); await loop; Assert.Same(expected, received); } [Fact] public async Task RunAsync_DriverReturnsCancelled_StopsLoop() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(20)); int calls = 0; var driver = new FakeDriver(m => { Interlocked.Increment(ref calls); return Result.Cancelled(); }); using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync(machine, driver, _ => { }, cts.Token); var completed = await Task.WhenAny(loop, Task.Delay(1000)) == loop; Assert.True(completed, "loop did not stop on cancelled result"); await loop; Assert.Equal(1, calls); } [Fact] public async Task RunAsync_ConsecutiveFails_EmitsSingleDisconnectedSnapshot_AtThreshold() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(10)); // Always fail: after `threshold` fails a synthetic Disconnected must be emitted once. var driver = new FakeDriver(m => Result.Fail(OperationError.Of("fake", "down"))); var emitted = new List(); using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync( machine, driver, s => { lock (emitted) { emitted.Add(s); } }, cts.Token, offlineThreshold: 3); // Plenty of time for many failed cycles. await Task.Delay(300); cts.Cancel(); await loop; lock (emitted) { // Only the synthetic Disconnected, and exactly once (no spam) despite many fails. Assert.Single(emitted); Assert.Equal(ConnectionState.Disconnected, emitted[0].ConnectionState); Assert.Equal(machine.Id, emitted[0].MachineId); Assert.Empty(emitted[0].Items); } } [Fact] public async Task RunAsync_BelowThreshold_NoSyntheticDisconnected() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(10)); int calls = 0; var emitted = new List(); // Fail twice then succeed forever: with threshold 3 the episode never trips. var driver = new FakeDriver(m => { int c = Interlocked.Increment(ref calls); if (c <= 2) { return Result.Fail(OperationError.Of("fake", "blip")); } return Result.Ok(Snapshot(machine.Id)); }); using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync( machine, driver, s => { lock (emitted) { emitted.Add(s); } }, cts.Token, offlineThreshold: 3); await Task.Delay(200); cts.Cancel(); await loop; lock (emitted) { Assert.NotEmpty(emitted); Assert.DoesNotContain(emitted, s => s.ConnectionState == ConnectionState.Disconnected); } } [Fact] public async Task RunAsync_RecoveryAfterOffline_ResetsAndEmitsConnected_ThenReDisconnectsOncePerEpisode() { var machine = MachineWithInterval(TimeSpan.FromMilliseconds(10)); int calls = 0; var emitted = new List(); // Episode 1: 3 fails -> Disconnected. Then 1 Ok -> Connected (resets). // Episode 2: 3 fails -> Disconnected again (proves reset + one-per-episode). var driver = new FakeDriver(m => { int c = Interlocked.Increment(ref calls); // cycles 1-3 fail, 4 ok, 5-7 fail, then ok forever. bool fail = c <= 3 || (c >= 5 && c <= 7); if (fail) { return Result.Fail(OperationError.Of("fake", "down")); } return Result.Ok(Snapshot(machine.Id)); }); using var cts = new CancellationTokenSource(); var loop = NewEngine().RunAsync( machine, driver, s => { lock (emitted) { emitted.Add(s); } }, cts.Token, offlineThreshold: 3); // Wait until we observe both disconnected episodes (2) or time out. var sw = System.Diagnostics.Stopwatch.StartNew(); while (sw.Elapsed < TimeSpan.FromSeconds(3)) { lock (emitted) { int disc = emitted.FindAll(s => s.ConnectionState == ConnectionState.Disconnected).Count; int conn = emitted.FindAll(s => s.ConnectionState == ConnectionState.Connected).Count; if (disc >= 2 && conn >= 1) break; } await Task.Delay(15); } cts.Cancel(); await loop; lock (emitted) { int disconnected = emitted.FindAll(s => s.ConnectionState == ConnectionState.Disconnected).Count; int connected = emitted.FindAll(s => s.ConnectionState == ConnectionState.Connected).Count; Assert.Equal(2, disconnected); // one per episode, no spam Assert.True(connected >= 1, "recovery must emit at least one Connected snapshot"); } } /// Hand-rolled fake driver; behavior supplied by a delegate. private sealed class FakeDriver : IProtocolDriver { private readonly Func> _behavior; public FakeDriver(Func> behavior) { _behavior = behavior; } public string ProtocolId => "fake"; public Task> ReadCurrentAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); return Task.FromResult(_behavior(null!)); } public Task>> ProbeAsync(CancellationToken cancellationToken) => Task.FromResult(Result>.Ok(Array.Empty())); } } }