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 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) } }