using System.Diagnostics; using System.Threading.Channels; using Xunit; using ytLive.Models; using ytLive.Services; using ytLive.Services.Encoder; namespace ytLive.Tests; /// /// The FFmpeg subprocess encoder (TASK 4 ship step 3). The integration test drives /// the full lifecycle against fakes: encoder probe → subprocess start with the right /// args → raw frames into stdin → stderr progress parsed into health → graceful stop. /// The units pin down the pure pieces (args, progress parser, encoder picker) and /// the failure edges. No real ffmpeg binary, no network. /// public class FfmpegEncoderTests { private sealed class QueuedReader : TextReader { private readonly Channel _channel = Channel.CreateUnbounded(); public void Enqueue(string s) => _channel.Writer.TryWrite(s); public void Complete() => _channel.Writer.TryComplete(); public override string? ReadLine() => ReadLineAsync().GetAwaiter().GetResult(); public override async Task ReadLineAsync() { try { return await _channel.Reader.ReadAsync().AsTask().ConfigureAwait(false); } catch (ChannelClosedException) { return null; // stream EOF — the process exited and the pipe closed } } } private sealed class FakeEncoderProcess : IEncoderProcess { public ProcessStartInfo? StartInfo { get; private set; } public Stream Stdin { get; } public Stream StandardInput => Stdin; public TextReader StandardOutput { get; } public bool HasExited { get; private set; } public int ExitCode { get; private set; } public bool Killed { get; private set; } public bool Started { get; private set; } private readonly TaskCompletionSource _exit = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly QueuedReader _error = new(); public FakeEncoderProcess(string probeOutput = "", Stream? stdin = null) { StandardOutput = new StringReader(probeOutput); Stdin = stdin ?? new MemoryStream(); } public TextReader StandardError => _error; public void EnqueueStderr(string line) => _error.Enqueue(line); public void Start(ProcessStartInfo startInfo) { StartInfo = startInfo; Started = true; } public void SignalExit(int code = 0) { ExitCode = code; HasExited = true; _error.Complete(); _exit.TrySetResult(); } public void Kill() { Killed = true; _error.Complete(); _exit.TrySetResult(); } public Task WaitForExitAsync(CancellationToken cancellationToken = default) => _exit.Task; public void Dispose() { _error.Complete(); _exit.TrySetResult(); } } private sealed class FakeEncoderProcessFactory { private readonly Queue _processes = new(); public void Return(FakeEncoderProcess p) => _processes.Enqueue(p); public FakeEncoderProcess Create() => _processes.Dequeue(); } private sealed class StubLocator : IFfmpegLocator { public Task LocateAsync(CancellationToken cancellationToken = default) => Task.FromResult(@"C:\tools\ffmpeg.exe"); } private static byte[] BgraFrame(int w, int h, byte b, byte g, byte r) { var bytes = new byte[w * h * 4]; for (var i = 0; i < bytes.Length; i += 4) { bytes[i] = b; bytes[i + 1] = g; bytes[i + 2] = r; bytes[i + 3] = 255; } return bytes; } [Fact] public async Task Start_ProbesEncoder_FeedsFrames_ParsesHealth_StopsGracefully() { const string encoders = " V..... h264_nvenc NVIDIA NVENC H.264 encoder (codec h264)\n" + " V..... libopenh264 OpenH264 H.264 (codec h264)\n"; var probe = new FakeEncoderProcess(encoders); var encoderProc = new FakeEncoderProcess(); var factory = new FakeEncoderProcessFactory(); factory.Return(probe); factory.Return(encoderProc); using var encoder = new FfmpegEncoder(new StubLocator(), factory.Create); var health = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); encoder.HealthUpdated += (_, h) => health.TrySetResult(h); var options = new EncoderOptions { RtmpUrl = "rtmp://a.rtmp.youtube.com/live2/abc-xyz", Width = 1920, Height = 1080, Fps = 60, BitrateKbps = 8000, }; await encoder.StartAsync(options); Assert.True(probe.Started, "the -encoders probe must run"); Assert.True(encoderProc.Started, "the encoder subprocess must start"); Assert.Equal(@"C:\tools\ffmpeg.exe", encoderProc.StartInfo!.FileName); var args = encoderProc.StartInfo.ArgumentList.ToArray(); var c = Array.IndexOf(args, "-c:v"); Assert.True( args[c + 1] == "h264_nvenc", "hardware NVENC must be preferred over the listed openh264"); await encoder.SubmitFrameAsync(new VideoFrame(1920, 1080, BgraFrame(1920, 1080, 255, 0, 0))); await encoder.SubmitFrameAsync(new VideoFrame(1920, 1080, BgraFrame(1920, 1080, 0, 255, 0))); // The drain task writes async — poll for both frames instead of racing it. const long twoFrames = 2 * 1920L * 1080 * 4; var deadline = DateTime.UtcNow.AddSeconds(5); while (encoderProc.Stdin.Length < twoFrames && DateTime.UtcNow < deadline) await Task.Delay(10); Assert.Equal(twoFrames, encoderProc.Stdin.Length); encoderProc.EnqueueStderr( "frame= 120 fps= 59.9 q=28.0 size= 1024KiB time=00:00:02.00 bitrate= 4000.1kbits/s speed=1.00x"); var h = await health.Task.WaitAsync(TimeSpan.FromSeconds(5)); Assert.Equal(4000.1, h.CurrentBitrate, 1); Assert.Equal(59.9, h.FPS, 1); Assert.Equal(TimeSpan.FromSeconds(2), h.StreamDuration); encoderProc.SignalExit(); await encoder.StopAsync(); Assert.False(encoder.IsRunning); Assert.False(encoderProc.Killed, "graceful stop must not kill the process"); Assert.Equal(StreamStatus.Offline, h.Status); } [Fact] public async Task Start_WithoutRtmpUrl_Throws() { using var encoder = new FfmpegEncoder(new StubLocator()); await Assert.ThrowsAsync(() => encoder.StartAsync(new EncoderOptions())); } [Fact] public void Args_RecordOnly_SkipsRtmp_WritesMp4() { var args = FfmpegArgs.Build( new EncoderOptions { StreamEnabled = false, RecordEnabled = true, RecordPath = @"C:\videos\ty-20260829-1405-0000.mp4", Width = 1920, Height = 1080, Fps = 60, BitrateKbps = 8000, }, "libopenh264").ToArray(); Assert.DoesNotContain("flv", args); Assert.DoesNotContain("rtmp", args); var idx = Array.IndexOf(args, "mp4"); Assert.True(idx > -1, "record-only must emit a -f mp4 output"); Assert.Equal("-f", args[idx - 1]); Assert.Equal(@"C:\videos\ty-20260829-1405-0000.mp4", args[idx + 1]); } [Fact] public void Args_DualOutput_HasTwoMapBlocks_FlvAndMp4() { var args = FfmpegArgs.Build( new EncoderOptions { RtmpUrl = "rtmp://a.rtmp.youtube.com/live2/key", StreamEnabled = true, RecordEnabled = true, RecordPath = @"C:\videos\ty-20260829-1405-0000.mp4", Width = 1280, Height = 720, Fps = 30, BitrateKbps = 4000, }, "h264_nvenc").ToArray(); // two video maps (one per output block) Assert.Equal(2, args.Count(a => a == "0:v")); Assert.Equal(2, args.Count(a => a == "-c:v")); Assert.Contains("flv", args); Assert.Contains("mp4", args); Assert.Contains("rtmp://a.rtmp.youtube.com/live2/key", args); Assert.Contains(@"C:\videos\ty-20260829-1405-0000.mp4", args); } [Fact] public async Task SubmitFrame_WhenNotRunning_Throws() { using var encoder = new FfmpegEncoder(new StubLocator()); await Assert.ThrowsAsync( () => encoder.SubmitFrameAsync(new VideoFrame(2, 2, new byte[16]))); } [Fact] public async Task Stop_WithoutStart_IsNoop() { using var encoder = new FfmpegEncoder(new StubLocator()); await encoder.StopAsync(); Assert.False(encoder.IsRunning); } [Fact] public async Task ProcessDeath_RaisesFailed() { var encoderProc = new FakeEncoderProcess(); var factory = new FakeEncoderProcessFactory(); factory.Return(new FakeEncoderProcess()); factory.Return(encoderProc); using var encoder = new FfmpegEncoder(new StubLocator(), factory.Create); var failed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); encoder.ProcessFailed += (_, msg) => failed.TrySetResult(msg); await encoder.StartAsync(new EncoderOptions { RtmpUrl = "rtmp://x/y" }); encoderProc.SignalExit(1); var msg = await failed.Task.WaitAsync(TimeSpan.FromSeconds(5)); Assert.Contains("1", msg); } [Fact] public void Args_Gop_IsFpsTimesFour() { var options = new EncoderOptions { Fps = 60 }; Assert.Equal(240, options.GopSize); var args = FfmpegArgs.Build(options, "libopenh264").ToArray(); var g = Array.IndexOf(args, "-g"); Assert.Equal("240", args[g + 1]); var keyint = Array.IndexOf(args, "-keyint_min"); Assert.Equal("240", args[keyint + 1]); } [Fact] public void Args_IncludeInputOutputAndCompliance() { var args = FfmpegArgs.Build( new EncoderOptions { RtmpUrl = "rtmp://a.rtmp.youtube.com/live2/key", Width = 1920, Height = 1080, Fps = 60, BitrateKbps = 8000, }, "libopenh264").ToArray(); Assert.Contains("-f", args); Assert.Contains("rawvideo", args); Assert.Contains("pipe:0", args); Assert.Contains("1920x1080", args); Assert.Contains(@"\\.\pipe\ytllive_audio", args); Assert.Contains("f32le", args); Assert.Contains("-map", args); Assert.Contains("0:v", args); Assert.Contains("1:a", args); Assert.Contains("-sc_threshold", args); Assert.Contains("-bf", args); Assert.Contains("yuv420p", args); Assert.Contains("aac", args); Assert.Contains("flv", args); Assert.Contains("rtmp://a.rtmp.youtube.com/live2/key", args); } [Fact] public void ProgressParser_ParsesRealStatsLine() { var p = FfmpegProgressParser.TryParse( "frame= 123 fps= 59.9 q=28.0 size= 1024KiB time=00:00:02.04 bitrate= 4000.1kbits/s speed=1.00x"); Assert.NotNull(p); Assert.Equal(123, p.Value.Frame); Assert.Equal(59.9, p.Value.Fps, 1); Assert.Equal(4000.1, p.Value.BitrateKbps, 1); Assert.Equal(2.04, p.Value.Duration.TotalSeconds, 2); Assert.Equal(1024 * 1024, p.Value.SizeBytes); } [Fact] public void ProgressParser_IgnoresBannerAndErrors() { Assert.Null(FfmpegProgressParser.TryParse("ffmpeg version 6.1 Copyright (c) 2000-2026 the FFmpeg developers")); Assert.Null(FfmpegProgressParser.TryParse("Error while opening encoder for output stream #0:0")); Assert.Null(FfmpegProgressParser.TryParse("")); } [Fact] public void EncoderPicker_PrefersHardware_NeverLibx264() { var listing = " V..... libx264 libx264 H.264 / AVC (codec h264)\n" + " V..... libopenh264 OpenH264 H.264 (codec h264)\n" + " V..... h264_nvenc NVIDIA NVENC H.264 encoder (codec h264)\n"; Assert.Equal("h264_nvenc", FfmpegEncoderPicker.Pick(listing)); var onlyGpl = " V..... libx264 libx264 H.264 / AVC (codec h264)\n"; Assert.Equal("libopenh264", FfmpegEncoderPicker.Pick(onlyGpl)); var software = " V..... libopenh264 OpenH264 H.264 (codec h264)\n"; Assert.Equal("libopenh264", FfmpegEncoderPicker.Pick(software)); Assert.Equal("libopenh264", FfmpegEncoderPicker.Pick("no encoders at all")); } /// A stdin slow enough that the encoder can't keep up — the pump's old /// blocking-write failure mode made demonstrable. private sealed class SlowSink : Stream { public long TotalBytes; public override bool CanRead => false; public override bool CanSeek => false; public override bool CanWrite => true; public override long Length => throw new NotSupportedException(); public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } public override void Flush() { } public override Task FlushAsync(CancellationToken ct) => Task.CompletedTask; public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); public override void SetLength(long value) => throw new NotSupportedException(); public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken ct) { await Task.Delay(40, ct).ConfigureAwait(false); Interlocked.Add(ref TotalBytes, count); } } /// The ONE integration test for slice 10 (the bounded-queue reshape): a /// lagging ffmpeg must NOT stall the frame producer. With 300 fast submits against /// a ~25fps-max sink, every queue slot fills and the NEWEST frames are dropped and /// counted; SubmitFrameAsync must return instantly (it never touched a pipe wait — /// the old code blocked in WriteAsync and the pump froze); StopAsync still flushes /// every accepted frame; and the sink ends with exactly (submitted − dropped) bytes /// — the drop counter and the recording agree (the burned-in-frame-counter premise). [Fact] public async Task Backpressure_QueueOverflow_DropsFrames_AndNeverBlocks() { var sink = new SlowSink(); var encoderProc = new FakeEncoderProcess(stdin: sink); var factory = new FakeEncoderProcessFactory(); factory.Return(new FakeEncoderProcess()); factory.Return(encoderProc); using var encoder = new FfmpegEncoder(new StubLocator(), factory.Create); await encoder.StartAsync(new EncoderOptions { RtmpUrl = "rtmp://x/y", Width = 64, Height = 48, Fps = 60, }); var frame = new VideoFrame(64, 48, BgraFrame(64, 48, 255, 0, 0)); const int submits = 300; var sw = System.Diagnostics.Stopwatch.StartNew(); for (var i = 0; i < submits; i++) await encoder.SubmitFrameAsync(frame); sw.Stop(); Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2), $"SubmitFrameAsync blocked on a saturated queue: {sw.Elapsed}"); var dropped = encoder.DroppedFrames; Assert.True(dropped > 0, "the bounded queue must overflow-drop when the encoder lags"); Assert.True(dropped < submits, $"the whole lot must not vanish ({dropped}/{submits})"); encoderProc.SignalExit(); await encoder.StopAsync(); Assert.Equal((submits - dropped) * 64 * 48 * 4L, sink.TotalBytes); } }