using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Xunit;
using ytLive.Services;
namespace ytLive.Tests;
///
/// TASK 32·1 — blip retry, exercised through with fake steps: no ffmpeg,
/// no network, no waiting on the real schedule. The rules that matter: the first attempt is immediate,
/// the audio pipe is recreated before every push, a reconnect that lands on a COMPLETE broadcast is
/// reported honestly instead of pretending to be live, and giving up is a definite outcome, not a hang.
///
public class StreamRecoveryTests
{
private static StreamRetryPolicy NoWaits(int attempts = 4)
=> new(Enumerable.Repeat(TimeSpan.Zero, attempts).ToArray());
[Fact]
public void DefaultPolicy_StartsImmediately_ThenOneTwoThreeSeconds()
{
var policy = StreamRetryPolicy.Default;
Assert.Equal(4, policy.Attempts);
Assert.Equal(TimeSpan.Zero, policy.WaitBefore(0));
Assert.Equal(TimeSpan.FromSeconds(1), policy.WaitBefore(1));
Assert.Equal(TimeSpan.FromSeconds(2), policy.WaitBefore(2));
Assert.Equal(TimeSpan.FromSeconds(3), policy.WaitBefore(3));
}
[Fact]
public async Task RecoversOnTheSecondAttempt_AndRestartsTheAudioPipeEachTime()
{
var pushes = 0;
var audioRestarts = 0;
var reported = new List();
var recovery = new StreamRecovery(NoWaits(), new StreamRecoverySteps
{
RestartAudio = _ => { audioRestarts++; return Task.CompletedTask; },
StartPush = _ => Task.FromResult(++pushes > 1), // first start fails, second works
CheckBroadcast = _ => Task.FromResult(BroadcastLiveness.Live),
Report = reported.Add,
Log = _ => { },
});
var result = await recovery.RunAsync();
Assert.Equal(RecoveryOutcome.Recovered, result.Outcome);
Assert.Equal(2, result.Attempts);
Assert.Equal(2, pushes);
Assert.Equal(2, audioRestarts); // the pipe is unusable after the old client died
Assert.Contains(reported, text => text.Contains("reconnecting (attempt 1"));
Assert.Contains(reported, text => text.StartsWith("Stream recovered"));
}
[Fact]
public async Task ABroadcastThatAlreadyEnded_IsReportedAsComplete_NotAsRecovered()
{
var reported = new List();
var recovery = new StreamRecovery(NoWaits(), new StreamRecoverySteps
{
RestartAudio = _ => Task.CompletedTask,
StartPush = _ => Task.FromResult(true),
CheckBroadcast = _ => Task.FromResult(BroadcastLiveness.Complete),
Report = reported.Add,
Log = _ => { },
});
var result = await recovery.RunAsync();
Assert.Equal(RecoveryOutcome.BroadcastComplete, result.Outcome);
Assert.Contains(reported, text => text.Contains("cannot be resumed"));
Assert.DoesNotContain(reported, text => text.StartsWith("Stream recovered"));
}
[Fact]
public async Task UnknownLiveness_CountsAsRecovered()
{
// If the API hiccups we cannot prove the broadcast is gone; treating that as failure would
// end sessions that are actually fine.
var recovery = new StreamRecovery(NoWaits(), new StreamRecoverySteps
{
RestartAudio = _ => Task.CompletedTask,
StartPush = _ => Task.FromResult(true),
CheckBroadcast = _ => Task.FromResult(BroadcastLiveness.Unknown),
Report = _ => { },
Log = _ => { },
});
Assert.Equal(RecoveryOutcome.Recovered, (await recovery.RunAsync()).Outcome);
}
[Fact]
public async Task EveryAttemptFailing_GivesUpWithTheCount()
{
var reported = new List();
var recovery = new StreamRecovery(NoWaits(3), new StreamRecoverySteps
{
RestartAudio = _ => Task.CompletedTask,
StartPush = _ => Task.FromResult(false),
CheckBroadcast = _ => Task.FromResult(BroadcastLiveness.Live),
Report = reported.Add,
Log = _ => { },
});
var result = await recovery.RunAsync();
Assert.Equal(RecoveryOutcome.GaveUp, result.Outcome);
Assert.Equal(3, result.Attempts);
Assert.Contains(reported, text => text.Contains("off air"));
}
[Fact]
public async Task CancellingMidRecovery_StopsWithoutTouchingThePush()
{
using var cts = new CancellationTokenSource();
var pushes = 0;
var recovery = new StreamRecovery(StreamRetryPolicy.Default, new StreamRecoverySteps
{
RestartAudio = _ => Task.CompletedTask,
StartPush = _ => { pushes++; return Task.FromResult(false); },
CheckBroadcast = _ => Task.FromResult(BroadcastLiveness.Live),
Report = _ => { },
Log = _ => { },
Delay = (_, ct) => { cts.Cancel(); ct.ThrowIfCancellationRequested(); return Task.CompletedTask; },
});
var result = await recovery.RunAsync(cts.Token);
Assert.Equal(RecoveryOutcome.Cancelled, result.Outcome);
Assert.True(pushes <= 1, $"cancelling must stop the loop, but the push was started {pushes} times");
}
[Fact]
public async Task AAudioRestartFault_IsAnAttemptFailure_NotACrash()
{
var attempts = 0;
var recovery = new StreamRecovery(NoWaits(2), new StreamRecoverySteps
{
RestartAudio = _ =>
{
attempts++;
throw new InvalidOperationException("pipe busy");
},
StartPush = _ => Task.FromResult(true),
CheckBroadcast = _ => Task.FromResult(BroadcastLiveness.Live),
Report = _ => { },
Log = _ => { },
});
var result = await recovery.RunAsync();
Assert.Equal(RecoveryOutcome.GaveUp, result.Outcome);
Assert.Equal(2, attempts);
}
}