databasus--databasus/agent/verification/internal/features/api/idle_timeout_reader_test.go

package api

import (
	"context"
	"fmt"
	"io"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func Test_ReadThroughIdleTimeoutReader_WhenBytesFlowContinuously_DoesNotCancelContext(t *testing.T) {
	ctx, cancel := context.WithCancelCause(t.Context())
	defer cancel(nil)

	pr, pw := io.Pipe()

	idleReader := NewIdleTimeoutReader(pr, 200*time.Millisecond, cancel)
	defer idleReader.Stop()

	go func() {
		for range 5 {
			_, _ = pw.Write([]byte("data"))
			time.Sleep(50 * time.Millisecond)
		}

		_ = pw.Close()
	}()

	data, err := io.ReadAll(idleReader)

	require.NoError(t, err)
	assert.Equal(t, "datadatadatadatadata", string(data))
	assert.NoError(t, ctx.Err(), "context should not be cancelled when bytes flow continuously")
}

func Test_ReadThroughIdleTimeoutReader_WhenNoBytesTransmitted_CancelsContext(t *testing.T) {
	ctx, cancel := context.WithCancelCause(t.Context())
	defer cancel(nil)

	pr, _ := io.Pipe()

	idleReader := NewIdleTimeoutReader(pr, 100*time.Millisecond, cancel)
	defer idleReader.Stop()

	time.Sleep(200 * time.Millisecond)

	assert.Error(t, ctx.Err(), "context should be cancelled when no bytes are transmitted")
	assert.Contains(t, context.Cause(ctx).Error(), "upload idle timeout")
}

func Test_ReadThroughIdleTimeoutReader_WhenBytesStopMidStream_CancelsContext(t *testing.T) {
	ctx, cancel := context.WithCancelCause(t.Context())
	defer cancel(nil)

	pr, pw := io.Pipe()

	idleReader := NewIdleTimeoutReader(pr, 100*time.Millisecond, cancel)
	defer idleReader.Stop()

	go func() {
		_, _ = pw.Write([]byte("initial"))
		// Stop writing — simulate stalled source
	}()

	buf := make([]byte, 1024)
	n, _ := idleReader.Read(buf)
	assert.Equal(t, "initial", string(buf[:n]))

	time.Sleep(200 * time.Millisecond)

	assert.Error(t, ctx.Err(), "context should be cancelled when bytes stop mid-stream")
	assert.Contains(t, context.Cause(ctx).Error(), "upload idle timeout")
}

func Test_StopIdleTimeoutReader_WhenCalledBeforeTimeout_DoesNotCancelContext(t *testing.T) {
	ctx, cancel := context.WithCancelCause(t.Context())
	defer cancel(nil)

	pr, _ := io.Pipe()

	idleReader := NewIdleTimeoutReader(pr, 100*time.Millisecond, cancel)
	idleReader.Stop()

	time.Sleep(200 * time.Millisecond)

	assert.NoError(t, ctx.Err(), "context should not be cancelled when reader is stopped before timeout")
}

func Test_ReadThroughIdleTimeoutReader_WhenReaderReturnsError_PropagatesError(t *testing.T) {
	ctx, cancel := context.WithCancelCause(t.Context())
	defer cancel(nil)

	pr, pw := io.Pipe()

	idleReader := NewIdleTimeoutReader(pr, 5*time.Second, cancel)
	defer idleReader.Stop()

	expectedErr := fmt.Errorf("test read error")
	_ = pw.CloseWithError(expectedErr)

	buf := make([]byte, 1024)
	_, err := idleReader.Read(buf)

	assert.ErrorIs(t, err, expectedErr)

	// Timer should be stopped after error — context should not be cancelled
	time.Sleep(100 * time.Millisecond)
	assert.NoError(t, ctx.Err(), "context should not be cancelled after reader error stops the timer")
}