deepdrft/DeepDrftTests/ProgressStreamContentTests.cs

using System.Net.Http;
using DeepDrftManager.Services;

namespace DeepDrftTests;

/// <summary>
/// Unit tests for <see cref="ProgressStreamContent"/> — the single mechanism feeding both the CMS
/// upload progress meter and the idle/heartbeat timeout. Two concerns are anchored here:
/// (1) progress reporting is monotonic and sums to the total content length, and
/// (2) the idle deadline pattern the content drives (reset CancelAfter on each tick) cancels a
/// stalled write yet never fires while writes are progressing.
/// </summary>
[TestFixture]
public class ProgressStreamContentTests
{
    // --- Progress reporting ---

    [Test]
    public async Task ReportsMonotonicallyIncreasingBytes_SummingToContentLength()
    {
        var payload = new byte[256 * 1024 + 7]; // non-chunk-aligned so the final partial read is exercised
        Random.Shared.NextBytes(payload);
        var reports = new List<long>();

        var content = new ProgressStreamContent(
            new MemoryStream(payload), payload.Length, written => reports.Add(written));

        using var sink = new MemoryStream();
        await content.CopyToAsync(sink);

        Assert.That(reports, Is.Not.Empty, "Expected at least one progress tick.");
        Assert.That(reports, Is.Ordered.Ascending, "Progress must be monotonically increasing.");
        Assert.That(reports[^1], Is.EqualTo(payload.Length), "Final tick must equal total content length.");
        Assert.That(sink.ToArray(), Is.EqualTo(payload), "Serialized bytes must match the source payload.");
    }

    [Test]
    public void TryComputeLength_ReturnsProvidedContentLength()
    {
        const long declared = 987_654;
        var content = new ProgressStreamContent(new MemoryStream(), declared, _ => { });

        Assert.That(content.Headers.ContentLength, Is.EqualTo(declared));
    }

    // --- Idle/heartbeat cancellation ---
    // The content does not own a timer; it owns the progress signal. The upload service wires that
    // signal to CancellationTokenSource.CancelAfter(idle). These tests exercise that exact contract by
    // driving the content with a stream whose read cadence we control.

    [Test]
    public async Task IdleTimeout_DoesNotFire_WhileWritesAreProgressing()
    {
        var idle = TimeSpan.FromMilliseconds(200);
        using var idleCts = new CancellationTokenSource();
        idleCts.CancelAfter(idle);

        // Five chunks, each arriving well within the idle window: progress keeps resetting the deadline.
        var source = new PacedStream(chunkCount: 5, chunkSize: 4096, delayPerChunk: TimeSpan.FromMilliseconds(50));
        var content = new ProgressStreamContent(source, source.TotalLength, _ => idleCts.CancelAfter(idle));

        using var sink = new MemoryStream();
        await content.CopyToAsync(sink);

        Assert.That(idleCts.IsCancellationRequested, Is.False,
            "A steadily progressing upload must not trip the idle heartbeat.");
        Assert.That(sink.Length, Is.EqualTo(source.TotalLength));
    }

    [Test]
    public void IdleTimeout_Fires_WhenAStallExceedsTheIdleWindow()
    {
        var idle = TimeSpan.FromMilliseconds(150);
        using var idleCts = new CancellationTokenSource();
        idleCts.CancelAfter(idle);

        // One quick chunk, then a stall longer than the idle window before the next read returns.
        var source = new PacedStream(
            chunkCount: 2, chunkSize: 4096,
            delayPerChunk: TimeSpan.FromMilliseconds(10),
            stallBeforeChunkIndex: 1, stallDuration: TimeSpan.FromMilliseconds(500));
        var content = new ProgressStreamContent(source, source.TotalLength, _ => idleCts.CancelAfter(idle));

        using var sink = new MemoryStream();

        Assert.That(
            async () => await content.CopyToAsync(sink, idleCts.Token),
            Throws.InstanceOf<OperationCanceledException>(),
            "A stall exceeding the idle window must cancel the in-flight copy.");
        Assert.That(idleCts.IsCancellationRequested, Is.True);
    }

    /// <summary>
    /// A read-only stream that yields a fixed number of equal chunks, pausing between reads to emulate
    /// network pacing. Optionally inserts a longer stall before a given chunk to emulate a stalled link.
    /// </summary>
    private sealed class PacedStream : Stream
    {
        private readonly int _chunkCount;
        private readonly int _chunkSize;
        private readonly TimeSpan _delayPerChunk;
        private readonly int _stallBeforeChunkIndex;
        private readonly TimeSpan _stallDuration;
        private int _chunksRead;

        public PacedStream(int chunkCount, int chunkSize, TimeSpan delayPerChunk,
            int stallBeforeChunkIndex = -1, TimeSpan stallDuration = default)
        {
            _chunkCount = chunkCount;
            _chunkSize = chunkSize;
            _delayPerChunk = delayPerChunk;
            _stallBeforeChunkIndex = stallBeforeChunkIndex;
            _stallDuration = stallDuration;
        }

        public long TotalLength => (long)_chunkCount * _chunkSize;

        public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
        {
            if (_chunksRead >= _chunkCount) return 0;

            if (_chunksRead == _stallBeforeChunkIndex)
            {
                await Task.Delay(_stallDuration, cancellationToken);
            }
            else
            {
                await Task.Delay(_delayPerChunk, cancellationToken);
            }

            var count = Math.Min(_chunkSize, buffer.Length);
            buffer.Span[..count].Clear();
            _chunksRead++;
            return count;
        }

        public override int Read(byte[] buffer, int offset, int count)
            => throw new NotSupportedException("Async-only paced stream.");

        public override bool CanRead => true;
        public override bool CanSeek => false;
        public override bool CanWrite => false;
        public override long Length => TotalLength;
        public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
        public override void Flush() { }
        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();
    }
}