jellyfin-plugin-enigma2/Tests/PatientProbeTests.cs

using System;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Plugin.Enigma2;

public static class PatientProbeTests
{
    public static async Task Run(Action<bool,string> check)
    {
        using var upstream = new RelayHandler();
        using var relay = new LiveStreamRelay(new HttpClient(upstream),8096,TimeSpan.FromMilliseconds(200));
        var url=relay.Register("http://receiver/patient");
        using var local = new HttpClient(new LocalRelayHandler(relay));
        var attempts=0;var snapshots=new System.Collections.Generic.List<string>();
        var limits=new LiveStreamProbe.SamplingLimits(TimeSpan.FromMilliseconds(20),TimeSpan.FromMilliseconds(80),188*8,188*128);
        const string good="""{"streams":[{"index":0,"codec_type":"video","codec_name":"h264","width":1920,"height":1080}]}""";
        const string bad="""{"streams":[{"index":0,"codec_type":"video","codec_name":"h264","width":0,"height":0}]}""";
        var probe=new LiveStreamProbe(local,(path,sid,extended,token)=>
        {snapshots.Add(path);return Task.FromResult(++attempts>=3?good:bad);},limits,TimeSpan.FromSeconds(3),true,TimeSpan.FromMilliseconds(40));
        var result=await Task.WhenAll(Enumerable.Range(0,4).Select(_=>probe.Get(url,CancellationToken.None)));
        check(attempts==3 && result.All(x=>x[0].Width==1920),"patient startup retries fresh bounded sample windows shared by all four viewers");
        check(upstream.Calls==1,"multiple metadata windows keep exactly one physical receiver connection");
        check(snapshots.All(path=>!File.Exists(path)&&!File.Exists(path.Replace(".sample.ts",""))),"patient success cleans every sample and capture window");
        await Task.Delay(350);
        check(System.Text.Json.JsonSerializer.Serialize(relay.Status()).Contains("\"ActiveSources\":0"),"patient startup releases receiver after final probe consumer");

        var started=new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var stopped=new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var cancellable=new LiveStreamProbe(async (source,token)=>
        {started.SetResult();try{await Task.Delay(10000,token);return good;}finally{stopped.SetResult();}},TimeSpan.FromSeconds(3));
        using var first=new CancellationTokenSource();using var second=new CancellationTokenSource();
        var a=cancellable.Get("cancel",first.Token);var b=cancellable.Get("cancel",second.Token);await started.Task;
        first.Cancel();try{await a;}catch(OperationCanceledException){}
        check(!stopped.Task.IsCompleted,"one viewer cancelling does not stop another viewer's patient startup");
        second.Cancel();try{await b;}catch(OperationCanceledException){}
        await stopped.Task.WaitAsync(TimeSpan.FromSeconds(1));
        check(stopped.Task.IsCompleted,"last viewer cancellation stops producer immediately instead of waiting 90 seconds");
        var bounded=new LiveStreamProbe(local,(path,sid,extended,token)=>Task.FromResult(bad),limits,TimeSpan.FromMilliseconds(250),true,TimeSpan.FromMilliseconds(20));
        try { await bounded.Get(url,CancellationToken.None); throw new Exception("Expected bounded startup failure"); }
        catch(InvalidOperationException error)
        { check(error.ToString().Contains("video_parameters_unavailable"),"patient deadline preserves the last input diagnosis instead of waiting forever"); }
    }
    private sealed class LocalRelayHandler(LiveStreamRelay relay) : HttpMessageHandler
    {
        protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request,CancellationToken token)
        {
            var sub=await relay.Subscribe(request.RequestUri.Segments.Last(),token);
            return new(HttpStatusCode.OK){Content=new StreamContent(new SubscriptionStream(sub))};
        }
    }
    private sealed class SubscriptionStream(LiveStreamRelay.Subscription subscription) : Stream
    {
        private byte[] chunk;private int at;private bool disposed;
        public override async ValueTask<int> ReadAsync(Memory<byte> buffer,CancellationToken token=default)
        {
            if(chunk==null||at==chunk.Length){chunk=await subscription.Reader.ReadAsync(token);at=0;}
            var n=Math.Min(buffer.Length,chunk.Length-at);chunk.AsMemory(at,n).CopyTo(buffer);at+=n;return n;
        }
        protected override void Dispose(bool disposing){if(disposing&&!disposed){disposed=true;subscription.DisposeAsync().AsTask().GetAwaiter().GetResult();}base.Dispose(disposing);}
        public override bool CanRead=>true;public override bool CanSeek=>false;public override bool CanWrite=>false;
        public override long Length=>throw new NotSupportedException();public override long Position{get=>0;set=>throw new NotSupportedException();}
        public override int Read(byte[] b,int o,int n)=>ReadAsync(b.AsMemory(o,n)).AsTask().GetAwaiter().GetResult();
        public override void Flush(){}public override long Seek(long o,SeekOrigin s)=>throw new NotSupportedException();
        public override void SetLength(long l)=>throw new NotSupportedException();public override void Write(byte[] b,int o,int n)=>throw new NotSupportedException();
    }
}