go-mxl-player/internal/playback/video_worker_test.go

package playback

import (
	"context"
	"errors"
	"testing"
	"time"
)

type videoOpenResult struct {
	reader VideoReader
	err    error
}

type scriptedVideoFactory struct {
	results []videoOpenResult
	calls   int
}

func (f *scriptedVideoFactory) OpenVideo(
	context.Context,
	FeedConfig,
) (VideoReader, error) {
	if f.calls >= len(f.results) {
		return nil, errors.New("unexpected video open attempt")
	}
	result := f.results[f.calls]
	f.calls++
	return result.reader, result.err
}

func activeVideoConfig() FeedConfig {
	return FeedConfig{
		Domain: "/dev/shm/mxl",
		UUID:   "video-uuid",
		Active: true,
	}
}

func newTestVideoWorker(
	t *testing.T,
	factory VideoReaderFactory,
	sink VideoSink,
	maxAttempts int,
	shouldRetry func(error) bool,
	observer StatusObserver,
) *VideoWorker {
	t.Helper()

	worker, err := NewVideoWorker(
		factory,
		sink,
		testRetryPolicy(maxAttempts),
		shouldRetry,
		observer,
	)
	if err != nil {
		t.Fatalf("NewVideoWorker() error = %v", err)
	}
	worker.wait = func(context.Context, time.Duration) error { return nil }
	return worker
}

func TestNewVideoWorkerValidatesDependencies(t *testing.T) {
	factory := &scriptedVideoFactory{}
	sink := &fakeVideoSink{}
	retry := testRetryPolicy(3)
	decide := func(error) bool { return true }

	tests := []struct {
		name        string
		factory     VideoReaderFactory
		sink        VideoSink
		retry       RetryPolicy
		shouldRetry func(error) bool
		wantErr     error
	}{
		{
			name:        "missing factory",
			sink:        sink,
			retry:       retry,
			shouldRetry: decide,
			wantErr:     ErrVideoFactoryRequired,
		},
		{
			name:        "missing sink",
			factory:     factory,
			retry:       retry,
			shouldRetry: decide,
			wantErr:     ErrVideoSinkRequired,
		},
		{
			name:    "missing retry decider",
			factory: factory,
			sink:    sink,
			retry:   retry,
			wantErr: ErrVideoRetryDeciderRequired,
		},
		{
			name:        "invalid retry policy",
			factory:     factory,
			sink:        sink,
			retry:       RetryPolicy{},
			shouldRetry: decide,
			wantErr:     ErrInvalidRetryDelay,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			worker, err := NewVideoWorker(
				tt.factory,
				tt.sink,
				tt.retry,
				tt.shouldRetry,
				nil,
			)
			if worker != nil {
				t.Fatal("NewVideoWorker() worker is not nil")
			}
			if !errors.Is(err, tt.wantErr) {
				t.Fatalf("NewVideoWorker() error = %v, want %v", err, tt.wantErr)
			}
		})
	}
}

func TestVideoWorkerRejectsInactiveFeed(t *testing.T) {
	factory := &scriptedVideoFactory{}
	var statuses []Status
	worker := newTestVideoWorker(
		t,
		factory,
		&fakeVideoSink{},
		3,
		func(error) bool { return true },
		func(status Status) { statuses = append(statuses, status) },
	)
	config := activeVideoConfig()
	config.Active = false

	err := worker.Run(context.Background(), config)
	if !errors.Is(err, ErrVideoFeedInactive) {
		t.Fatalf("Run() error = %v, want %v", err, ErrVideoFeedInactive)
	}
	if factory.calls != 0 {
		t.Errorf("factory calls = %d, want 0", factory.calls)
	}
	if len(statuses) != 0 {
		t.Errorf("status count = %d, want 0", len(statuses))
	}
}

func TestVideoWorkerStatusesInheritGeneration(t *testing.T) {
	openErr := errors.New("unavailable")
	var statuses []Status
	worker := newTestVideoWorker(
		t,
		&scriptedVideoFactory{results: []videoOpenResult{{err: openErr}}},
		&fakeVideoSink{},
		1,
		func(error) bool { return true },
		func(status Status) { statuses = append(statuses, status) },
	)
	_ = worker.Run(withGeneration(context.Background(), 7), activeVideoConfig())
	if len(statuses) == 0 {
		t.Fatal("no statuses emitted")
	}
	for _, status := range statuses {
		if status.Generation != 7 {
			t.Fatalf("status generation = %d, want 7: %+v", status.Generation, status)
		}
	}
}

func TestVideoWorkerExhaustsOpenRetries(t *testing.T) {
	openErr := errors.New("producer unavailable")
	factory := &scriptedVideoFactory{
		results: []videoOpenResult{
			{err: openErr},
			{err: openErr},
		},
	}
	var statuses []Status
	worker := newTestVideoWorker(
		t,
		factory,
		&fakeVideoSink{},
		2,
		func(error) bool { return true },
		func(status Status) { statuses = append(statuses, status) },
	)

	err := worker.Run(context.Background(), activeVideoConfig())
	if !errors.Is(err, openErr) {
		t.Fatalf("Run() error = %v, want %v", err, openErr)
	}
	if factory.calls != 2 {
		t.Errorf("factory calls = %d, want 2", factory.calls)
	}

	wantStates := []State{
		StateConnecting,
		StateReconnecting,
		StateReconnecting,
		StateFailed,
	}
	if len(statuses) != len(wantStates) {
		t.Fatalf("status count = %d, want %d: %+v", len(statuses), len(wantStates), statuses)
	}
	for i, want := range wantStates {
		if statuses[i].State != want {
			t.Errorf("status %d state = %v, want %v", i, statuses[i].State, want)
		}
		if statuses[i].Unit != UnitVideo {
			t.Errorf("status %d unit = %v, want %v", i, statuses[i].Unit, UnitVideo)
		}
	}
	final := statuses[len(statuses)-1]
	if final.Attempt != 2 || final.FailedAttempts != 2 {
		t.Errorf("final status = %+v, want attempt=2 failedAttempts=2", final)
	}
	if !errors.Is(final.Err, openErr) {
		t.Errorf("final error = %v, want %v", final.Err, openErr)
	}
}

func TestVideoWorkerStablePlaybackResetsRetryCounter(t *testing.T) {
	readErr := errors.New("video disconnected")
	ctx, cancel := context.WithCancel(context.Background())
	first := &fakeVideoReader{
		frames:  []VideoFrame{{Index: 1, Payload: []byte{1}}},
		readErr: readErr,
	}
	second := &fakeVideoReader{
		frames:  []VideoFrame{{Index: 2, Payload: []byte{2}}},
		readErr: readErr,
	}
	third := &fakeVideoReader{
		read: func(ctx context.Context) (VideoFrame, error) {
			cancel()
			return VideoFrame{}, ctx.Err()
		},
	}
	factory := &scriptedVideoFactory{
		results: []videoOpenResult{
			{reader: first},
			{reader: second},
			{reader: third},
		},
	}
	var statuses []Status
	worker := newTestVideoWorker(
		t,
		factory,
		&fakeVideoSink{},
		2,
		func(error) bool { return true },
		func(status Status) { statuses = append(statuses, status) },
	)

	err := worker.Run(ctx, activeVideoConfig())
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("Run() error = %v, want context.Canceled", err)
	}
	if factory.calls != 3 {
		t.Fatalf("factory calls = %d, want 3", factory.calls)
	}

	var playingAttempts []int
	var retryFailures []int
	for _, status := range statuses {
		switch status.State {
		case StatePlaying:
			playingAttempts = append(playingAttempts, status.Attempt)
		case StateReconnecting:
			if status.RetryIn > 0 {
				retryFailures = append(retryFailures, status.FailedAttempts)
			}
		}
	}
	if len(playingAttempts) != 2 || playingAttempts[0] != 1 || playingAttempts[1] != 2 {
		t.Errorf("playing attempts = %v, want [1 2]", playingAttempts)
	}
	if len(retryFailures) != 2 || retryFailures[0] != 1 || retryFailures[1] != 1 {
		t.Errorf("retry failure counts = %v, want [1 1]", retryFailures)
	}
	wantEnding := []State{StateStopping, StateIdle}
	if len(statuses) < 2 {
		t.Fatalf("status count = %d, want at least 2", len(statuses))
	}
	ending := statuses[len(statuses)-2:]
	for i, want := range wantEnding {
		if ending[i].State != want {
			t.Errorf("ending status %d = %v, want %v", i, ending[i].State, want)
		}
	}
}

func TestVideoWorkerDoesNotRetrySinkFailure(t *testing.T) {
	sinkErr := errors.New("renderer failed")
	reader := &fakeVideoReader{
		frames: []VideoFrame{{Index: 1, Payload: []byte{1}}},
	}
	factory := &scriptedVideoFactory{
		results: []videoOpenResult{{reader: reader}},
	}
	deciderCalls := 0
	var statuses []Status
	worker := newTestVideoWorker(
		t,
		factory,
		&fakeVideoSink{err: sinkErr},
		0,
		func(error) bool {
			deciderCalls++
			return true
		},
		func(status Status) { statuses = append(statuses, status) },
	)

	err := worker.Run(context.Background(), activeVideoConfig())
	if !errors.Is(err, sinkErr) {
		t.Fatalf("Run() error = %v, want %v", err, sinkErr)
	}
	if factory.calls != 1 {
		t.Errorf("factory calls = %d, want 1", factory.calls)
	}
	if deciderCalls != 0 {
		t.Errorf("source retry decider calls = %d, want 0", deciderCalls)
	}
	if !reader.closed {
		t.Fatal("reader was not closed")
	}
	if len(statuses) != 2 {
		t.Fatalf("status count = %d, want 2: %+v", len(statuses), statuses)
	}
	if statuses[0].State != StateConnecting || statuses[1].State != StateFailed {
		t.Errorf("status states = [%v %v], want [Connecting Failed]", statuses[0].State, statuses[1].State)
	}
}