junction/tests/Junction.Tests/Unit/PollingEngineTests.cs

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
{
    /// <summary>
    /// Behavior tests for <see cref="PollingEngine"/> using a hand-rolled fake driver.
    /// Timings are deliberately generous to avoid CI flake.
    /// </summary>
    public class PollingEngineTests
    {
        private static PollingEngine NewEngine() =>
            new PollingEngine(NullLogger<PollingEngine>.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<DataItem>());

        [Fact]
        public async Task RunAsync_PollsRepeatedly_OverShortInterval()
        {
            var machine = MachineWithInterval(TimeSpan.FromMilliseconds(25));
            var driver = new FakeDriver(m => Result<MachineSnapshot>.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<MachineSnapshot>.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<MachineSnapshot>();

            // 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<MachineSnapshot>.Fail(OperationError.Of("fake", "boom"));
                }

                return Result<MachineSnapshot>.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<MachineSnapshot>.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<MachineSnapshot>.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<MachineSnapshot>.Fail(OperationError.Of("fake", "down")));

            var emitted = new List<MachineSnapshot>();
            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<MachineSnapshot>();

            // 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<MachineSnapshot>.Fail(OperationError.Of("fake", "blip"));
                }

                return Result<MachineSnapshot>.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<MachineSnapshot>();

            // 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<MachineSnapshot>.Fail(OperationError.Of("fake", "down"));
                }

                return Result<MachineSnapshot>.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");
            }
        }

        /// <summary>Hand-rolled fake driver; behavior supplied by a delegate.</summary>
        private sealed class FakeDriver : IProtocolDriver
        {
            private readonly Func<Machine, Result<MachineSnapshot>> _behavior;

            public FakeDriver(Func<Machine, Result<MachineSnapshot>> behavior)
            {
                _behavior = behavior;
            }

            public string ProtocolId => "fake";

            public Task<Result<MachineSnapshot>> ReadCurrentAsync(CancellationToken cancellationToken)
            {
                cancellationToken.ThrowIfCancellationRequested();
                return Task.FromResult(_behavior(null!));
            }

            public Task<Result<IReadOnlyList<DataItemDescriptor>>> ProbeAsync(CancellationToken cancellationToken) =>
                Task.FromResult(Result<IReadOnlyList<DataItemDescriptor>>.Ok(Array.Empty<DataItemDescriptor>()));
        }
    }
}