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